pre-

eventkit/event.py

@@ -14,7 +14,6 @@
     Final,
     Iterable,
     List,
-    Optional,
     Tuple,
     Union,
 )

eventkit/ops/create.py

@@ -3,7 +3,7 @@
 import time
 
 from ..event import Event
-from ..util import NO_VALUE, timerange
+from ..util import NO_VALUE, get_event_loop, timerange
 from .op import Op
 
 
@@ -12,13 +12,18 @@ class Wait(Event):
 
     def __init__(self, future, name="wait"):
         Event.__init__(self, name)
-        self._task = asyncio.create_task(future)
-        self._task.add_done_callback(self._on_task_done)
+        if future.done():
+            self._task = None
+            self.set_done()
+        else:
+            # Note: the loop= *is* necessary here.
+            self._task = asyncio.ensure_future(future, loop=get_event_loop())
+            future.add_done_callback(self._on_task_done)
 
     def _on_task_done(self, task):
         try:
             result = task.result()
-        except (asyncio.CancelledError, asyncio.InvalidStateError) as error:
+        except Exception as error:
             result = NO_VALUE
             self.error_event.emit(self, error)
 
@@ -36,7 +41,9 @@ class Aiterate(Event):
 
     def __init__(self, ait):
         Event.__init__(self, ait.__qualname__)
-        self._task = asyncio.create_task(self._looper(ait))
+
+        # Note: the loop= *is* necessary here.
+        self._task = asyncio.ensure_future(self._looper(ait), loop=get_event_loop())
 
     async def _looper(self, ait):
         try:
@@ -48,6 +55,10 @@ async def _looper(self, ait):
         self._task = None
         self.set_done()
 
+    def __del__(self):
+        if self._task:
+            self._task.cancel()
+
 
 class Sequence(Aiterate):
     __slots__ = ()

eventkit/util.py

@@ -18,13 +18,7 @@ def __repr__(self):
 
 def get_event_loop():
     """Get asyncio event loop or create one if it doesn't exist."""
-    try:
-        # https://docs.python.org/3/library/asyncio-eventloop.html#asyncio.get_running_loop
-        loop = asyncio.get_running_loop()
-    except RuntimeError:
-        loop = asyncio.new_event_loop()
-        asyncio.set_event_loop(loop)
-
+    loop = asyncio.get_event_loop_policy().get_event_loop()
     return loop
 
 

pyproject.toml

@@ -33,7 +33,6 @@ optional = true
 ruff = "^0.5.7"
 pre-commit = "^3.5.0"
 pytest = "^8.3.2"
-pytest-asyncio = "^0.24.0"
 pytest-xdist = "^3.6.1"
 mypy = "^1.11.1"
 
@@ -69,8 +68,5 @@ ignore = [
     "E402",
     "F401",
 ]
-
 [tool.pytest.ini_options]
-asyncio_default_fixture_loop_scope = "session"
-asyncio_mode = "auto"
 log_cli = true

tests/aggregate_test.py

@@ -1,50 +1,51 @@
-import asyncio
+import unittest
 
 from eventkit import Event
 
 array = list(range(10))
 
 
-class TestAggregate:
-    async def test_min(self):
-        res = await Event.sequence(array).min().list()
-        assert res == [0] * 10
+class AggregateTest(unittest.TestCase):
+    def test_min(self):
+        event = Event.sequence(array).min()
+        self.assertEqual(event.run(), [0] * 10)
 
-    async def test_max(self):
-        res = await Event.sequence(array).max().list()
-        assert res == array
+    def test_max(self):
+        event = Event.sequence(array).max()
+        self.assertEqual(event.run(), array)
 
-    async def test_sum(self):
-        res = await Event.sequence(array).sum().list()
-        assert res == [0, 1, 3, 6, 10, 15, 21, 28, 36, 45]
+    def test_sum(self):
+        event = Event.sequence(array).sum()
+        self.assertEqual(event.run(), [0, 1, 3, 6, 10, 15, 21, 28, 36, 45])
 
-    async def test_product(self):
-        res = await Event.sequence(array[1:]).product().list()
-        assert res == [1, 2, 6, 24, 120, 720, 5040, 40320, 362880]
+    def test_product(self):
+        event = Event.sequence(array[1:]).product()
+        self.assertEqual(event.run(), [1, 2, 6, 24, 120, 720, 5040, 40320, 362880])
 
-    async def test_any(self):
-        res = await Event.sequence(array).any().list()
-        assert res == [False, True, True, True, True, True, True, True, True, True]
+    def test_any(self):
+        event = Event.sequence(array).any()
+        self.assertEqual(
+            event.run(), [False, True, True, True, True, True, True, True, True, True]
+        )
 
-    async def test_all(self):
+    def test_all(self):
         x = [True] * 10 + [False] * 10
-        res = await Event.sequence(x).all().list()
-        assert res == x
+        event = Event.sequence(x).all()
+        self.assertEqual(event.run(), x)
 
-    async def test_pairwaise(self):
-        res = await Event.sequence(array).pairwise().list()
-        assert res == list(zip(array, array[1:]))
+    def test_pairwaise(self):
+        event = Event.sequence(array).pairwise()
+        self.assertEqual(event.run(), list(zip(array, array[1:])))
 
-    async def test_chunk(self):
-        res = await Event.sequence(array).chunk(3).list()
-        assert res == [[0, 1, 2], [3, 4, 5], [6, 7, 8], [9]]
+    def test_chunk(self):
+        event = Event.sequence(array).chunk(3)
+        self.assertEqual(event.run(), [[0, 1, 2], [3, 4, 5], [6, 7, 8], [9]])
 
-    async def test_chunkwith(self):
+    def test_chunkwith(self):
         timer = Event.timer(0.029, 10)
-        res = await Event.sequence(array, 0.01).chunkwith(timer).list()
-        assert res == [[0, 1, 2], [3, 4, 5], [6, 7, 8], [9]]
-        await asyncio.sleep(0.5)
+        event = Event.sequence(array, 0.01).chunkwith(timer)
+        self.assertEqual(event.run(), [[0, 1, 2], [3, 4, 5], [6, 7, 8], [9]])
 
-    async def test_array(self):
-        res = await Event.sequence(array).array(5).last().list()
-        assert list(res[0]) == array[-5:]
+    def test_array(self):
+        event = Event.sequence(array).array(5).last()
+        self.assertEqual(list(event.run()[0]), array[-5:])

tests/combine_test.py

@@ -7,60 +7,47 @@
 array3 = list(range(200, 210))
 
 
-class TestCombine:
-    async def test_merge(self):
+class CombineTest(unittest.TestCase):
+    def test_merge(self):
         e1 = Event.sequence(array1, interval=0.01)
         e2 = Event.sequence(array2, interval=0.01).delay(0.001)
         event = e1.merge(e2)
-        assert await event.list() == [i for j in zip(array1, array2) for i in j]
+        self.assertEqual(event.run(), [i for j in zip(array1, array2) for i in j])
 
-    async def test_switch(self):
+    def test_switch(self):
         e1 = Event.sequence(array1, interval=0.01)
         e2 = Event.sequence(array2, interval=0.01).delay(0.001)
         e3 = Event.sequence(array3, interval=0.01).delay(0.002)
         event = e1.switch(e2, e3, e2)
-        assert await event.list() == [0, 100] + array3
+        self.assertEqual(event.run(), [0, 100] + array3)
 
-    async def test_concat(self):
+    def test_concat(self):
         e1 = Event.sequence(array1, interval=0.02)
         e2 = Event.sequence(array2, interval=0.02).delay(0.07)
         event = e1.concat(e2)
-        assert await event.list() == [
-            0,
-            1,
-            2,
-            3,
-            100,
-            101,
-            102,
-            103,
-            104,
-            105,
-            106,
-            107,
-            108,
-            109,
-        ]
+        self.assertEqual(
+            event.run(), [0, 1, 2, 3, 100, 101, 102, 103, 104, 105, 106, 107, 108, 109]
+        )
 
-    async def test_chain(self):
+    def test_chain(self):
         e1 = Event.sequence(array1, interval=0.01)
         e2 = Event.sequence(array2, interval=0.01).delay(0.001)
         event = e1.chain(e2, e1)
-        assert await event.list() == array1 + array2 + array1
+        self.assertEqual(event.run(), array1 + array2 + array1)
 
-    async def test_zip(self):
+    def test_zip(self):
         e1 = Event.sequence(array1)
         e2 = Event.sequence(array2).delay(0.001)
         event = e1.zip(e2)
-        assert await event.list() == list(zip(array1, array2))
+        self.assertEqual(event.run(), list(zip(array1, array2)))
 
-    async def test_zip_self(self):
+    def test_zip_self(self):
         e1 = Event.sequence(array1)
         event = e1.zip(e1)
-        assert await event.list(), list(zip(array1, array1))
+        self.assertEqual(event.run(), list(zip(array1, array1)))
 
-    async def test_ziplatest(self):
+    def test_ziplatest(self):
         e1 = Event.sequence([0, 1], interval=0.01)
         e2 = Event.sequence([2, 3], interval=0.01).delay(0.001)
         event = e1.ziplatest(e2)
-        assert await event.list() == [(0, Event.NO_VALUE), (0, 2), (1, 2), (1, 3)]
+        self.assertEqual(event.run(), [(0, Event.NO_VALUE), (0, 2), (1, 2), (1, 3)])

tests/create_test.py

@@ -1,32 +1,31 @@
 import asyncio
+import unittest
 
 from eventkit import Event
+from eventkit.util import get_event_loop
 
 array1 = list(range(10))
 array2 = list(range(100, 110))
 
 
-class TestCreate:
-    """test create"""
+class CreateTest(unittest.TestCase):
+    def test_wait(self):
+        loop = get_event_loop()
+        fut = asyncio.Future(loop=loop)
+        loop.call_later(0.001, fut.set_result, 42)
+        event = Event.wait(fut)
+        self.assertEqual(event.run(), [42])
 
-    async def test_wait(self):
-        async def coro():
-            await asyncio.sleep(0)
-            return 42
-
-        res = await Event.wait(coro())
-        assert res == 42
-
-    async def test_aiterate(self):
+    def test_aiterate(self):
         async def ait():
             await asyncio.sleep(0)
             for i in array1:
                 yield i
 
-        res = await Event.aiterate(ait()).list()
-        assert res == array1
+        event = Event.aiterate(ait())
+        self.assertEqual(event.run(), array1)
 
-    async def test_marble(self):
+    def test_marble(self):
         s = "   a b c   d e f"
-        res = await Event.marble(s, interval=0.001).list()
-        assert res == [c for c in "abcdef"]
+        event = Event.marble(s, interval=0.001)
+        self.assertEqual(event.run(), [c for c in "abcdef"])