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,
    **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.
        **settings_overrides: Forwarded to :func:`make_settings`.

    Returns:
        A fully wired :class:`AppHarness` ready for test use.
    """
    return cls(
        app=App(
            name=name,
            version=version,
            dry_run=dry_run,
            lifespan=lifespan,
            store=store,
        ),
        mqtt=MockMqttClient(),
        clock=FakeClock(),
        settings=make_settings(**settings_overrides),
        shutdown_event=asyncio.Event(),
    )

async def run(self) -> None:
    """Run ``_run_async`` with the harness's test doubles."""
    await self.app._run_async(
        settings=self.settings,
        shutdown_event=self.shutdown_event,
        mqtt=self.mqtt,
        clock=self.clock,
    )

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

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

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.
    """
    await asyncio.sleep(0)
    if seconds > 0:
        self._time += seconds

async def publish(
    self,
    topic: str,
    payload: str,
    *,
    retain: bool = False,
    qos: int = 1,
) -> None:
    """Record a publish call, or raise if ``raise_on_publish`` is set."""
    if self.raise_on_publish is not None:
        raise self.raise_on_publish
    self.published.append((topic, payload, retain, qos))

async def subscribe(self, topic: str) -> None:
    """Record a subscribe call."""
    self.subscriptions.append(topic)

def on_message(self, callback: MessageCallback) -> None:
    """Register an inbound-message callback."""
    self._callbacks.append(callback)

async def deliver(self, topic: str, payload: str) -> None:
    """Simulate an inbound message by invoking all callbacks."""
    for cb in self._callbacks:
        await cb(topic, payload)

def reset(self) -> None:
    """Clear all recorded data, callbacks, and failure injection."""
    self.published.clear()
    self.subscriptions.clear()
    self._callbacks.clear()
    self.raise_on_publish = None

def get_messages_for(
    self,
    topic: str,
) -> list[tuple[str, bool, int]]:
    """Return ``(payload, retain, qos)`` tuples for *topic*."""
    return [
        (payload, retain, qos)
        for t, payload, retain, qos in self.published
        if t == topic
    ]

async def publish(
    self,
    topic: str,
    payload: str,  # noqa: ARG002
    *,
    retain: bool = False,  # noqa: ARG002
    qos: int = 1,  # noqa: ARG002
) -> None:
    """Silently discard a publish request."""
    logger.debug("NullMqttClient.publish(%s) — discarded", topic)

async def subscribe(self, topic: str) -> None:
    """Silently discard a subscribe request."""
    logger.debug("NullMqttClient.subscribe(%s) — discarded", topic)

def load(self, key: str) -> dict[str, object] | None:
    """Return a deep copy of the stored dict, or ``None``."""
    value = self._data.get(key)
    if value is None:
        return None
    return copy.deepcopy(value)

def save(self, key: str, data: dict[str, object]) -> None:
    """Store a deep copy of *data*."""
    self._data[key] = copy.deepcopy(data)