add unit test

Author: shijiesheng

cadence/workflow/deterministic_event_loop.py

@@ -1,4 +1,4 @@
-from asyncio import AbstractEventLoop, BaseEventLoop, EventLoop, futures, tasks
+from asyncio import AbstractEventLoop, futures, tasks
 from asyncio import Future
 import logging
 import collections
@@ -108,6 +108,9 @@ def create_task(self, coro, **kwargs):
             # task.exception().__traceback__->BaseEventLoop.create_task->task
             del task
 
+    def create_future(self):
+        return futures.Future(loop=self)
+
     def _run_once(self):
         ntodo = len(self._ready)
         for i in range(ntodo):
@@ -165,7 +168,5 @@ def _run_until_complete_cb(fut: Future):
     if not fut.cancelled():
         exc = fut.exception()
         if isinstance(exc, (SystemExit, KeyboardInterrupt)):
-            # Issue #22429: run_forever() already finished, no need to
-            # stop it.
             return
     fut.get_loop().stop()

tests/cadence/workflow/test_deterministic_event_loop.py

@@ -0,0 +1,72 @@
+import pytest
+import asyncio
+from cadence.workflow.deterministic_event_loop import DeterministicEventLoop
+
+
+async def coro_append(results: list, i: int):
+    results.append(i)
+
+async def coro_await(size: int):
+    results = []
+    for i in range(size):
+        await coro_append(results, i)
+    return results
+
+async def coro_await_future(future: asyncio.Future):
+    return await future
+
+async def coro_await_task(size: int):
+    results = []
+    for i in range(size):
+        asyncio.create_task(coro_append(results, i))
+    return results
+
+class TestDeterministicEventLoop:
+    """Test suite for DeterministicEventLoop using table-driven tests."""
+
+    def setup_method(self):
+        """Setup method called before each test."""
+        self.loop = DeterministicEventLoop()
+
+    def teardown_method(self):
+        """Teardown method called after each test."""
+        if not self.loop.is_closed():
+            self.loop.close()
+        assert self.loop.is_closed() is True
+
+    def test_call_soon(self):
+        """Test _run_once executes single callback."""
+        results = []
+        expected = []
+        for i in range(10000):
+            expected.append(i)
+            self.loop.call_soon(lambda x=i: results.append(x))
+
+        self.loop._run_once()
+
+        assert results == expected
+        assert self.loop.is_running() is False
+
+    def test_run_until_complete(self):
+        size = 10000
+        results = self.loop.run_until_complete(coro_await(size))
+        assert results == list(range(size))
+        assert self.loop.is_running() is False
+        assert self.loop.is_closed() is False
+
+    @pytest.mark.parametrize("result, exception, expected, expected_exception",
+        [(10000, None, 10000, None), (None, ValueError("test"), None, ValueError)])
+    def test_create_future(self, result, exception, expected, expected_exception):
+        future = self.loop.create_future()
+        if expected_exception is not None:
+            with pytest.raises(expected_exception):
+                future.set_exception(exception)
+                self.loop.run_until_complete(coro_await_future(future))
+        else:
+            future.set_result(result)
+            assert self.loop.run_until_complete(coro_await_future(future)) == expected
+
+    def test_create_task(self):
+        size = 10000
+        results = self.loop.run_until_complete(coro_await_task(size))
+        assert results == list(range(size))