Testing Utilities¶

@classmethod
def create(
    cls,
    *,
    name: str = "testapp",
    version: str = "1.0.0",
    dry_run: bool = False,
    lifespan: LifespanFunc | None = None,
    store: Store | None = None,
    run_periodic: bool = False,
    error_type_map: dict[type[Exception], str] | None = None,
    disclose_messages_for: frozenset[type[Exception]] | None = None,
    retained_cleanup_snapshot_key: SecretStr | None = None,
    clock: ClockPort | None = None,
    **settings_overrides: Any,
) -> Self:
    """Create a harness with fresh test doubles.

    Args:
        name: App name.
        version: App version.
        dry_run: When True, forward to App for dry-run adapter variants.
        lifespan: Optional lifespan context manager forwarded to
            :class:`App`.
        store: Optional :class:`Store` backend for device persistence.
        run_periodic: When True, periodic tasks will be started; when False,
            they will be suppressed for testing.
        error_type_map: Optional app-level exception → ``error_type`` map
            forwarded to :class:`App`, so tests can exercise the LEAK-01
            targeted opt-in end-to-end (see ADR-011).
        disclose_messages_for: Optional app-level message-disclosure set
            forwarded to :class:`App`, so tests can exercise the F-DP1
            decoupled opt-in end-to-end (see ADR-061).
        retained_cleanup_snapshot_key: Optional ADR-048 retained-cleanup
            snapshot HMAC signing key forwarded to :class:`App`, so tests
            can exercise the F-DP3 opt-in signing/verification end-to-end
            (see ADR-063).
        clock: Optional :class:`ClockPort` double to drive the harness.
            Defaults to :class:`FakeClock`, whose ``sleep()`` self-completes.
            Pass a :class:`ManualClock` to gate runner sleeps and assert
            that a scheduled tick did *not* fire — see ADR-071.
        **settings_overrides: Forwarded to :func:`make_settings`.

    Returns:
        A fully wired :class:`AppHarness` ready for test use.

    Raises:
        TypeError: If a ``settings_overrides`` key names neither a
            ``Settings`` field nor a supported runtime kwarg — a
            mistyped or unsupported keyword (e.g. ``clok=``) fails
            loudly instead of being silently ignored.
    """
    return cls(
        app=App(
            name=name,
            version=version,
            dry_run=dry_run,
            lifespan=lifespan,
            store=store,
            error_type_map=error_type_map,
            disclose_messages_for=disclose_messages_for,
            retained_cleanup_snapshot_key=retained_cleanup_snapshot_key,
        ),
        mqtt=MockMqttClient(),
        clock=clock if clock is not None else FakeClock(),
        settings=make_settings(**settings_overrides),
        shutdown_event=asyncio.Event(),
        run_periodic=run_periodic,
    )

async def run(self) -> None:
    """Run ``_run_async`` with the harness's test doubles."""
    periodic_backup = list(self.app._periodic)
    streams_backup = list(self.app._streams)
    if not self.run_periodic:
        self.app._periodic = []
    # Always suppress streams in harness.run() — use inject_stream() instead
    self.app._streams = []
    try:
        await self.app._run_async(
            settings=self.settings,
            shutdown_event=self.shutdown_event,
            mqtt=self.mqtt,
            clock=self.clock,
        )
    finally:
        self.app._periodic = periodic_backup
        self.app._streams = streams_backup

def trigger_shutdown(self) -> None:
    """Signal the shutdown event."""
    self.shutdown_event.set()

async def inject_stream(
    self,
    name: str,
    *items: Any,
    shutdown: bool = True,
    ctx: DeviceContext | None = None,
    store: Store | None = None,
    providers: dict[type, Any] | None = None,
    adapters: dict[type, object] | None = None,
) -> None:
    """Push items into a named stream handler for testing.

    Finds the registered @app.stream handler by name, creates a Stream,
    pushes the provided items, optionally signals shutdown, and runs the
    handler directly (bypassing adapter lifecycle).

    Args:
        name: Stream handler name as registered with @app.stream.
        *items: Items to push into the stream.
        shutdown: When True (default), call stream.shutdown() after all
            items are pushed so the handler's async for loop terminates.
        ctx: Optional :class:`DeviceContext` override.  When ``None``
            (default), a context is constructed from the harness doubles
            (mqtt, settings, clock, shutdown_event) so handlers can call
            ``ctx.publish_state`` etc. and assertions use
            ``harness.mqtt.published``.
        store: Optional :class:`Store` backend for persistence.  When
            ``None``, falls back to ``app._store`` if configured.  The
            harness creates a :class:`DeviceStore` keyed by *name*,
            loads it before the handler runs, and saves it afterward.
        providers: Extra DI providers merged into the provider map with
            the highest priority (override everything else).
        adapters: Concrete adapter instances injected by their concrete
            type into both the DI provider map and the
            :class:`DeviceContext` adapters dict.  Allows stream handlers
            to access adapters for non-lifecycle operations without
            running the hardware lifecycle (open/start_scan).

    Note:
        When *ctx* is supplied it replaces the entire :class:`DeviceContext`
        — harness doubles (mqtt, clock, shutdown_event) are not merged in.
        *adapters* are added to the DI providers map but **not** injected
        into the explicitly supplied *ctx*.  If you need both a custom
        context and adapter injection, build the context with the adapters
        you need and pass both *ctx* and *adapters*.

    Raises:
        ValueError: If no stream handler with *name* is registered.
        TypeError: If a required dependency (e.g. :class:`DeviceStore`)
            cannot be resolved from the provider map.
    """
    try:
        reg = next(r for r in self.app._streams if r.name == name)
    except StopIteration:
        msg = f"No stream handler named '{name}' found"
        raise ValueError(msg) from None

    stream: Stream[Any] = Stream()
    for item in items:
        stream.put(item)

    if shutdown:
        asyncio.create_task(
            _stream_auto_shutdown(stream), name=f"inject-shutdown:{name}"
        )

    resolved_adapters: dict[type, object] = dict(adapters) if adapters else {}
    ctx = self._make_stream_ctx(name, reg, resolved_adapters, ctx)
    device_store = await self._make_device_store(name, store, providers)
    base_providers = self._build_inject_providers(
        name, ctx, resolved_adapters, device_store, providers
    )

    try:
        await _run_stream_handler(reg, stream, base_providers, self.app._reactors)
    finally:
        # Retrieve the DeviceStore from final providers — _make_device_store
        # returns None when a pre-supplied store was passed via providers,
        # but in that case base_providers.get(DeviceStore) still returns it.
        await async_save_store_on_shutdown(base_providers.get(DeviceStore), name)

def override_state(self, state_type: type, instance: Any) -> None:
    """Override a @app.state factory with a pre-built test double.

    Bypasses the factory entirely; *instance* is injected directly
    into the DI container at bootstrap.  Call before :meth:`run`.

    Args:
        state_type: The type returned by the factory (the DI key).
        instance: The test double to inject.

    Raises:
        TypeError: If *instance* is not an instance of *state_type*.
    """
    if not isinstance(instance, state_type):
        raise TypeError(
            f"override_state: expected an instance of {state_type.__name__!r}, "
            f"got {type(instance).__name__!r}"
        )
    self.app._state_overrides[state_type] = instance

async def tick_periodic(self, name: str) -> None:
    """Invoke one cycle of the named periodic handler (bypasses interval).

    Directly calls the handler's function with injected arguments —
    skips the asyncio sleep so you can test the handler logic
    without waiting for the interval.

    Args:
        name: The periodic task name as registered with ``@app.periodic``.

    Raises:
        ValueError: if no periodic task with *name* exists.
    """
    from cosalette._injection import resolve_request_kwargs

    try:
        reg = next(r for r in self.app._periodic if r.name == name)
    except StopIteration:
        msg = f"No periodic task named '{name}' found"
        raise ValueError(msg) from None

    # Build a provider map matching production _build_periodic_providers:
    # settings under every Settings base class, clock, logger, state overrides
    providers: dict[type, Any] = {}
    settings = self.settings
    for cls in type(settings).__mro__:
        if isinstance(cls, type) and issubclass(cls, Settings):
            providers[cls] = settings
    providers[ClockPort] = self.clock
    providers[logging.Logger] = logging.getLogger(f"cosalette.periodic.{name}")
    providers.update(self.app._state_overrides)
    kwargs = resolve_request_kwargs(reg.injection_plan, providers)
    await reg.func(**kwargs)

def published(self) -> list[tuple[str, str, bool, int]]:
    """Return a snapshot of all MQTT messages published so far.

    Returns:
        Snapshot list of ``(topic, payload, retain, qos)`` tuples. This
        is a copy — mutating the returned list does not affect the
        :class:`MockMqttClient` internal state.
    """
    return list(self.mqtt.published)

def messages_for(self, topic: str) -> list[tuple[str, bool, int]]:
    """Return all messages published to *topic*.

    Args:
        topic: MQTT topic filter (exact match only).

    Returns:
        List of ``(payload, retain, qos)`` tuples for the given *topic*.
    """
    return self.mqtt.get_messages_for(topic)

def last_published(self) -> tuple[str, str, bool, int] | None:
    """Return the most recent MQTT publish, or ``None`` if no publishes.

    Returns:
        ``(topic, payload, retain, qos)`` tuple or ``None``.
    """
    return self.mqtt.published[-1] if self.mqtt.published else None

def assert_state(
    self,
    topic: str,
    expected: dict[str, Any],
    *,
    count: int | None = None,
) -> None:
    """Assert that *topic* has a retained JSON message containing *expected*.

    Checks that at least one retained message on *topic* has a JSON payload
    that is a deep recursive superset of *expected*.  An empty *expected*
    dict matches any retained JSON message.

    Args:
        topic: MQTT topic to check (exact match).
        expected: Key/value subset that must appear in at least one payload.
            List-typed values in *expected* use exact equality, not element
            containment.
        count: Optional exact number of messages that must have been
            published to *topic*.

    Raises:
        AssertionError: If no messages for *topic*; if *count* mismatches;
            if no payload is a superset of *expected*; or if the matching
            message was not retained.
    """
    messages = self.messages_for(topic)
    if not messages:
        raise AssertionError(f"No messages published to {topic!r}")
    if count is not None and len(messages) != count:
        raise AssertionError(
            f"Expected {count} message(s) to {topic!r}, got {len(messages)}"
        )
    found_retained, found_any_subset, parseable_payloads, skipped = (
        _scan_state_messages(messages, expected)
    )
    if found_retained:
        return
    if found_any_subset:
        raise AssertionError(
            f"Matching message on {topic!r} exists but was not retained "
            f"(state publications must be retained)"
        )
    skip_note = (
        f"\n({skipped} message(s) skipped — non-JSON-object payload)"
        if skipped
        else ""
    )
    raise AssertionError(
        f"No message on {topic!r} contains {expected!r}.\n"
        f"Parseable JSON-dict payloads: {parseable_payloads}{skip_note}"
    )

def assert_subscribed(self, topic: str) -> None:
    """Assert that *topic* has been subscribed to.

    Args:
        topic: Exact MQTT topic string to look up.

    Raises:
        AssertionError: If *topic* is not in ``self.mqtt.subscriptions``.
    """
    if topic not in self.mqtt.subscriptions:
        raise AssertionError(
            f"Topic {topic!r} not subscribed.\n"
            f"Actual subscriptions: {self.mqtt.subscriptions!r}"
        )

def assert_published(
    self,
    topic: str,
    *,
    contains: str | None = None,
    count: int | None = None,
) -> None:
    """Assert that *topic* has published messages matching criteria.

    Args:
        topic: MQTT topic to check (exact match).
        contains: Optional substring that must appear in at least one
            payload for *topic*.
        count: Optional exact number of messages that must have been
            published to *topic*.

    Raises:
        AssertionError: If no messages for *topic*, or if *contains* is
            not found in any payload, or if message count doesn't match
            *count*.
    """
    messages = self.messages_for(topic)
    if not messages:
        raise AssertionError(f"No messages published to {topic!r}")
    if count is not None and len(messages) != count:
        raise AssertionError(
            f"Expected {count} message(s) to {topic!r}, got {len(messages)}"
        )
    if contains is not None and not any(
        contains in payload for payload, _, _ in messages
    ):
        raise AssertionError(f"No message on {topic!r} contains {contains!r}")

async def inject_command(
    self,
    device: str | None,
    payload: str | dict[str, Any],
    *,
    topic: str | None = None,
    unsafe: bool = False,
) -> None:
    """Simulate an inbound MQTT command to *device*.

    Delivers a message to ``{topic_prefix}/{device}/set`` (or
    ``{topic_prefix}/set`` for root commands) via the
    :class:`MockMqttClient`, triggering registered command callbacks.

    This is an MQTT-delivery helper — the app must be running and
    callbacks must be registered for the command to be processed.

    Args:
        device: Device name as registered with ``@app.command``, or
            ``None`` for root commands (matching ``@app.command(None)``
            registration semantics). ``None`` constructs the topic as
            ``{prefix}/set``; any non-empty string constructs
            ``{prefix}/{device}/set``.
        payload: MQTT payload — either a JSON string or a dict that will
            be serialized to JSON.
        topic: Optional explicit topic override. When ``None`` (default),
            the topic is constructed from *device*.
        unsafe: Skip device-name validation to allow adversarial topics
            (wildcards, separators, control characters) that production
            registration would reject. Defaults to ``False``.

    See Also:
        :meth:`call_command` for direct command handler invocation without
            requiring the app to be running.
    """
    if topic is None:
        topic_prefix = self._topic_prefix
        if device:
            # Guard against tests that would silently exercise topics the
            # production registration path rejects (wildcards, control
            # chars, path separators). Adversarial tests can opt out with
            # ``unsafe=True``.
            if not unsafe:
                from cosalette._registration._validation import (
                    validate_mqtt_name,
                )

                validate_mqtt_name(device)
            topic = f"{topic_prefix}/{device}/set"
        else:
            topic = f"{topic_prefix}/set"
    payload_str = _json_dumps(payload) if isinstance(payload, dict) else payload
    await self.mqtt.deliver(topic, payload_str)

async def call_command(
    self,
    name: str,
    payload: str | dict[str, object],
    *,
    topic: str | None = None,
) -> None:
    """Directly invoke a registered ``@app.command`` handler.

    Resolves the handler by *name*, injects dependencies, calls it with
    the deserialized *payload*, and publishes any returned state to
    ``harness.mqtt`` — mirroring production execution without requiring
    the app to be running.

    Supports production request binding including typed Pydantic payloads
    (``Annotated[Model, Payload()]``), ``payload``/``topic``/``message``
    parameters, ``DeviceContext``, and simple DI providers available to
    ``CommandRunner``. Does NOT run adapter lifecycle, state factory
    lifecycle, or reactors.

    Args:
        name: Command handler name as registered with ``@app.command``.
            Supports router-prefixed names like ``"router/sub"``. For
            root commands registered with ``@app.command(None)``, pass
            the function name.
        payload: MQTT payload — either a JSON string or a dict that will
            be serialized to JSON.
        topic: Optional MQTT topic string. When ``None`` (default),
            constructs ``{prefix}/{name}/set`` or ``{prefix}/set`` for
            root commands.

    Raises:
        ValueError: If no command handler with *name* is registered.
        Exception: Any exception raised by the handler is propagated.

    Note:
        For tests requiring adapter lifecycle, state factory lifecycle,
        or reactor dispatch, use :meth:`inject_command` with the app
        running. ``init=`` command callbacks are NOT run; handlers that
        cache ``init`` results will receive ``None`` for those
        dependencies. Reactor dispatch is disabled; if the handler
        triggers side-effects via reactors, use :meth:`inject_command`
        with the app running instead.

    See Also:
        :meth:`inject_command` for MQTT-delivery simulation requiring the
        app to be running.
    """
    topic_prefix = self._topic_prefix

    # Find the command registration
    try:
        reg = next(r for r in self.app._commands if r.name == name)
    except StopIteration:
        msg = f"No command handler named '{name}' found"
        raise ValueError(msg) from None

    # Construct topic if not provided
    if topic is None:
        if reg.is_root:
            topic = f"{topic_prefix}/set"
        else:
            topic = f"{topic_prefix}/{name}/set"

    # Serialize payload using the project's JSON backend (orjson) for
    # consistency with production encoding behaviour.
    payload_str = _json_dumps(payload) if isinstance(payload, dict) else payload

    # Build DeviceContext for command execution
    ctx = DeviceContext(
        name=name,
        settings=self.settings,
        mqtt=self.mqtt,
        topic_prefix=topic_prefix,
        shutdown_event=self.shutdown_event,
        adapters={},
        clock=self.clock,
        is_root=reg.is_root,
    )

    # Create CommandRunner and execute (reactors=None skips reactor dispatch)
    cmd_runner = CommandRunner(store=self.app._store)
    error_publisher = ErrorPublisher(
        mqtt=self.mqtt,
        topic_prefix=topic_prefix,
    )

    await cmd_runner.run_command(
        reg=reg,
        ctx=ctx,
        topic=topic,
        payload=payload_str,
        error_publisher=error_publisher,
        reactors=None,
    )

async def advance_time(self, seconds: float) -> None:
    """Move the harness clock forward by *seconds*, releasing due sleeps.

    Honest scheduler control under a gating :class:`ManualClock`: this
    delegates to :meth:`ManualClock.advance`, so virtual time steps
    deadline by deadline, every sleeper due within the span is released,
    and the event loop is driven to quiescence before returning. A runner
    parked on ``clock.sleep`` therefore wakes exactly when the name says
    it should.

    Under a non-gating clock (the default :class:`FakeClock`) there are no
    waiters to release, so this delegates to ``clock.sleep`` — it advances
    virtual time by *seconds* and yields once to the event loop, the same
    behaviour these tests have always had.

    Args:
        seconds: Virtual seconds to advance.

    See Also:
        ADR-071 for the two clock doubles and their contracts.
    """
    if isinstance(self.clock, ManualClock):
        await self.clock.advance(seconds)
    else:
        await self.clock.sleep(seconds)

async def wait_for_publish_count(
    self, topic: str, count: int, *, max_rounds: int = 10_000
) -> None:
    """Wait until *topic* has at least *count* published messages.

    The supported replacement for the hand-rolled
    ``for _ in range(10_000): await asyncio.sleep(0)`` spin: it yields to
    the event loop until the awaited publish lands, then returns.

    Under a gating :class:`ManualClock` it delegates to
    ``settle(until=...)`` — a real wait that never moves virtual time and
    raises if the count is never reached. Under a non-gating clock it
    yields up to *max_rounds* times, raising the same way on timeout.

    Note:
        This never advances time. Under :class:`ManualClock`, a publish
        gated behind a scheduled sleep needs an explicit
        :meth:`advance_time` (or ``clock.advance``) first; this call then
        waits for the released work to reach the wire.

    Args:
        topic: Exact MQTT topic to count publishes on.
        count: Target number of messages on *topic* (``>=``).
        max_rounds: Event-loop rounds to spend before giving up.

    Raises:
        ValueError: If *count* or *max_rounds* is not a positive integer.
        RuntimeError: If *count* messages never land within *max_rounds*.
    """
    if count < 1:
        msg = f"count must be a positive integer, got {count!r}"
        raise ValueError(msg)
    if max_rounds < 1:
        msg = f"max_rounds must be a positive integer, got {max_rounds!r}"
        raise ValueError(msg)

    def reached() -> bool:
        return len(self.messages_for(topic)) >= count

    if isinstance(self.clock, ManualClock):
        try:
            await self.clock.settle(until=reached, max_rounds=max_rounds)
        except RuntimeError as exc:
            raise self._publish_count_timeout(topic, count, max_rounds) from exc
        return
    for _ in range(max_rounds):
        if reached():
            return
        await asyncio.sleep(0)
    # Re-check after the final yield: a publish that lands during that last
    # ``asyncio.sleep(0)`` would otherwise time out spuriously.
    if reached():
        return
    raise self._publish_count_timeout(topic, count, max_rounds)

async def run_stream(
    self,
    func: Any,
    adapters: dict[type, Any],
    *,
    shutdown: asyncio.Event | None = None,
) -> None:
    """Run a stream handler's full lifecycle (open → scan → close).

    Constructs a minimal :class:`_StreamRegistration` from *func*, then
    calls :func:`run_stream` with the provided *adapters*.  Useful for
    testing stream handler behaviour without wiring a full app.

    Args:
        func: The async-generator stream handler to run.
        adapters: Resolved adapter map keyed by port type
            (e.g. ``{StreamablePort[Item]: my_port_instance}``).
        shutdown: Optional :class:`asyncio.Event` to trigger graceful
            shutdown.  Defaults to :attr:`shutdown_event`.
    """
    from cosalette._injection import build_injection_plan
    from cosalette._registration import _StreamRegistration
    from cosalette._runners._stream_runner import run_stream as _run_stream

    plan = build_injection_plan(func)
    reg = _StreamRegistration(
        name="test_stream",
        func=func,
        injection_plan=plan,
        enabled_spec=True,
        summary=None,
        behavior=None,
        effects=None,
    )
    await _run_stream(
        reg, adapters, {}, shutdown if shutdown is not None else self.shutdown_event
    )

def now(self) -> float:
    """Return the manually set time value."""
    return self._time

def advance(self, seconds: float) -> None:
    """Move virtual time forward by *seconds*, without sleeping.

    Relative to the current value, unlike assigning ``_time``,
    which sets virtual time absolutely.  Unlike :meth:`sleep` this
    does not yield to the event loop, so no other task gets to run
    — use it to simulate work that consumed time.

    Args:
        seconds: Virtual seconds to add.  ``0`` is a no-op.

    Raises:
        ValueError: If *seconds* is negative.  A monotonic clock
            never runs backwards.

    Example::

        clock = FakeClock(10.0)
        clock.advance(5.0)
        assert clock.now() == 15.0
    """
    self._reject_negative(seconds)
    self._time += seconds

async def sleep(self, seconds: float) -> None:
    """Advance virtual time by *seconds* with no real delay.

    Allows tests to exercise sleep-dependent code paths
    without wall-clock waiting.  The ``asyncio.sleep(0)``
    yields to the event loop so concurrent tasks interleave
    correctly.

    The duration is charged to the awaiting task's own deadline, not
    to a single shared accumulator: ``now()`` moves to the latest
    deadline any task has reached, so a concurrent sleeper never
    pushes this one's next wake further out.

    The task's base is read *before* the yield, so a concurrent
    sleeper that moves ``now()`` while this call is parked cannot be
    mistaken for this task's own starting point — the per-task
    guarantee then holds whatever order the loop resumes the sleepers.
    """
    if seconds <= 0:
        await asyncio.sleep(0)
        return
    task = asyncio.current_task()
    if task is None:  # pragma: no cover — sleep() is always awaited in a task
        await asyncio.sleep(0)
        self._time += seconds
        return
    base = self._wakes.get(task, self._time)
    await asyncio.sleep(0)
    self._time = max(self._time, self._charge(task, base, seconds))
    self._seen = self._time

def now(self) -> float:
    """Return the manually set time value."""
    return self._time

async def sleep(self, seconds: float) -> None:
    """Block until :meth:`advance` moves time to ``now() + seconds``.

    Nothing else completes a *positive* sleep: no wall-clock time
    passes, and no number of event-loop iterations releases it.  The
    deadline is captured per call, so a concurrent sleeper's duration
    never leaks into this one's timeline.

    A non-positive *seconds* is the deliberate carve-out to that
    guarantee: it is already elapsed by definition, so it yields to
    the event loop once and returns, exactly like ``asyncio.sleep(0)``.
    The framework's own ``sleep(max(0.0, deadline - now()))`` throttle
    arithmetic depends on this — a gating ``sleep(0)`` would deadlock
    the runners this clock exists to test.  The cost is real: a
    consumer whose deadline has already gone stale computes ``0.0``
    every cycle and free-runs without any :meth:`advance`, so this
    clock does not gate it at all.  Such a loop churns the observed
    operation counter, so :meth:`settle` catches it and raises rather
    than returning a false "quiescent" — but only after the loop has
    run some cycles, and the work those cycles did has really
    happened.

    Args:
        seconds: Virtual seconds to wait.  ``<= 0`` yields only.

    Raises:
        ValueError: If *seconds* is NaN or infinite — such a deadline
            would never be reached, gating the sleeper forever.
    """
    self._reject_non_finite(seconds, caller="sleep")
    self._ops += 1
    if seconds <= 0:
        await asyncio.sleep(0)
        return
    waiter = _Waiter(self._time + seconds, asyncio.Event())
    self._waiters.append(waiter)
    try:
        await waiter.event.wait()
    finally:
        self._ops += 1
        # A cancel before release leaves the waiter registered; a
        # release has already dropped it, so removal is best-effort.
        with contextlib.suppress(ValueError):
            self._waiters.remove(waiter)