Changes

- functools.cache is out
- pytest-async plugin in
- I had to rewrite all tests so they don't use unittest, so now all test are on pytest
- All test pass, except when they are run with pytest-xdist, poetry run pytest -vs -n auto gives a warning. It's a warning, not an error

 > task: <Task pending name='Task-18' coro=<Aiterate._looper() done, defined at ~/Documents/dev/eventkit/eventkit/ops/create.py:42> wait_for=>

- updated Event.repeat, as the parameters where in the wrong position. it should be Ok now.
- update Aiterate and Wait to use tasks. removed __del__ method from Aiterate as it causes warnings. we can put it back and live with warnings on tests, which is not bad.
- right now everything is task based, except timing module which uses loop.call_soon and loop.call_at

eventkit/event.py

@@ -679,7 +679,7 @@ def repeat(
             times: Relative times for individual values, in seconds since
                 start of event. The sequence should match ``values``.
         """
-        return Repeat(interval, value, count, times)
+        return Repeat(value=value, count=count, interval=interval, times=times)
 
     @staticmethod
     def range(
@@ -970,7 +970,7 @@ def map(self, func, timeout=None, ordered=True, task_limit=None) -> "Map":
         ``timeout``, ``ordered`` and ``task_limit`` apply to
         async functions only.
         """
-        return Map(func, timeout, ordered, task_limit, self)
+        return Map(func, timeout, ordered, task_limit, source=self)
 
     def emap(self, constr, joiner: "AddableJoinOp") -> "Emap":
         """

eventkit/ops/create.py

@@ -1,29 +1,25 @@
 import asyncio
 import itertools
 import time
+from typing import Awaitable
 
 from ..event import Event
-from ..util import NO_VALUE, get_event_loop, timerange
+from ..util import NO_VALUE, timerange
 from .op import Op
 
 
 class Wait(Event):
     __slots__ = ("_task",)
 
-    def __init__(self, future, name="wait"):
+    def __init__(self, future: Awaitable, name="wait"):
         Event.__init__(self, name)
-        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)
+        self._task = asyncio.create_task(future)
+        self._task.add_done_callback(self._on_task_done)
 
     def _on_task_done(self, task):
         try:
             result = task.result()
-        except Exception as error:
+        except (asyncio.CancelledError, asyncio.InvalidStateError) as error:
             result = NO_VALUE
             self.error_event.emit(self, error)
 
@@ -41,9 +37,7 @@ class Aiterate(Event):
 
     def __init__(self, ait):
         Event.__init__(self, ait.__qualname__)
-
-        # Note: the loop= *is* necessary here.
-        self._task = asyncio.ensure_future(self._looper(ait), loop=get_event_loop())
+        self._task = asyncio.create_task(self._looper(ait))
 
     async def _looper(self, ait):
         try:
@@ -55,10 +49,6 @@ 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

@@ -1,6 +1,5 @@
 import asyncio
 import datetime as dt
-import functools
 from typing import AsyncIterator, Final
 
 
@@ -17,7 +16,6 @@ def __repr__(self):
 NO_VALUE: Final = _NoValue()
 
 
-@functools.cache
 def get_event_loop():
     """Get asyncio event loop or create one if it doesn't exist."""
     try:

pyproject.toml

@@ -33,6 +33,7 @@ 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"
 
@@ -68,3 +69,7 @@ ignore = [
     "E402",
     "F401",
 ]
+
+[tool.pytest.ini_options]
+asyncio_default_fixture_loop_scope = "session"
+asyncio_mode = "auto"

tests/aggregate_test.py

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

tests/combine_test.py

@@ -7,47 +7,60 @@
 array3 = list(range(200, 210))
 
 
-class CombineTest(unittest.TestCase):
-    def test_merge(self):
+class TestCombine:
+    async 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)
-        self.assertEqual(event.run(), [i for j in zip(array1, array2) for i in j])
+        assert await event.list() == [i for j in zip(array1, array2) for i in j]
 
-    def test_switch(self):
+    async 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)
-        self.assertEqual(event.run(), [0, 100] + array3)
+        assert await event.list() == [0, 100] + array3
 
-    def test_concat(self):
+    async 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)
-        self.assertEqual(
-            event.run(), [0, 1, 2, 3, 100, 101, 102, 103, 104, 105, 106, 107, 108, 109]
-        )
+        assert await event.list() == [
+            0,
+            1,
+            2,
+            3,
+            100,
+            101,
+            102,
+            103,
+            104,
+            105,
+            106,
+            107,
+            108,
+            109,
+        ]
 
-    def test_chain(self):
+    async 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)
-        self.assertEqual(event.run(), array1 + array2 + array1)
+        assert await event.list() == array1 + array2 + array1
 
-    def test_zip(self):
+    async def test_zip(self):
         e1 = Event.sequence(array1)
         e2 = Event.sequence(array2).delay(0.001)
         event = e1.zip(e2)
-        self.assertEqual(event.run(), list(zip(array1, array2)))
+        assert await event.list() == list(zip(array1, array2))
 
-    def test_zip_self(self):
+    async def test_zip_self(self):
         e1 = Event.sequence(array1)
         event = e1.zip(e1)
-        self.assertEqual(event.run(), list(zip(array1, array1)))
+        assert await event.list(), list(zip(array1, array1))
 
-    def test_ziplatest(self):
+    async 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)
-        self.assertEqual(event.run(), [(0, Event.NO_VALUE), (0, 2), (1, 2), (1, 3)])
+        assert await event.list() == [(0, Event.NO_VALUE), (0, 2), (1, 2), (1, 3)]

tests/create_test.py

@@ -1,31 +1,32 @@
 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 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])
+class TestCreate:
+    """test create"""
 
-    def test_aiterate(self):
+    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):
         async def ait():
             await asyncio.sleep(0)
             for i in array1:
                 yield i
 
-        event = Event.aiterate(ait())
-        self.assertEqual(event.run(), array1)
+        res = await Event.aiterate(ait()).list()
+        assert res == array1
 
-    def test_marble(self):
+    async def test_marble(self):
         s = "   a b c   d e f"
-        event = Event.marble(s, interval=0.001)
-        self.assertEqual(event.run(), [c for c in "abcdef"])
+        res = await Event.marble(s, interval=0.001).list()
+        assert res == [c for c in "abcdef"]