Testing

This guide covers how to write unit and integration tests for EDRI applications. The framework's explicit, event-driven design makes components straightforward to test in isolation.

---

Overview

EDRI applications have three layers to test:

Layer	Test focus	Isolation level
Event definitions	Correct field types, response auto-init	Pure Python
Manager handlers	Business logic, response fields, emitted events	Unit — mock router queue
API layer	URL matching, input extraction, validation	Integration — live Listener

The standard library unittest is used throughout the project. unittest.mock handles the multiprocessing primitives (Queue, Connection, Process).

---

Testing event definitions

Events are dataclasses — test that they construct correctly and that their responses auto-initialize.

import unittest
from edri.dataclass.event import event, Event
from edri.dataclass.response import Response, response, ResponseStatus


@response
class GreetingResponse(Response):
    message: str


@event
class Greeting(Event):
    name: str
    response: GreetingResponse


class TestGreetingEvent(unittest.TestCase):

    def test_response_auto_instantiated(self):
        """response field is created automatically — not passed by the caller."""
        ev = Greeting(name="Alice")
        self.assertIsInstance(ev.response, GreetingResponse)
        self.assertIsNone(ev.response.message)

    def test_status_starts_none(self):
        ev = Greeting(name="Alice")
        self.assertEqual(ev.response.get_status(), ResponseStatus.NONE)

    def test_status_auto_set_on_field_assignment(self):
        ev = Greeting(name="Alice")
        ev.response.message = "Hello, Alice!"
        self.assertEqual(ev.response.get_status(), ResponseStatus.OK)

    def test_has_response_false_before_modification(self):
        ev = Greeting(name="Alice")
        self.assertFalse(ev.has_response())

    def test_has_response_true_after_modification(self):
        ev = Greeting(name="Alice")
        ev.response.message = "Hello, Alice!"
        self.assertTrue(ev.has_response())

---

Testing Manager handlers

The recommended test pattern uses the ManagerTestHarness helper (edri.testing, Experimental). It wraps the internal setup steps so your tests use only public API:

import unittest
from unittest.mock import MagicMock
from multiprocessing.queues import Queue

from edri.abstract import ManagerBase, request
from edri.dataclass.directive.http import NotFoundResponseDirective
from edri.dataclass.response import ResponseStatus
from edri.testing import ManagerTestHarness

# Import the events being tested
from events.items import ItemGet, ItemCreate


class ItemsManager(ManagerBase):
    def __init__(self, router_queue=None):
        super().__init__(router_queue)
        self._items: dict[int, dict] = {}

    @request
    def get_item(self, event: ItemGet) -> None:
        item = self._items.get(event.item_id)
        if item is None:
            event.response.set_status(ResponseStatus.FAILED)
            event.response.add_directive(NotFoundResponseDirective())
            return
        event.response.name = item["name"]


class TestItemsManager(unittest.TestCase):

    def setUp(self):
        self.router_queue = MagicMock(spec=Queue)
        self.harness = ManagerTestHarness(
            ItemsManager(router_queue=self.router_queue)
        )
        self.harness.prepare()
        # Pre-seed some data
        self.harness.manager._items = {1: {"id": 1, "name": "widget", "quantity": 5}}

    def test_get_existing_item(self):
        event = ItemGet(item_id=1)
        self.harness.resolve(event)

        self.assertEqual(self.harness.get_status(event), ResponseStatus.OK)
        self.assertEqual(event.response.name, "widget")

    def test_get_missing_item_returns_404(self):
        event = ItemGet(item_id=999)
        self.harness.resolve(event)

        self.assertEqual(self.harness.get_status(event), ResponseStatus.FAILED)
        directives = self.harness.get_directives(event)
        self.assertTrue(any(isinstance(d, NotFoundResponseDirective) for d in directives))

Testing that a Manager emits a new event

When a handler emits a new event via self.router_queue.put(...), assert on the mock:

def test_manager_emits_notification(self):
    event = ItemCreate(name="gadget", quantity=3)
    self.harness.resolve(event)

    # Verify the response was set
    self.assertEqual(self.harness.get_status(event), ResponseStatus.OK)

    # Verify a new event was emitted to the router
    self.router_queue.put.assert_called_once()
    emitted = self.router_queue.put.call_args[0][0]
    self.assertIsInstance(emitted, SomeNotificationEvent)

Testing after_start()

Override the method and call it directly — there is no need to start an actual process:

def test_after_start_seeds_cache(self):
    self.harness.manager.after_start()
    self.assertTrue(
        self.harness.manager.has_config("default-config")
    )

---

Testing response directives

Check that the correct directives are attached to the response:

from edri.dataclass.directive.http import (
    NotFoundResponseDirective,
    AccessDeniedResponseDirective,
    BadRequestResponseDirective,
)


def test_access_denied_directive_attached(self):
    event = ProtectedEvent(user_id=None)
    self.harness.resolve(event)

    directives = self.harness.get_directives(event)
    self.assertTrue(
        any(isinstance(d, AccessDeniedResponseDirective) for d in directives),
        "Expected AccessDeniedResponseDirective in response directives"
    )

---

Testing Workers

Workers run their do() method synchronously in a test — no process needed. Pass a mocked Manager connection to capture emitted events.

import unittest
from unittest.mock import MagicMock
from multiprocessing.connection import Connection

from edri.abstract.worker import WorkerProcess
from events.items import ItemCreate


class ProcessingWorker(WorkerProcess[ItemCreate]):
    def do(self) -> None:
        result = self.event.name.upper()
        self.event.response.processed_name = result
        self.event_send(self.event)


class TestProcessingWorker(unittest.TestCase):

    def setUp(self):
        self.pipe = MagicMock(spec=Connection)
        self.event = ItemCreate(name="widget", quantity=1)
        self.worker = ProcessingWorker(self.pipe, self.event, "TestWorker")

    def test_do_sends_event_with_processed_name(self):
        self.worker.do()

        self.pipe.send.assert_called_once()
        sent_event = self.pipe.send.call_args[0][0]
        self.assertEqual(sent_event.response.processed_name, "WIDGET")

Testing that Worker handles errors gracefully

class FailingWorker(WorkerProcess[ItemCreate]):
    def do(self) -> None:
        raise ValueError("something went wrong")


def test_worker_sends_quit_on_exception(self):
    worker = FailingWorker(self.pipe, self.event, "FailWorker")
    worker.run()  # must not raise

    # The last event sent is WorkerQuit — an internal framework event.
    # Application tests typically match on event attributes instead of
    # importing internal types, but contributor tests may need to assert
    # on the exact event class.
    last_sent = self.pipe.send.call_args_list[-1][0][0]
    self.assertIsNotNone(last_sent)

---

Testing Middleware

Middleware can be tested without starting a server — call process_request and process_response directly.

import unittest
from unittest.mock import MagicMock

from edri.api.middleware import Middleware, MiddlewareControl
from edri.dataclass.directive.base import UnauthorizedResponseDirective
from edri.dataclass.response import ResponseStatus


class AuthMiddleware(Middleware):
    def process_request(self, event, info) -> MiddlewareControl:
        if not getattr(event, "bearer_token", None):
            event.response.set_status(ResponseStatus.FAILED)
            event.response.add_directive(UnauthorizedResponseDirective("Missing token"))
        return MiddlewareControl.continue_processing()


class TestAuthMiddleware(unittest.TestCase):

    def setUp(self):
        self.middleware = AuthMiddleware()

    def test_missing_token_sets_failed_status(self):
        event = MagicMock()
        event.bearer_token = None
        event.response = MagicMock()

        result = self.middleware.process_request(event, None)

        self.assertEqual(result.action, MiddlewareControl.ActionType.CONTINUE)
        event.response.set_status.assert_called_once_with(ResponseStatus.FAILED)
        event.response.add_directive.assert_called_once()

    def test_valid_token_continues(self):
        event = MagicMock()
        event.bearer_token = ["Bearer valid-token"]
        event.response = MagicMock()

        result = self.middleware.process_request(event, None)

        self.assertEqual(result.action, MiddlewareControl.ActionType.CONTINUE)
        event.response.set_status.assert_not_called()

---

Testing validators

Validators are plain callables — call them directly:

import unittest
from edri.utility.validation import IntegerValidator, StringValidator


class TestValidators(unittest.TestCase):

    def test_integer_validator_accepts_valid(self):
        result = IntegerValidator("42", minimum=1, maximum=100)
        self.assertEqual(result, 42)

    def test_integer_validator_rejects_below_minimum(self):
        with self.assertRaises(ValueError):
            IntegerValidator("0", minimum=1)

    def test_string_validator_rejects_too_short(self):
        with self.assertRaises(ValueError):
            StringValidator("ab", minimum_length=3)

Integration tests

To test an EDRI application against a live Listener, use the subprocess pattern from examples/integration_test.py. The test spawns a minimal application defined in examples/integration_app/main.py as a subprocess:

examples/integration_app/main.py

"""Minimal EDRI app for integration test."""
import os
from http import HTTPMethod

from edri import EDRI
from edri.abstract import ManagerBase, request
from edri.api.dataclass.api_event import ApiType, api
from edri.dataclass.event import Event
from edri.dataclass.response import Response, response


@response
class GreetingResponse(Response):
    message: str


@api(
    url="/hello/{name}",
    method=HTTPMethod.GET,
    exclude={ApiType.HTML, ApiType.WS},
)
class Greeting(Event):
    name: str
    response: GreetingResponse


class GreetingManager(ManagerBase):
    @request
    def greet(self, event: Greeting) -> None:
        event.response.message = f"Hello, {event.name}!"


if __name__ == "__main__":
    os.environ.setdefault("EDRI_HOST", "127.0.0.1")
    os.environ.setdefault("EDRI_PORT", "8080")

    edri = EDRI()
    edri.add_component(GreetingManager(edri.router_queue))
    edri.start_api()
    edri.run()

The test fixture:

examples/integration_test.py

"""tests for my app -- integration"""
import http.client
import json
import os
import signal
import socket
import subprocess
import sys
import tempfile
import time
import unittest


def _free_port():
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
        s.bind(("127.0.0.1", 0))
        return s.getsockname()[1]


def _wait_for_port(host, port, proc, timeout=30):
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        if proc.poll() is not None:
            return False
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
            s.settimeout(1)
            if s.connect_ex((host, port)) == 0:
                return True
        time.sleep(0.1)
    return False


@unittest.skipUnless(os.environ.get("MYAPP_RUN_INTEGRATION"), "set MYAPP_RUN_INTEGRATION=1")
class TestMyAppIntegration(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.host = "127.0.0.1"
        cls.port = _free_port()
        examples_dir = os.path.dirname(__file__)
        docs_source_dir = os.path.dirname(examples_dir)
        repo_root = os.path.dirname(os.path.dirname(docs_source_dir))
        env = dict(os.environ, EDRI_HOST=cls.host, EDRI_PORT=str(cls.port))
        env.setdefault("PYTHONPATH", repo_root)
        cls.stdout_file = tempfile.NamedTemporaryFile(delete=False, suffix=".stdout")
        cls.stderr_file = tempfile.NamedTemporaryFile(delete=False, suffix=".stderr")
        cls.proc = subprocess.Popen(
            [sys.executable, "-m", "examples.integration_app.main"],
            cwd=docs_source_dir,
            env=env,
            start_new_session=True,
            stdout=cls.stdout_file,
            stderr=cls.stderr_file,
        )
        cls.stdout_file.close()
        cls.stderr_file.close()
        if not _wait_for_port(cls.host, cls.port, cls.proc):
            stdout = cls._read_log(cls.stdout_file.name)
            stderr = cls._read_log(cls.stderr_file.name)
            cls._kill_process()
            cls._cleanup_tempfiles()
            raise RuntimeError(
                f"server did not start on {cls.host}:{cls.port}\n"
                f"stdout:\n{stdout}\n"
                f"stderr:\n{stderr}"
            )

    @classmethod
    def tearDownClass(cls):
        cls._kill_process()
        cls._cleanup_tempfiles()

    @classmethod
    def _cleanup_tempfiles(cls):
        for path in (cls.stdout_file.name, cls.stderr_file.name):
            if os.path.exists(path):
                os.unlink(path)

    @classmethod
    def _kill_process(cls):
        if cls.proc.poll() is not None:
            return
        # os.killpg is POSIX-only (macOS, Linux). On Windows, use
        # cls.proc.terminate() instead.
        try:
            os.killpg(os.getpgid(cls.proc.pid), signal.SIGTERM)
            cls.proc.wait(timeout=10)
        except (OSError, subprocess.TimeoutExpired):
            try:
                os.killpg(os.getpgid(cls.proc.pid), signal.SIGKILL)
                cls.proc.wait(timeout=5)
            except (OSError, subprocess.TimeoutExpired):
                pass

    @classmethod
    def _read_log(cls, path):
        try:
            with open(path) as f:
                return f.read()
        except OSError:
            return "(unreadable)"

    def test_greeting(self):
        conn = http.client.HTTPConnection(self.host, self.port, timeout=15)
        try:
            conn.request("GET", "/hello/World", headers={"Accept": "application/json"})
            resp = conn.getresponse()
            self.assertEqual(resp.status, 200)
            data = json.loads(resp.read())
            self.assertEqual(data["data"]["message"], "Hello, world!")
        finally:
            conn.close()


if __name__ == "__main__":
    unittest.main()

---

Running tests

# All tests with coverage
coverage run -m xmlrunner discover -s tests -p "test_*.py" --output-file report.xml
coverage report -m

# A single test module
python -m unittest tests.abstract.manager.test_manager_base

# A single test case
python -m unittest tests.abstract.manager.test_manager_base.TestManagerBase.test_resolve

# Integration tests (example from the tutorial above)
MYAPP_RUN_INTEGRATION=1 python docs/source/examples/integration_test.py

# EDRI's own integration tests
EDRI_RUN_INTEGRATION=1 python -m xmlrunner discover -s tests/integration -p "test_*.py" -t .

---

Tips for testing in EDRI

• Do not start actual processes in unit tests. Mock Queue and Connection — test the logic, not the multiprocessing plumbing.

• Use ManagerTestHarness to call prepare() then resolve() — it uses only public API so your tests survive internal refactoring.

• Check router_queue.put.call_args_list to assert on all emitted events, not just the first.

• Workers' run() is safe to call in tests — it calls do(), catches exceptions, and sends WorkerQuit. Test do() directly if you want to isolate the work itself.

• Event fields are plain Python attributes after construction — read and write them directly in tests.

• Use get_directives(event) (from ManagerTestHarness) to assert on directive types — avoids accessing private _directives.