Thank you for taking the time to report a bug! Your support is essential for the maintenance of this project. Please fill out the following fields to ease bug hunting and resolving this issue as soon as possible:
Describe the bug
============================= test session starts ==============================
platform linux -- Python 3.10.1, pytest-6.2.5, py-1.11.0, pluggy-1.0.0
rootdir: /build/transitions-0.8.10, configfile: pytest.ini
collected 2463 items
tests/test_add_remove.py ... [ 0%]
tests/test_async.py ..................................F
=================================== FAILURES ===================================
________________________ TestAsync.test_multiple_models ________________________
self = <tests.test_async.TestAsync testMethod=test_multiple_models>
def test_multiple_models(self):
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
m1 = self.machine_cls(states=['A', 'B', 'C'], initial='A', name="m1")
m2 = self.machine_cls(states=['A'], initial='A', name='m2')
m1.add_transition(trigger='go', source='A', dest='B', before=self.cancel_soon)
m1.add_transition(trigger='fix', source='A', dest='C', after=self.cancel_soon)
m1.add_transition(trigger='check', source='C', dest='B', conditions=self.await_false)
m1.add_transition(trigger='reset', source='C', dest='A')
m2.add_transition(trigger='go', source='A', dest=None, conditions=m1.is_C, after=m1.reset)
> loop.run_until_complete(asyncio.gather(m1.go(), # should block before B
self.call_delayed(m1.fix, 0.05), # should cancel task and go to C
self.call_delayed(m1.check, 0.07), # should exit before m1.fix
self.call_delayed(m2.go, 0.1))) # should cancel m1.fix
tests/test_async.py:144:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
/nix/store/b798fp24zf2fdafmyyc4sxfr48ly5yy9-python3-3.10.1/lib/python3.10/asyncio/tasks.py:776: in gather
fut = _ensure_future(arg, loop=loop)
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
coro_or_future = <coroutine object AsyncEvent.trigger at 0x7ffff58bfc30>
def _ensure_future(coro_or_future, *, loop=None):
if futures.isfuture(coro_or_future):
if loop is not None and loop is not futures._get_loop(coro_or_future):
raise ValueError('The future belongs to a different loop than '
'the one specified as the loop argument')
return coro_or_future
if not coroutines.iscoroutine(coro_or_future):
if inspect.isawaitable(coro_or_future):
coro_or_future = _wrap_awaitable(coro_or_future)
else:
raise TypeError('An asyncio.Future, a coroutine or an awaitable '
'is required')
if loop is None:
> loop = events._get_event_loop(stacklevel=4)
E DeprecationWarning: There is no current event loop
/nix/store/b798fp24zf2fdafmyyc4sxfr48ly5yy9-python3-3.10.1/lib/python3.10/asyncio/tasks.py:633: DeprecationWarning
=========================== short test summary info ============================
FAILED tests/test_async.py::TestAsync::test_multiple_models - DeprecationWarn...
!!!!!!!!!!!!!!!!!!!!!!!!!! stopping after 1 failures !!!!!!!!!!!!!!!!!!!!!!!!!!!
========================= 1 failed, 37 passed in 2.41s =========================
Thank you for taking the time to report a bug! Your support is essential for the maintenance of this project. Please fill out the following fields to ease bug hunting and resolving this issue as soon as possible:
Describe the bug
Minimal working example
n/a
Expected behavior Tests to pass.
Additional context https://docs.python.org/3.10/library/asyncio-eventloop.html#asyncio.get_event_loop