test: Rewrite the test suite

conftest.py

@@ -1,49 +1,15 @@
-from typing import Any
-
 import pytest
 from injection import Module
 
-from cq import Bus, CommandBus, EventBus, QueryBus, Router
-from cq._core.routing.dispatchers.bus import SimpleBus
+from cq import Router
 from cq.ext.injection import InjectionAdapter
-from tests.helpers.history import HistoryMiddleware
-
-
-@pytest.fixture(scope="function")
-def router(injection_module: Module) -> Router:
-    return Router(InjectionAdapter(injection_module)).register_defaults()
-
-
-@pytest.fixture(scope="function")
-def bus() -> Bus[Any, Any]:
-    return SimpleBus()
-
 
-@pytest.fixture(scope="function", autouse=True)
-def ensure_test_dependencies(
-    router: Router,
-    history: HistoryMiddleware,
-    injection_module: Module,
-) -> None:
-    injection_module.injectable(
-        lambda: router.new_command_bus().add_middlewares(history),
-        on=CommandBus,
-    )
-    injection_module.injectable(
-        lambda: router.new_event_bus().add_middlewares(history),
-        on=EventBus,
-    )
-    injection_module.injectable(
-        lambda: router.new_query_bus().add_middlewares(history),
-        on=QueryBus,
-    )
 
-
-@pytest.fixture(scope="function")
-def history() -> HistoryMiddleware:
-    return HistoryMiddleware()
-
-
-@pytest.fixture(scope="function")
+@pytest.fixture
 def injection_module() -> Module:
     return Module()
+
+
+@pytest.fixture
+def router(injection_module: Module) -> Router:
+    return Router(InjectionAdapter(injection_module)).register_defaults()

cq/_core/related_events.py

@@ -25,7 +25,7 @@ class AnyIORelatedEvents(RelatedEvents):
     task_group: TaskGroup = field(default_factory=create_task_group)
     history: list[Event] = field(default_factory=list, init=False)
 
-    def __bool__(self) -> bool:  # pragma: no cover
+    def __bool__(self) -> bool:
         return bool(self.history)
 
     async def __aenter__(self) -> Self:

cq/middlewares/retry.py

@@ -21,6 +21,9 @@ def __init__(
         delay: float = 0,
         exceptions: Sequence[type[BaseException]] = (Exception,),
     ) -> None:
+        if retry < 1:
+            raise ValueError(f"`retry` must be at least 1, got {retry}.")
+
         self.__delay = delay
         self.__exceptions = tuple(exceptions)
         self.__retry = retry

tests/core/queuing/queues/test_memory.py

@@ -1,50 +1,114 @@
-from dataclasses import dataclass, field
+import math
+from typing import Any
 
 import pytest
 from anyio import ClosedResourceError
 
-from cq import MemoryQueue
+from cq import MemoryQueue, MiddlewareResult
+
+
+async def _receive_all[T](queue: MemoryQueue[T]) -> list[T]:
+    messages = []
+
+    async for delivery in queue:
+        async with delivery as message:
+            messages.append(message)
+
+    return messages
 
 
 class TestMemoryQueue:
-    async def test_draining_with_success(self) -> None:
-        history = _History[str]()
+    async def test_aiter_with_success_yields_sent_messages(self) -> None:
+        queue = MemoryQueue[str](math.inf)
+        await queue.send("first")
+        await queue.send("second")
+        await queue.close()
+
+        assert await _receive_all(queue) == ["first", "second"]
+
+    async def test_call_with_success_sends_message(self) -> None:
+        queue = MemoryQueue[str](math.inf)
+        await queue("message")
+        await queue.close()
+
+        assert await _receive_all(queue) == ["message"]
+
+    async def test_send_with_closed_queue_raises_closed_resource_error(self) -> None:
+        queue = MemoryQueue[str]()
+        await queue.close()
+
+        with pytest.raises(ClosedResourceError):
+            await queue.send("message")
+
+    async def test_send_with_exited_queue_raises_closed_resource_error(self) -> None:
+        async with MemoryQueue[str]() as queue:
+            pass
+
+        with pytest.raises(ClosedResourceError):
+            await queue.send("message")
+
+    async def test_draining_with_success_dispatches_sent_messages_then_closes_queue(
+        self,
+    ) -> None:
+        dispatched = []
+
+        async def dispatcher(message: str) -> None:
+            dispatched.append(message)
+
+        async with MemoryQueue[str]().draining(dispatcher) as queue:
+            await queue.send("first")
+            await queue.send("second")
 
-        async with MemoryQueue[str]().draining(history) as queue:
-            await queue.send("message_1")
-            await queue.send("message_2")
+        assert dispatched == ["first", "second"]
 
         with pytest.raises(ClosedResourceError):
-            await queue.send("message_3")
+            await queue.send("third")
+
+    async def test_draining_with_middlewares_wraps_dispatcher(self) -> None:
+        calls = []
 
-        assert len(history) == 2
+        async def middleware(message: str) -> MiddlewareResult[Any]:
+            calls.append(f"before {message}")
+            yield
+            calls.append(f"after {message}")
 
-    async def test_draining_with_fail_silently(self) -> None:
         async def dispatcher(message: str) -> None:
-            raise NotImplementedError
+            calls.append(f"dispatch {message}")
 
-        async with MemoryQueue[str]().draining(dispatcher, fail_silently=True) as queue:
-            await queue.send("message_1")
-            await queue.send("message_2")
+        queue = MemoryQueue[str]()
 
-    async def test_draining_without_fail_silently(self) -> None:
+        async with queue.draining(dispatcher, middlewares=[middleware]):
+            await queue.send("message")
+
+        assert calls == ["before message", "dispatch message", "after message"]
+
+    async def test_draining_with_dispatcher_error_raises_exception_group(
+        self,
+    ) -> None:
         async def dispatcher(message: str) -> None:
-            raise NotImplementedError
+            raise ValueError(message)
+
+        queue = MemoryQueue[str]()
 
-        with pytest.raises(ExceptionGroup):
-            async with MemoryQueue[str]().draining(
-                dispatcher,
-                fail_silently=False,
-            ) as queue:
+        with pytest.raises(ExceptionGroup) as exc_info:
+            async with queue.draining(dispatcher):
                 await queue.send("message")
 
+        assert exc_info.group_contains(ValueError, match="message")
+
+    async def test_draining_with_dispatcher_error_and_fail_silently_dispatches_next_messages(
+        self,
+    ) -> None:
+        dispatched = []
+
+        async def dispatcher(message: str) -> None:
+            dispatched.append(message)
+            raise ValueError(message)
 
-@dataclass
-class _History[T]:
-    __records: list[T] = field(default_factory=list, init=False)
+        queue = MemoryQueue[str]()
 
-    async def __call__(self, message: T) -> None:
-        self.__records.append(message)
+        async with queue.draining(dispatcher, fail_silently=True):
+            await queue.send("first")
+            await queue.send("second")
 
-    def __len__(self) -> int:
-        return len(self.__records)
+        assert dispatched == ["first", "second"]