add tests for errors and exceptional cases

Author: mahenzon

tests/test_error.py

@@ -0,0 +1,88 @@
+import pytest
+
+from taskiq.error import Error
+from taskiq.exceptions import SecurityError, TaskiqResultTimeoutError
+
+
+class SimpleError(Error):
+    __template__ = "simple"
+
+
+class ValueTemplateError(Error):
+    __template__ = "value={value}"
+    value: int
+
+
+class DefaultValueTemplateError(Error):
+    __template__ = "value={value}"
+    value: int = 10
+
+
+class BaseError(Error):
+    __template__ = "base={base}, child={child}"
+    base: int = 1
+
+
+class ChildError(BaseError):
+    child: str
+
+
+class MissingAnnotationError(Error):
+    __template__ = "value={value}"
+
+
+class IndexedTemplateError(Error):
+    __template__ = "{payload[key]}"
+    payload: dict[str, str]
+
+
+def test_simple_error_message_and_repr() -> None:
+    error = SimpleError()
+    assert str(error) == "simple"
+    assert error.args == ("simple",)
+    assert repr(error).endswith(".SimpleError()")
+
+
+def test_template_with_required_value() -> None:
+    error = ValueTemplateError(value=3)
+    assert str(error) == "value=3"
+    assert repr(error).endswith(".ValueTemplateError(value=3)")
+
+
+def test_missing_argument_raises_type_error() -> None:
+    with pytest.raises(TypeError, match="Missing arguments: 'value'"):
+        ValueTemplateError()
+
+
+def test_undeclared_argument_raises_type_error() -> None:
+    with pytest.raises(TypeError, match="Undeclared arguments: 'extra'"):
+        ValueTemplateError(value=1, extra=2)
+
+
+def test_default_value_is_used_without_kwargs() -> None:
+    error = DefaultValueTemplateError()
+    assert str(error) == "value=10"
+    assert repr(error).endswith(".DefaultValueTemplateError(value=10)")
+
+
+def test_annotations_are_collected_from_inheritance() -> None:
+    error = ChildError(child="ok")
+    assert str(error) == "base=1, child=ok"
+    assert repr(error).endswith(".ChildError(base=1, child='ok')")
+
+
+def test_template_fields_must_be_annotated() -> None:
+    with pytest.raises(ValueError, match="Fields must be annotated: 'value'"):
+        MissingAnnotationError()
+
+
+def test_indexed_template_field_does_not_require_extra_annotation() -> None:
+    error = IndexedTemplateError(payload={"key": "value"})
+    assert str(error) == "value"
+
+
+def test_taskiq_exceptions_use_error_base_correctly() -> None:
+    timeout_error = TaskiqResultTimeoutError(timeout=1.5)
+    security_error = SecurityError(description="boom")
+    assert str(timeout_error) == "Waiting for task results has timed out, timeout=1.5"
+    assert str(security_error) == "Security exception occurred: boom"

tests/test_exceptions_flow.py

@@ -0,0 +1,66 @@
+import re
+
+import pytest
+
+from taskiq import InMemoryBroker
+from taskiq.brokers.shared_broker import AsyncSharedBroker
+from taskiq.exceptions import (
+    SharedBrokerListenError,
+    SharedBrokerSendTaskError,
+    TaskBrokerMismatchError,
+    UnknownTaskError,
+)
+from taskiq.message import BrokerMessage
+
+
+def _broker_message(task_name: str) -> BrokerMessage:
+    return BrokerMessage(
+        task_id="task-id",
+        task_name=task_name,
+        message=b"{}",
+        labels={},
+    )
+
+
+async def test_inmemory_broker_raises_unknown_task_error() -> None:
+    broker = InMemoryBroker()
+
+    with pytest.raises(
+        UnknownTaskError,
+        match=re.escape(
+            "Cannot send unknown task to the queue, task name - missing.task",
+        ),
+    ):
+        await broker.kick(_broker_message("missing.task"))
+
+
+async def test_shared_broker_raises_send_task_error() -> None:
+    broker = AsyncSharedBroker()
+
+    with pytest.raises(
+        SharedBrokerSendTaskError,
+        match="You cannot use kiq directly on shared task",
+    ):
+        await broker.kick(_broker_message("any.task"))
+
+
+async def test_shared_broker_raises_listen_error() -> None:
+    broker = AsyncSharedBroker()
+
+    with pytest.raises(SharedBrokerListenError, match="Shared broker cannot listen"):
+        await broker.listen()
+
+
+def test_registering_task_in_another_broker_raises_mismatch_error() -> None:
+    first_broker = InMemoryBroker()
+    second_broker = InMemoryBroker()
+
+    @first_broker.task(task_name="test.task")
+    async def test_task() -> None:
+        return None
+
+    with pytest.raises(
+        TaskBrokerMismatchError,
+        match="Task already has a different broker",
+    ):
+        second_broker._register_task(test_task.task_name, test_task)