Test Utils

Test utilities for hassette apps.

Tier 1 APIs (AppTestHarness, RecordingApi, make_test_config, event factories) are stable and documented for end users.

Tier 2 symbols (HassetteHarness, SimpleTestServer, fixtures, web helpers, etc.) are re-exported here for hassette's own internal test suite. WebSocket stubs (build_fake_ws) come from hassette.test_utils.ws_mocks. They are not in __all__ and may change without notice.

ApiCall dataclass

Record of a single API method invocation.

Write methods (call_service, set_state, fire_event) record their positional arguments in both args and kwargs so that :meth:RecordingApi.assert_called can use kwargs-only matching uniformly::

recorder.assert_called("turn_on", entity_id="light.kitchen")

args is available for direct positional inspection when needed, but assert_called does not check it — use kwargs for assertions.

Attributes:

Name	Type	Description
method	str	Name of the method that was called (e.g. "turn_on").
args	tuple[Any, ...]	Positional arguments passed to the method (for inspection only).
kwargs	dict[str, Any]	Keyword arguments — the primary assertion surface. Write methods include positional args here as well for uniform kwargs-based matching.

Source code in src/hassette/test_utils/api_call.py

@dataclass
class ApiCall:
    """Record of a single API method invocation.

    Write methods (``call_service``, ``set_state``, ``fire_event``) record their
    positional arguments in both ``args`` and ``kwargs`` so that
    :meth:`RecordingApi.assert_called` can use kwargs-only matching uniformly::

        recorder.assert_called("turn_on", entity_id="light.kitchen")

    ``args`` is available for direct positional inspection when needed, but
    ``assert_called`` does not check it — use ``kwargs`` for assertions.

    Attributes:
        method: Name of the method that was called (e.g. "turn_on").
        args: Positional arguments passed to the method (for inspection only).
        kwargs: Keyword arguments — the primary assertion surface. Write methods
            include positional args here as well for uniform kwargs-based matching.
    """

    method: str
    args: tuple[Any, ...] = field(default_factory=tuple)
    kwargs: dict[str, Any] = field(default_factory=dict)

AppConfigurationError

Bases: Exception

Raised when the config dict fails validation against the app's AppConfig subclass.

Attributes:

Name	Type	Description
app_cls	type[App]	The App class whose config failed validation.
original_error	ValidationError	The underlying pydantic ValidationError.

Source code in src/hassette/test_utils/app_harness.py

class AppConfigurationError(Exception):
    """Raised when the config dict fails validation against the app's AppConfig subclass.

    Attributes:
        app_cls: The App class whose config failed validation.
        original_error: The underlying pydantic ValidationError.
    """

    app_cls: type[App]
    original_error: pydantic.ValidationError

    def __init__(self, app_cls: type[App], original_error: pydantic.ValidationError) -> None:
        self.app_cls = app_cls
        self.original_error = original_error
        count = original_error.error_count()
        errors = original_error.errors()
        # Build a compact summary of the first error
        first = errors[0] if errors else {}
        field = ".".join(str(loc) for loc in first.get("loc", ())) or "<unknown>"
        msg_detail = first.get("msg", "")
        summary = f"{count} validation error{'s' if count != 1 else ''} — field '{field}': {msg_detail}"
        super().__init__(f"AppConfigurationError for {app_cls.__name__}: {summary}")

AppTestHarness

Bases: SimulationMixin, TimeControlMixin, Generic[AppType]

Async context manager that wires an App class into Hassette test infrastructure.

Provides a fully initialized app instance with access to its bus, scheduler, api_recorder, and states. Handles teardown in the correct LIFO order via AsyncExitStack.

Usage::

async with AppTestHarness(MyApp, config={"my_setting": "value"}) as harness:
    harness.app      # MyApp instance
    harness.bus      # test Bus
    harness.scheduler  # test Scheduler
    harness.api_recorder  # RecordingApi — records calls your app makes
    harness.states   # StateManager

Note

Holds no shared mutable state. Each harness synthesizes its own manifest and passes it to the app constructor, so concurrent harnesses for the same App class are independent — safe for sequential tests, xdist workers, and concurrent use via asyncio.gather.

Source code in src/hassette/test_utils/app_harness.py

class AppTestHarness(SimulationMixin, TimeControlMixin, Generic[AppType]):
    """Async context manager that wires an App class into Hassette test infrastructure.

    Provides a fully initialized app instance with access to its bus, scheduler,
    api_recorder, and states. Handles teardown in the correct LIFO order via
    AsyncExitStack.

    Usage::

        async with AppTestHarness(MyApp, config={"my_setting": "value"}) as harness:
            harness.app      # MyApp instance
            harness.bus      # test Bus
            harness.scheduler  # test Scheduler
            harness.api_recorder  # RecordingApi — records calls your app makes
            harness.states   # StateManager

    Note:
        Holds no shared mutable state. Each harness synthesizes its own manifest
        and passes it to the app constructor, so concurrent harnesses for the same
        App class are independent — safe for sequential tests, xdist workers, and
        concurrent use via asyncio.gather.
    """

    def __init__(
        self,
        app_cls: type[AppType],
        config: Mapping[str, Any] | None = None,
        *,
        tmp_path: Path | None = None,
    ) -> None:
        """Store args. No resource allocation.

        Args:
            app_cls: The App subclass to instantiate and test.
            config: Config values to validate against app_cls.app_config_cls. Defaults
                to an empty mapping for apps without required settings.
            tmp_path: Optional directory for Hassette data. Auto-created and cleaned
                up if not provided.
        """
        self._app_cls = app_cls
        self._config_dict = dict(config or {})
        self._tmp_path = tmp_path

        # Set during __aenter__
        self._exit_stack: AsyncExitStack | None = None
        self._harness: HassetteHarness | None = None
        self._app: AppType | None = None

        # Time control (set by freeze_time)
        self._test_clock = None
        self._time_patcher: list[object] | None = None
        self._time_patcher_registered: bool = False
        self._freeze_time_lock_held: bool = False

    async def __aenter__(self) -> "AppTestHarness[AppType]":
        """Set up the full harness in 11 steps with LIFO teardown via AsyncExitStack."""
        exit_stack = AsyncExitStack()
        self._exit_stack = exit_stack

        try:
            await self._setup(exit_stack)
        except Exception:
            await exit_stack.aclose()
            self._exit_stack = None
            raise

        return self

    async def _setup(self, exit_stack: AsyncExitStack) -> None:
        """Execute all setup steps, registering teardown callbacks as we go."""
        # Step 1: Resolve data directory
        if self._tmp_path is not None:
            data_dir = self._tmp_path
        else:
            data_dir = Path(tempfile.mkdtemp(prefix="hassette_test_"))
            exit_stack.callback(self._cleanup_tmpdir, data_dir)

        # Step 2: Create minimal HassetteConfig
        hassette_config = make_test_config(data_dir=data_dir)

        # Step 3: Resolve app config class (read-only, safe outside lock).
        app_config_cls = get_app_config_class(self._app_cls)

        # Step 4: Create HassetteHarness (skip_global_set=True — we handle ContextVar below)
        harness = (
            HassetteHarness(
                hassette_config,
                skip_global_set=True,
            )
            .with_bus()
            .with_scheduler()
            .with_state_proxy()
            .with_state_registry()
        )
        self._harness = harness

        # Step 5: Pre-configure hassette.api mock before state proxy starts.
        # HassetteHarness.start() checks "if not self.hassette.api" before setting it,
        # so we set it here first with get_states_raw returning [] to prevent
        # StateProxy.load_cache() from failing when on_initialize() runs.
        api_mock = AsyncMock()
        api_mock.sync = AsyncMock()
        api_mock.get_states_raw = AsyncMock(return_value=[])
        harness.hassette._api = api_mock

        # Step 6: Start harness — registers stop() as teardown (early registration = late unwind)
        await harness.start()
        exit_stack.push_async_callback(harness.stop)

        # Step 7: Set global hassette ContextVar — use context.use() so cleanup is always
        # registered unconditionally. set_global_hassette() returns None when the same
        # instance is already set (e.g., nested harnesses), which would silently skip token
        # cleanup and leave the next test with a stale ContextVar value. context.use()
        # always calls var.set() and registers var.reset(token) on exit, regardless of
        # whether the value was already present.
        exit_stack.enter_context(
            context.use(context.HASSETTE_INSTANCE, cast("Hassette", harness.hassette))  # pyright: ignore[reportArgumentType]
        )

        # Step 8: Mark state proxy ready
        mark_ready(harness.state_proxy, reason="AppTestHarness: mark ready for test")

        # Step 9: Synthesize a per-instance manifest and validate the config.
        # synthesize_manifest is a pure function of the class. make_hermetic_config writes
        # a shared closure cell, but writes and reads it with no await in between, so
        # concurrent harnesses cannot interleave there. Neither path mutates the App class,
        # so the per-instance manifest needs no locking — each harness owns its own.
        manifest = synthesize_manifest(self._app_cls)
        validated_config = make_hermetic_config(self._app_cls, app_config_cls, self._config_dict)

        # Step 10: Instantiate the app with RecordingApi injected via constructor.
        # The manifest is passed per instance, so its app_key is authoritative for
        # the test and no class-level attribute is touched.
        app = self._app_cls(
            hassette=harness.hassette,  # pyright: ignore[reportArgumentType]
            app_config=validated_config,
            index=0,
            app_key=manifest.app_key,
            app_manifest=manifest,
            api_factory=RecordingApi,
        )

        # Add app as child of hassette mock
        harness.hassette.children.append(app)
        app.parent = harness.hassette  # pyright: ignore[reportAttributeAccessIssue]

        self._app = app

        # Step 11: Register app shutdown first (late registration = early unwind)
        # This ensures app shuts down before harness.stop() runs.
        # Wrapped to prevent shutdown exceptions from masking the original test failure.
        exit_stack.push_async_callback(self._safe_app_shutdown, app)

        # Start the app lifecycle
        start(app)
        await wait_for(
            lambda: app.status == ResourceStatus.RUNNING,
            desc=f"{app.class_name} RUNNING",
            timeout=Timeouts.WAIT_FOR_READY,
        )

    @staticmethod
    async def _safe_app_shutdown(app: App) -> None:
        """Shut down the app, logging but not re-raising exceptions.

        Prevents a crashing ``app.shutdown()`` from masking the original test
        failure during ``AsyncExitStack`` teardown.
        """
        try:
            await app.shutdown()
        except Exception:
            LOGGER.warning("AppTestHarness: app.shutdown() raised during teardown", exc_info=True)
        try:
            await app.cache.close()
        except Exception:
            LOGGER.warning("AppTestHarness: app.cache.close() raised during teardown", exc_info=True)

    def _cleanup_tmpdir(self, data_dir: Path) -> None:
        """Remove auto-created tmpdir on teardown."""
        try:
            shutil.rmtree(data_dir, ignore_errors=True)
        except Exception as exc:
            LOGGER.warning("Failed to clean up tmpdir %s: %s", data_dir, exc)

    async def __aexit__(self, exc_type: Any, exc: Any, tb: Any) -> None:
        """Delegate teardown to the AsyncExitStack (LIFO order)."""
        if self._exit_stack is not None:
            await self._exit_stack.__aexit__(exc_type, exc, tb)
            self._exit_stack = None

    @property
    def app(self) -> AppType:
        """The fully initialized App instance."""
        if self._app is None:
            raise RuntimeError("AppTestHarness is not active — use 'async with AppTestHarness(...) as harness'")
        return self._app

    @property
    def bus(self) -> Bus:
        """The test Bus owned by the app."""
        return self.app.bus

    @property
    def scheduler(self) -> Scheduler:
        """The test Scheduler owned by the app."""
        return self.app.scheduler

    @property
    def api_recorder(self) -> RecordingApi:
        """The RecordingApi injected into the app (records calls the app makes)."""
        api = self.app.api
        if not isinstance(api, RecordingApi):
            raise RuntimeError(
                f"Expected app.api to be a RecordingApi but got {type(api).__name__}. "
                "Ensure api_factory=RecordingApi was passed at app construction."
            )
        return api

    @property
    def states(self) -> StateManager:
        """The StateManager owned by the app."""
        return self.app.states

    async def set_state(self, entity_id: str, state: str, **attributes: Any) -> None:
        """Seed an entity's state in the StateProxy.

        Uses make_state_dict() internally with a past sentinel timestamp
        (1970-01-01T00:00:00Z). Simulated events sent via ``simulate_state_change``
        bypass ``StateProxy``'s staleness guard entirely (they use
        ``harness.seed_state()``), so the epoch timestamp does not play a protective
        ordering role — it simply marks seeded state as obviously synthetic.

        Call ``set_state`` **before** ``simulate_state_change`` for the same entity.
        Calling it afterward will overwrite the simulated state with the seeded value.

        This is for pre-test setup only and does NOT fire bus events.

        Args:
            entity_id: The entity ID to seed (e.g., "light.kitchen").
            state: The state value (e.g., "on", "off", "25.5").
            **attributes: Entity attribute key/value pairs.
        """
        state_dict = cast(
            "HassStateDict",
            make_state_dict(
                entity_id,
                state,
                dict(attributes),
                EPOCH_TIMESTAMP,
                EPOCH_TIMESTAMP,
            ),
        )
        await self.require_harness().seed_state(entity_id, state_dict)

    def seed_helper(self, record: BaseModel) -> None:
        """Seed a stored helper config for tests that read helper CRUD.

        Domain is derived from the record class. Passing a record of a type
        not registered in RECORD_TYPE_TO_DOMAIN raises ValueError immediately.

        The record is deep-copied before storage, so later mutations of the
        caller's `record` object will not leak into harness state — matching
        the isolation guarantees of ``list_*`` / ``create_*`` / ``update_*``.

        Args:
            record: A helper Record model instance (e.g., InputBooleanRecord).

        Raises:
            ValueError: If the record's type is not a known helper record type,
                or if a record with the same id is already seeded.
        """
        try:
            domain, _deep_copy = RECORD_TYPE_TO_DOMAIN[type(record)]
        except KeyError as exc:
            raise ValueError(
                f"Unknown helper record type: {type(record).__name__}. "
                f"Expected one of: {sorted(t.__name__ for t in RECORD_TYPE_TO_DOMAIN)}"
            ) from exc
        if record.id in self.api_recorder.helpers.helper_definitions[domain]:  # pyright: ignore[reportAttributeAccessIssue]
            raise ValueError(
                f"A {type(record).__name__} with id={record.id!r} is already seeded. "  # pyright: ignore[reportAttributeAccessIssue]
                f"Use a unique id or call harness.api_recorder.reset() first."
            )
        # Deep-copy to isolate the harness store from later caller-side mutations.
        # Shallow copy is insufficient for InputSelectRecord because of options: list[str].
        self.api_recorder.helpers.helper_definitions[domain][record.id] = record.model_copy(  # pyright: ignore[reportAttributeAccessIssue]
            deep=True
        )

    async def set_states(self, states: dict[str, str | tuple[str, dict]]) -> None:
        """Seed multiple entities at once.

        Example::

            await harness.set_states({
                "light.kitchen": "on",
                "sensor.temp": ("25.5", {"unit_of_measurement": "°C"}),
            })

        Args:
            states: Dict mapping entity_id to state string or (state, attrs) tuple.
        """
        for entity_id, value in states.items():
            if isinstance(value, tuple):
                state, attrs = value
                await self.set_state(entity_id, state, **attrs)
            else:
                await self.set_state(entity_id, value)

app: AppType property

The fully initialized App instance.

bus: Bus property

The test Bus owned by the app.

scheduler: Scheduler property

The test Scheduler owned by the app.

api_recorder: RecordingApi property

The RecordingApi injected into the app (records calls the app makes).

states: StateManager property

The StateManager owned by the app.

__init__(app_cls: type[AppType], config: Mapping[str, Any] | None = None, *, tmp_path: Path | None = None) -> None

Store args. No resource allocation.

Parameters:

Name	Type	Description	Default
app_cls	type[AppType]	The App subclass to instantiate and test.	required
config	Mapping[str, Any] | None	Config values to validate against app_cls.app_config_cls. Defaults to an empty mapping for apps without required settings.	None
tmp_path	Path | None	Optional directory for Hassette data. Auto-created and cleaned up if not provided.	None

Source code in src/hassette/test_utils/app_harness.py

def __init__(
    self,
    app_cls: type[AppType],
    config: Mapping[str, Any] | None = None,
    *,
    tmp_path: Path | None = None,
) -> None:
    """Store args. No resource allocation.

    Args:
        app_cls: The App subclass to instantiate and test.
        config: Config values to validate against app_cls.app_config_cls. Defaults
            to an empty mapping for apps without required settings.
        tmp_path: Optional directory for Hassette data. Auto-created and cleaned
            up if not provided.
    """
    self._app_cls = app_cls
    self._config_dict = dict(config or {})
    self._tmp_path = tmp_path

    # Set during __aenter__
    self._exit_stack: AsyncExitStack | None = None
    self._harness: HassetteHarness | None = None
    self._app: AppType | None = None

    # Time control (set by freeze_time)
    self._test_clock = None
    self._time_patcher: list[object] | None = None
    self._time_patcher_registered: bool = False
    self._freeze_time_lock_held: bool = False

__aenter__() -> AppTestHarness[AppType] async

Set up the full harness in 11 steps with LIFO teardown via AsyncExitStack.

Source code in src/hassette/test_utils/app_harness.py

async def __aenter__(self) -> "AppTestHarness[AppType]":
    """Set up the full harness in 11 steps with LIFO teardown via AsyncExitStack."""
    exit_stack = AsyncExitStack()
    self._exit_stack = exit_stack

    try:
        await self._setup(exit_stack)
    except Exception:
        await exit_stack.aclose()
        self._exit_stack = None
        raise

    return self

__aexit__(exc_type: Any, exc: Any, tb: Any) -> None async

Delegate teardown to the AsyncExitStack (LIFO order).

Source code in src/hassette/test_utils/app_harness.py

async def __aexit__(self, exc_type: Any, exc: Any, tb: Any) -> None:
    """Delegate teardown to the AsyncExitStack (LIFO order)."""
    if self._exit_stack is not None:
        await self._exit_stack.__aexit__(exc_type, exc, tb)
        self._exit_stack = None

set_state(entity_id: str, state: str, **attributes: Any) -> None async

Seed an entity's state in the StateProxy.

Uses make_state_dict() internally with a past sentinel timestamp (1970-01-01T00:00:00Z). Simulated events sent via simulate_state_change bypass StateProxy's staleness guard entirely (they use harness.seed_state()), so the epoch timestamp does not play a protective ordering role — it simply marks seeded state as obviously synthetic.

Call set_state before simulate_state_change for the same entity. Calling it afterward will overwrite the simulated state with the seeded value.

This is for pre-test setup only and does NOT fire bus events.

Parameters:

Name	Type	Description	Default
entity_id	str	The entity ID to seed (e.g., "light.kitchen").	required
state	str	The state value (e.g., "on", "off", "25.5").	required
**attributes	Any	Entity attribute key/value pairs.	{}

Source code in src/hassette/test_utils/app_harness.py

async def set_state(self, entity_id: str, state: str, **attributes: Any) -> None:
    """Seed an entity's state in the StateProxy.

    Uses make_state_dict() internally with a past sentinel timestamp
    (1970-01-01T00:00:00Z). Simulated events sent via ``simulate_state_change``
    bypass ``StateProxy``'s staleness guard entirely (they use
    ``harness.seed_state()``), so the epoch timestamp does not play a protective
    ordering role — it simply marks seeded state as obviously synthetic.

    Call ``set_state`` **before** ``simulate_state_change`` for the same entity.
    Calling it afterward will overwrite the simulated state with the seeded value.

    This is for pre-test setup only and does NOT fire bus events.

    Args:
        entity_id: The entity ID to seed (e.g., "light.kitchen").
        state: The state value (e.g., "on", "off", "25.5").
        **attributes: Entity attribute key/value pairs.
    """
    state_dict = cast(
        "HassStateDict",
        make_state_dict(
            entity_id,
            state,
            dict(attributes),
            EPOCH_TIMESTAMP,
            EPOCH_TIMESTAMP,
        ),
    )
    await self.require_harness().seed_state(entity_id, state_dict)

seed_helper(record: BaseModel) -> None

Seed a stored helper config for tests that read helper CRUD.

Domain is derived from the record class. Passing a record of a type not registered in RECORD_TYPE_TO_DOMAIN raises ValueError immediately.

The record is deep-copied before storage, so later mutations of the caller's record object will not leak into harness state — matching the isolation guarantees of list_* / create_* / update_*.

Parameters:

Name	Type	Description	Default
record	BaseModel	A helper Record model instance (e.g., InputBooleanRecord).	required

Raises:

Type	Description
ValueError	If the record's type is not a known helper record type, or if a record with the same id is already seeded.

Source code in src/hassette/test_utils/app_harness.py

def seed_helper(self, record: BaseModel) -> None:
    """Seed a stored helper config for tests that read helper CRUD.

    Domain is derived from the record class. Passing a record of a type
    not registered in RECORD_TYPE_TO_DOMAIN raises ValueError immediately.

    The record is deep-copied before storage, so later mutations of the
    caller's `record` object will not leak into harness state — matching
    the isolation guarantees of ``list_*`` / ``create_*`` / ``update_*``.

    Args:
        record: A helper Record model instance (e.g., InputBooleanRecord).

    Raises:
        ValueError: If the record's type is not a known helper record type,
            or if a record with the same id is already seeded.
    """
    try:
        domain, _deep_copy = RECORD_TYPE_TO_DOMAIN[type(record)]
    except KeyError as exc:
        raise ValueError(
            f"Unknown helper record type: {type(record).__name__}. "
            f"Expected one of: {sorted(t.__name__ for t in RECORD_TYPE_TO_DOMAIN)}"
        ) from exc
    if record.id in self.api_recorder.helpers.helper_definitions[domain]:  # pyright: ignore[reportAttributeAccessIssue]
        raise ValueError(
            f"A {type(record).__name__} with id={record.id!r} is already seeded. "  # pyright: ignore[reportAttributeAccessIssue]
            f"Use a unique id or call harness.api_recorder.reset() first."
        )
    # Deep-copy to isolate the harness store from later caller-side mutations.
    # Shallow copy is insufficient for InputSelectRecord because of options: list[str].
    self.api_recorder.helpers.helper_definitions[domain][record.id] = record.model_copy(  # pyright: ignore[reportAttributeAccessIssue]
        deep=True
    )

set_states(states: dict[str, str | tuple[str, dict]]) -> None async

Seed multiple entities at once.

Example::

await harness.set_states({
    "light.kitchen": "on",
    "sensor.temp": ("25.5", {"unit_of_measurement": "°C"}),
})

Parameters:

Name	Type	Description	Default
states	dict[str, str | tuple[str, dict]]	Dict mapping entity_id to state string or (state, attrs) tuple.	required

Source code in src/hassette/test_utils/app_harness.py

async def set_states(self, states: dict[str, str | tuple[str, dict]]) -> None:
    """Seed multiple entities at once.

    Example::

        await harness.set_states({
            "light.kitchen": "on",
            "sensor.temp": ("25.5", {"unit_of_measurement": "°C"}),
        })

    Args:
        states: Dict mapping entity_id to state string or (state, attrs) tuple.
    """
    for entity_id, value in states.items():
        if isinstance(value, tuple):
            state, attrs = value
            await self.set_state(entity_id, state, **attrs)
        else:
            await self.set_state(entity_id, value)

DrainError

Bases: DrainFailure

Raised when AppTestHarness drain surfaces handler task exceptions.

Aggregates all non-cancellation exceptions from completed tasks during drain so test failures report the real cause instead of silently masking handler crashes with misleading assertion failures.

Attributes:

Name	Type	Description
task_exceptions	list[tuple[str, BaseException]]	List of (task_name, exception) tuples collected from completed handler tasks during the drain pass.

Source code in src/hassette/test_utils/exceptions.py

class DrainError(DrainFailure):
    """Raised when AppTestHarness drain surfaces handler task exceptions.

    Aggregates all non-cancellation exceptions from completed tasks during drain
    so test failures report the real cause instead of silently masking handler
    crashes with misleading assertion failures.

    Attributes:
        task_exceptions: List of ``(task_name, exception)`` tuples collected
            from completed handler tasks during the drain pass.
    """

    task_exceptions: list[tuple[str, BaseException]]

    def __init__(self, task_exceptions: list[tuple[str, BaseException]]) -> None:
        if not task_exceptions:
            raise ValueError(
                "DrainError requires at least one (task_name, exception) tuple. "
                "Callers must guard with `if collected_exceptions:` before raising."
            )
        self.task_exceptions = task_exceptions
        count = len(task_exceptions)
        first_name, first_exc = task_exceptions[0]
        parts = [
            f"{count} handler task exception{'s' if count != 1 else ''} during drain.",
            f"First: {first_name}: {type(first_exc).__name__}: {first_exc}",
        ]
        if count > 1:
            parts.append(f"({count - 1} more — see .task_exceptions)")
        super().__init__(" ".join(parts))

DrainFailure

Bases: Exception

Base class for all AppTestHarness drain failures.

Lets callers catch both handler exceptions and drain deadline timeouts uniformly with except DrainFailure:. Do not raise this class directly — raise one of its subclasses (:class:DrainError or :class:DrainTimeout).

Note

The Failure suffix is intentional and deviates from the project's *Error-suffix convention for exceptions. It signals that this class is a hierarchy root, not something to raise directly. The two concrete subclasses below use the conventional Error / Timeout suffixes.

Source code in src/hassette/test_utils/exceptions.py

class DrainFailure(Exception):  # noqa: N818  # base class; concrete subclasses use the Error/Timeout suffix
    """Base class for all AppTestHarness drain failures.

    Lets callers catch both handler exceptions and drain deadline timeouts
    uniformly with ``except DrainFailure:``. Do not raise this class directly —
    raise one of its subclasses (:class:`DrainError` or :class:`DrainTimeout`).

    Note:
        The ``Failure`` suffix is intentional and deviates from the project's
        ``*Error``-suffix convention for exceptions. It signals that this
        class is a hierarchy root, not something to raise directly. The two
        concrete subclasses below use the conventional ``Error`` / ``Timeout``
        suffixes.
    """

DrainTimeout

Bases: DrainFailure

Raised when AppTestHarness drain does not reach quiescence within its deadline.

Carries a diagnostic message built by _raise_drain_timeout that includes pending task counts, pending task names, and — when applicable — a hint about debounce windows.

Does NOT inherit from :class:TimeoutError. Callers that previously caught TimeoutError around drain calls should catch DrainTimeout (or the broader DrainFailure) instead.

Source code in src/hassette/test_utils/exceptions.py

class DrainTimeout(DrainFailure):
    """Raised when AppTestHarness drain does not reach quiescence within its deadline.

    Carries a diagnostic message built by ``_raise_drain_timeout`` that
    includes pending task counts, pending task names, and — when applicable —
    a hint about debounce windows.

    Does NOT inherit from :class:`TimeoutError`. Callers that previously
    caught ``TimeoutError`` around drain calls should catch ``DrainTimeout``
    (or the broader ``DrainFailure``) instead.
    """

RecordingApi

Bases: Resource

Test double for hassette.api.Api.

Records write-method calls for assertion in tests. Delegates read methods to StateProxy so tests see seeded state values. get_state() raises EntityNotFoundError for unseeded entities (matching real Api behavior).

on_initialize() calls mark_ready(self) — required for the Resource lifecycle.

sync attribute is a RecordingSyncFacade instance. Write calls via api.sync.* are recorded to the same calls list as the async side. Read methods delegate to the StateProxy. Methods not covered by the facade raise NotImplementedError.

Unstubbed methods raise NotImplementedError with guidance on alternatives.

Authoring constraints (enforced by the RecordingSyncFacade generator):

1. Methods must not call other async def methods on self directly; use sync helpers (_get_raw_state, _convert_state) instead. Violating this constraint will fail the generator with a clear error pointing at the offending call site.

2. Stub methods — those that should raise NotImplementedError on the sync side rather than be body-copied into the facade — should use self.not_implemented(name) for the canonical helpful error message on the async side. The RecordingSyncFacade generator detects stub-tier methods by recognizing any body that contains only docstrings, raise statements, and/or not_implemented() calls, so raise NotImplementedError(...) works too, but self.not_implemented(name) is preferred because the helper returns an exception with the project's standard seed-state guidance.

Example::

async with AppTestHarness(MotionLights, config={}) as harness:
    await harness.simulate_state_change("sensor.test", old_value="off", new_value="on")
    harness.api_recorder.assert_called("turn_on", entity_id="light.kitchen")

Source code in src/hassette/test_utils/recording_api.py

class RecordingApi(Resource):
    """Test double for hassette.api.Api.

    Records write-method calls for assertion in tests. Delegates read methods to
    StateProxy so tests see seeded state values. get_state() raises
    EntityNotFoundError for unseeded entities (matching real Api behavior).

    on_initialize() calls mark_ready(self) — required for the Resource lifecycle.

    sync attribute is a RecordingSyncFacade instance. Write calls via api.sync.*
    are recorded to the same `calls` list as the async side. Read methods delegate
    to the StateProxy. Methods not covered by the facade raise NotImplementedError.

    Unstubbed methods raise NotImplementedError with guidance on alternatives.

    Authoring constraints (enforced by the ``RecordingSyncFacade`` generator):

    1. Methods must not call other ``async def`` methods on ``self`` directly;
       use sync helpers (``_get_raw_state``, ``_convert_state``) instead.
       Violating this constraint will fail the generator with a clear error
       pointing at the offending call site.

    2. Stub methods — those that should raise ``NotImplementedError`` on the
       sync side rather than be body-copied into the facade — should use
       ``self.not_implemented(name)`` for the canonical helpful error message
       on the async side. The ``RecordingSyncFacade`` generator detects
       stub-tier methods by recognizing any body that contains only
       docstrings, ``raise`` statements, and/or ``not_implemented()`` calls,
       so ``raise NotImplementedError(...)`` works too, but
       ``self.not_implemented(name)`` is preferred because the helper returns
       an exception with the project's standard seed-state guidance.

    Example::

        async with AppTestHarness(MotionLights, config={}) as harness:
            await harness.simulate_state_change("sensor.test", old_value="off", new_value="on")
            harness.api_recorder.assert_called("turn_on", entity_id="light.kitchen")
    """

    calls: list[ApiCall]
    # Access via `harness.api_recorder.helpers`; do not import the type directly.
    helpers: "RecordingHelperClient"
    # Access via `harness.api_recorder.sync`; do not import the type directly.
    sync: "RecordingSyncFacade"

    # Methods whose __getattr__ message should redirect users to get_state()
    _STATE_CONVERSION_METHODS: ClassVar[frozenset[str]] = frozenset(
        {
            "get_state_value",
            "get_attribute",
        }
    )

    def __init__(
        self,
        hassette: "Hassette",
        *,
        state_proxy: "StateProxy | None" = None,
        parent: Resource | None = None,
    ) -> None:
        super().__init__(hassette, parent=parent)
        # state_proxy may be injected directly (e.g. in unit tests) or resolved
        # lazily from hassette._state_proxy (when created via App.add_child()).
        self._state_proxy_override = state_proxy
        self.calls = []
        self.helpers = RecordingHelperClient(self)
        self.sync = RecordingSyncFacade(self)

    @property
    def _state_proxy(self) -> "StateProxy":
        """Resolve the state proxy: injected override takes precedence, else hassette._state_proxy."""
        if self._state_proxy_override is not None:
            return self._state_proxy_override
        sp = self.hassette._state_proxy
        if sp is None:
            raise RuntimeError(
                "RecordingApi: no StateProxy available. Ensure HassetteHarness is started with with_state_proxy()."
            )
        return sp

    async def on_initialize(self) -> None:
        """Mark this resource ready. Called by Resource.initialize()."""
        mark_ready(self, reason="RecordingApi initialized")

    # Signatures must exactly match hassette.api.Api.

    async def turn_on(self, entity_id: str | StrEnum, domain: str | None = None, **data: Any) -> None:
        """Record a turn_on call directly under its own method name."""
        entity_id = str(entity_id)
        if domain is None:
            domain = entity_id.split(".", 1)[0]
        self._record_call(
            ApiCall(
                method="turn_on",
                args=(entity_id,),
                kwargs={"entity_id": entity_id, "domain": domain, **data},
            )
        )

    async def turn_off(self, entity_id: str | StrEnum, domain: str | None = None, **data: Any) -> None:
        """Record a turn_off call directly under its own method name."""
        entity_id = str(entity_id)
        if domain is None:
            domain = entity_id.split(".", 1)[0]
        self._record_call(
            ApiCall(
                method="turn_off",
                args=(entity_id,),
                kwargs={"entity_id": entity_id, "domain": domain, **data},
            )
        )

    async def toggle(self, entity_id: str | StrEnum, domain: str | None = None, **data: Any) -> None:
        """Record a toggle call directly under its own method name."""
        entity_id = str(entity_id)
        if domain is None:
            domain = entity_id.split(".", 1)[0]
        self._record_call(
            ApiCall(
                method="toggle",
                args=(entity_id,),
                kwargs={"entity_id": entity_id, "domain": domain, **data},
            )
        )

    async def call_service(
        self,
        domain: str,
        service: str,
        target: dict[str, str] | dict[str, list[str]] | None = None,
        return_response: bool | None = False,
        **data: Any,
    ) -> ServiceResponse | None:
        """Record a call_service call. Returns stub ServiceResponse when return_response=True."""
        self._record_call(
            ApiCall(
                method="call_service",
                args=(domain, service),
                kwargs={
                    "domain": domain,
                    "service": service,
                    "target": copy.deepcopy(target),
                    "return_response": return_response,
                    **data,
                },
            )
        )
        if return_response:
            return ServiceResponse(context=Context(id=None, parent_id=None, user_id=None))
        return None

    async def set_state(
        self,
        entity_id: str | StrEnum,
        state: Any,
        attributes: dict[str, Any] | None = None,
    ) -> dict:
        """Record a set_state call. Returns an empty dict stub."""
        entity_id = str(entity_id)
        self._record_call(
            ApiCall(
                method="set_state",
                args=(entity_id, state),
                kwargs={
                    "entity_id": entity_id,
                    "state": state,
                    "attributes": copy.deepcopy(attributes),
                },
            )
        )
        return {}

    async def fire_event(
        self,
        event_type: str,
        event_data: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        """Record a fire_event call. Returns an empty dict stub."""
        self._record_call(
            ApiCall(
                method="fire_event",
                args=(event_type,),
                kwargs={"event_type": event_type, "event_data": copy.deepcopy(event_data)},
            )
        )
        return {}

    def _get_raw_state(self, entity_id: str) -> "HassStateDict":
        """Look up raw state dict from the proxy, raising EntityNotFoundError if absent."""
        raw = self._state_proxy.states.get(entity_id)
        if raw is None:
            raise EntityNotFoundError(f"Entity '{entity_id}' not found in StateProxy (not seeded).")
        return raw

    def _convert_state(self, raw: "HassStateDict", entity_id: str | None = None) -> BaseState:
        """Convert a raw HassStateDict to a typed BaseState via the state registry.

        Args:
            raw: Raw state dict from the StateProxy.
            entity_id: Optional entity ID passed to the state registry for accurate domain
                resolution. Matches the behaviour of the real Api and StateManager.
        """
        return self.hassette.state_registry.try_convert_state(raw, entity_id)

    async def get_state(self, entity_id: str) -> BaseState:
        """Return the typed state for entity_id. Raises EntityNotFoundError if not seeded."""
        raw = self._get_raw_state(entity_id)
        return self._convert_state(raw, entity_id)

    async def get_states(self) -> list[BaseState]:
        """Return typed states for all seeded entities."""
        # Snapshot the dict to avoid RuntimeError from concurrent mutation.
        items = list(self._state_proxy.states.items())
        return [self._convert_state(raw, eid) for eid, raw in items]

    async def get_entity(self, entity_id: str, model: type[BaseEntity]) -> BaseEntity:
        """Return a pydantic-validated entity wrapper for entity_id.

        Matches the real ``Api.get_entity`` signature exactly — ``model`` is required
        and must be a :class:`~hassette.models.entities.base.BaseEntity` subclass.
        Callers that want registry-converted state without a specific entity model
        should call :meth:`get_state` instead.

        Raises:
            TypeError: If ``model`` is not a ``BaseEntity`` subclass.
            EntityNotFoundError: If ``entity_id`` is not seeded.
        """
        if not issubclass(model, BaseEntity):  # runtime check — mirrors Api.get_entity
            raise TypeError(f"Model {model!r} is not a valid BaseEntity subclass")

        raw = self._get_raw_state(entity_id)
        return model.model_validate({"state": raw})

    async def get_entity_or_none(self, entity_id: str, model: type[BaseEntity]) -> BaseEntity | None:
        """Return a pydantic-validated entity wrapper for entity_id, or None if not seeded.

        Inlines the logic from :meth:`get_entity` using sync helpers only — no peer
        ``async def`` calls on ``self`` — to satisfy the authoring constraint required
        by the ``RecordingSyncFacade`` generator. Matches the real
        ``Api.get_entity_or_none`` signature; see :meth:`get_entity` for semantics.
        """
        if not issubclass(model, BaseEntity):  # runtime check — mirrors Api.get_entity
            raise TypeError(f"Model {model!r} is not a valid BaseEntity subclass")

        try:
            raw = self._get_raw_state(entity_id)
        except EntityNotFoundError:
            return None
        return model.model_validate({"state": raw})

    async def entity_exists(self, entity_id: str) -> bool:
        """Return True if entity_id is seeded in the StateProxy."""
        return entity_id in self._state_proxy.states

    async def get_state_or_none(self, entity_id: str) -> BaseState | None:
        """Return the typed state for entity_id, or None if not seeded.

        Inlines the logic from :meth:`get_state` using sync helpers only — no peer
        ``async def`` calls on ``self`` — to satisfy the authoring constraint required
        by the ``RecordingSyncFacade`` generator.
        """
        try:
            raw = self._get_raw_state(entity_id)
        except EntityNotFoundError:
            return None
        return self._convert_state(raw, entity_id)

    async def get_state_raw(self, entity_id: str) -> dict:
        """Not implemented — raises NotImplementedError."""
        not_implemented("get_state_raw")

    async def get_states_raw(self) -> list[dict]:
        """Not implemented — raises NotImplementedError."""
        not_implemented("get_states_raw")

    async def get_history(self, entity_id: str, *args: Any, **kwargs: Any) -> list:
        """Not implemented — raises NotImplementedError."""
        not_implemented("get_history")

    async def render_template(self, template: str, variables: dict | None = None) -> str:
        """Not implemented — raises NotImplementedError."""
        not_implemented("render_template")

    async def ws_send_and_wait(self, **data: Any) -> Any:
        """Not implemented — raises NotImplementedError."""
        not_implemented("ws_send_and_wait")

    async def ws_send_json(self, **data: Any) -> None:
        """Not implemented — raises NotImplementedError."""
        not_implemented("ws_send_json")

    async def rest_request(self, method: str, url: str, **kwargs: Any) -> Any:
        """Not implemented — raises NotImplementedError."""
        not_implemented("rest_request")

    async def delete_entity(self, entity_id: str) -> None:
        """Not implemented — raises NotImplementedError."""
        not_implemented("delete_entity")

    @classmethod
    def _validate_recorded_method(cls, method: str) -> None:
        """Reject method names that this test double can never record."""
        if method in RECORDED_API_METHODS:
            return

        suggestions = get_close_matches(method, RECORDED_API_METHODS, n=3, cutoff=0.5)
        hint = f" Did you mean {', '.join(repr(name) for name in suggestions)}?" if suggestions else ""
        raise ValueError(f"Unknown recorded API method {method!r}.{hint}")

    @staticmethod
    def _resolve_helper_method_fields(method: str) -> set[str]:
        """Resolve valid assertion fields for helper CRUD method names like 'create_input_boolean'."""
        lookup: dict[str, type] = {}
        for params_type, record_type in CREATE_PARAMS_TO_RECORD_TYPE.items():
            domain, _ = RECORD_TYPE_TO_DOMAIN[record_type]
            lookup[f"create_{domain}"] = params_type
        for params_type, record_type in UPDATE_PARAMS_TO_RECORD_TYPE.items():
            domain, _ = RECORD_TYPE_TO_DOMAIN[record_type]
            lookup[f"update_{domain}"] = params_type
        for domain, _ in RECORD_TYPE_TO_DOMAIN.values():
            lookup[f"delete_{domain}"] = type(None)

        params_type = lookup.get(method)
        if params_type is None or params_type is type(None):
            if method.startswith("delete_"):
                return {"helper_id"}
            return set()
        fields = set(params_type.model_fields)
        if method.startswith("update_"):
            fields.add("helper_id")
        return fields

    @classmethod
    def _validate_expected_kwargs(cls, method: str, expected: dict[str, Any]) -> None:
        """Reject unknown assertion keys when the recording signature is closed."""
        target = getattr(cls, method, None)
        if target is not None:
            parameters = inspect.signature(target).parameters.values()
            if any(parameter.kind is inspect.Parameter.VAR_KEYWORD for parameter in parameters):
                return

            valid_names: set[str] = set()
            for parameter in parameters:
                if parameter.name == "self":
                    continue
                annotation = parameter.annotation
                if isinstance(annotation, type) and issubclass(annotation, BaseModel):
                    valid_names.update(annotation.model_fields)
                else:
                    valid_names.add(parameter.name)
        else:
            valid_names = cls._resolve_helper_method_fields(method)
            if not valid_names:
                return

        unknown_names = expected.keys() - valid_names
        if not unknown_names:
            return

        hints = []
        for name in sorted(unknown_names):
            suggestions = get_close_matches(name, valid_names, n=1, cutoff=0.5)
            hint = f" (did you mean {suggestions[0]!r}?)" if suggestions else ""
            hints.append(f"{name!r}{hint}")
        raise ValueError(f"Unknown assertion keyword(s) for {method!r}: {', '.join(hints)}")

    def _record_call(self, call: ApiCall) -> None:
        """Record a call after checking it against the assertion contract."""
        self._validate_recorded_method(call.method)
        self.calls.append(call)

    @staticmethod
    def _describe_closest_call(calls: list[ApiCall], expected: dict[str, Any], *, exact: bool = False) -> str:
        """Describe how the closest recorded call differs from expected kwargs."""
        candidates: list[tuple[int, int, dict[str, Any], dict[str, Any], dict[str, Any]]] = []
        for index, call in enumerate(calls, start=1):
            missing = {key: value for key, value in expected.items() if key not in call.kwargs}
            mismatched = {
                key: {"expected": value, "actual": call.kwargs[key]}
                for key, value in expected.items()
                if key in call.kwargs and call.kwargs[key] != value
            }
            extra = {key: value for key, value in call.kwargs.items() if exact and key not in expected}
            distance = len(missing) + len(mismatched) + len(extra)
            candidates.append((distance, index, missing, mismatched, extra))

        _, index, missing, mismatched, extra = min(candidates, key=lambda candidate: candidate[0])
        differences = [f"missing={missing!r}", f"mismatched={mismatched!r}"]
        if exact:
            differences.append(f"extra={extra!r}")
        return f"Closest call #{index}: {'; '.join(differences)}"

    def __getattr__(self, name: str) -> Any:
        """Raise NotImplementedError for public attributes not defined on RecordingApi.

        Private/dunder attributes fall through to the default AttributeError so that
        Resource internals (e.g. ``_unique_name``) and Python machinery work correctly.

        State-conversion methods (get_state_value, get_attribute)
        get a tailored message directing users to ``await self.api.get_state(entity_id)``.
        All other unimplemented methods get the generic "Seed state" guidance.
        """
        if name.startswith("_"):
            raise AttributeError(name)
        if name in self._STATE_CONVERSION_METHODS:
            raise NotImplementedError(
                f"RecordingApi.{name} is not implemented. "
                f"Call `await self.api.get_state(entity_id)` and read the returned state directly."
            )
        raise NotImplementedError(
            f"RecordingApi.{name}() is not implemented. "
            "Seed state via AppTestHarness.set_state() for read methods, "
            "or use a full integration test for methods requiring a live HA connection."
        )

    def get_calls(self, method: str | None = None) -> list[ApiCall]:
        """Return all recorded calls, optionally filtered by method name.

        Args:
            method: If given, return only calls for this method name.

        Returns:
            List of ApiCall records (a copy — callers may modify safely).
        """
        if method is None:
            return list(self.calls)
        return [c for c in self.calls if c.method == method]

    def assert_called(self, method: str, **kwargs: Any) -> None:
        """Assert that method was called at least once with matching kwargs.

        Performs **partial** (subset) matching: the call passes if all specified
        ``kwargs`` are present in the recorded call's kwargs with matching values.
        Extra kwargs in the recorded call are ignored. Positional arguments
        recorded in ``call.args`` are also checked via the recorded ``kwargs``
        dict — write methods record their positional args as both ``args`` and
        ``kwargs`` so assertions like
        ``assert_called("turn_on", entity_id="light.kitchen")`` work.

        This is a partial-match alias. See also :meth:`assert_called_partial`
        (identical semantics, explicit name) and :meth:`assert_called_exact`
        (no extra kwargs allowed in the recorded call).

        Args:
            method: Method name to check.
            **kwargs: Expected keyword arguments that must appear in at least one call.

        Raises:
            AssertionError: If no call matches.
        """
        self._validate_recorded_method(method)
        self._validate_expected_kwargs(method, kwargs)
        matching = self.get_calls(method)
        if not matching:
            raise AssertionError(f"Expected '{method}' to have been called, but it was never called.")

        if kwargs:
            for call in matching:
                if all(k in call.kwargs and call.kwargs[k] == v for k, v in kwargs.items()):
                    return
            closest = self._describe_closest_call(matching, kwargs)
            raise AssertionError(
                f"'{method}' was called {len(matching)} time(s), but none matched kwargs {kwargs!r}. "
                f"{closest}. Calls recorded: {[{'args': c.args, 'kwargs': c.kwargs} for c in matching]}"
            )

    def assert_called_partial(self, method: str, **kwargs: Any) -> None:
        """Assert that method was called at least once with matching kwargs (partial match).

        Non-deprecated alias for :meth:`assert_called`. Performs **partial**
        (subset) matching: the call passes if all specified ``kwargs`` are
        present in the recorded call's kwargs with matching values. Extra kwargs
        in the recorded call are ignored.

        Use this name when you want to make the partial-match intent explicit in
        test code. Both ``assert_called`` and ``assert_called_partial`` behave
        identically; they differ only in name clarity.

        See also :meth:`assert_called_exact` for exact (no-extra-kwargs) matching.

        Args:
            method: Method name to check.
            **kwargs: Expected keyword arguments that must appear in at least one call.

        Raises:
            AssertionError: If no call matches.
        """
        self.assert_called(method, **kwargs)

    def assert_called_exact(self, method: str, **kwargs: Any) -> None:
        """Assert that method was called at least once with exactly the specified kwargs.

        Performs **exact** matching: the call passes only when the recorded
        call's ``kwargs`` dict is exactly equal to the provided ``kwargs`` —
        no extra keys are allowed. This is stricter than :meth:`assert_called`
        and :meth:`assert_called_partial`, which allow extra keys in the
        recorded call.

        Use this when you need to verify that no unexpected kwargs were passed.
        For example, if a method should be called *only* with ``entity_id``
        and nothing else, use ``assert_called_exact("turn_off", entity_id="light.x")``
        rather than ``assert_called("turn_off", entity_id="light.x")`` — the latter
        would pass even if ``domain="light"`` was also recorded.

        Args:
            method: Method name to check.
            **kwargs: The exact keyword arguments expected in at least one call.

        Raises:
            AssertionError: If no call was recorded with exactly the specified kwargs.

        Example::

            await api.turn_off("light.x")
            # Passes — recorded kwargs are {"entity_id": "light.x", "domain": "light"}
            api.assert_called("turn_off", entity_id="light.x")       # partial: OK
            # Fails — extra "domain" key is present
            api.assert_called_exact("turn_off", entity_id="light.x") # exact: fails
            # Passes — matches exactly
            api.assert_called_exact("turn_off", entity_id="light.x", domain="light")
        """
        self._validate_recorded_method(method)
        self._validate_expected_kwargs(method, kwargs)
        matching = self.get_calls(method)
        if not matching:
            raise AssertionError(f"Expected '{method}' to have been called, but it was never called.")

        for call in matching:
            if call.kwargs == kwargs:
                return
        closest = self._describe_closest_call(matching, kwargs, exact=True)
        raise AssertionError(
            f"'{method}' was called {len(matching)} time(s), but none matched kwargs exactly {kwargs!r}. "
            f"{closest}. Calls recorded: {[{'args': c.args, 'kwargs': c.kwargs} for c in matching]}"
        )

    def assert_not_called(self, method: str, **kwargs: Any) -> None:
        """Assert that method was never called.

        Args:
            method: Method name to check.
            **kwargs: If provided, only calls whose recorded kwargs match all of these
                key/value pairs count as a violation (partial match, consistent with
                assert_called). This lets you assert "turn_on was never called for
                light.bedroom" even when turn_on was called for other entities.

        Raises:
            AssertionError: If a matching call was recorded.
        """
        self._validate_recorded_method(method)
        self._validate_expected_kwargs(method, kwargs)
        matching = self.get_calls(method)
        if kwargs:
            matching = [c for c in matching if all(k in c.kwargs and c.kwargs[k] == v for k, v in kwargs.items())]
            if matching:
                raise AssertionError(
                    f"Expected '{method}' not to have been called with kwargs {kwargs!r}, "
                    f"but it was called {len(matching)} matching time(s). "
                    f"Matching calls: {[{'args': c.args, 'kwargs': c.kwargs} for c in matching]}"
                )
            return
        if matching:
            raise AssertionError(
                f"Expected '{method}' not to have been called, but it was called {len(matching)} time(s)."
            )

    def assert_call_count(self, method: str, count: int, **kwargs: Any) -> None:
        """Assert that method was called exactly count times.

        Args:
            method: Method name to check.
            count: Expected number of calls (positional). With kwargs, only calls
                matching all the given keyword arguments are counted.
            **kwargs: If provided, only calls whose recorded kwargs match all of these
                key/value pairs are counted toward ``count`` (partial match, consistent
                with assert_called).

        Raises:
            AssertionError: If the call count does not match.
        """
        self._validate_recorded_method(method)
        self._validate_expected_kwargs(method, kwargs)
        if count < 0:
            raise ValueError(f"assert_call_count() count must be non-negative, got {count}.")

        matching = self.get_calls(method)
        if kwargs:
            matching = [c for c in matching if all(k in c.kwargs and c.kwargs[k] == v for k, v in kwargs.items())]
        actual = len(matching)
        if actual != count:
            if kwargs:
                raise AssertionError(
                    f"Expected '{method}' to have been called {count} time(s) with kwargs {kwargs!r}, "
                    f"but it was called {actual} matching time(s)."
                )
            raise AssertionError(
                f"Expected '{method}' to have been called {count} time(s), but it was called {actual} time(s)."
            )

    def reset(self) -> None:
        """Clear all recorded calls and reset self.helpers to empty-per-domain state.

        Replaces the calls list with a new empty list rather than mutating the
        existing list in place. This preserves any snapshots callers hold
        (e.g., ``saved = api.calls`` before a ``simulate_*`` call) — they
        will still see the original calls after reset, as expected.

        Replaces ``self.helpers`` with a fresh ``RecordingHelperClient`` rather than
        mutating its ``helper_definitions`` dict in place, for the same snapshot-
        preservation reason.
        """
        self.calls = []
        self.helpers = RecordingHelperClient(self)

on_initialize() -> None async

Mark this resource ready. Called by Resource.initialize().

Source code in src/hassette/test_utils/recording_api.py

async def on_initialize(self) -> None:
    """Mark this resource ready. Called by Resource.initialize()."""
    mark_ready(self, reason="RecordingApi initialized")

turn_on(entity_id: str | StrEnum, domain: str | None = None, **data: Any) -> None async

Record a turn_on call directly under its own method name.

Source code in src/hassette/test_utils/recording_api.py

async def turn_on(self, entity_id: str | StrEnum, domain: str | None = None, **data: Any) -> None:
    """Record a turn_on call directly under its own method name."""
    entity_id = str(entity_id)
    if domain is None:
        domain = entity_id.split(".", 1)[0]
    self._record_call(
        ApiCall(
            method="turn_on",
            args=(entity_id,),
            kwargs={"entity_id": entity_id, "domain": domain, **data},
        )
    )

turn_off(entity_id: str | StrEnum, domain: str | None = None, **data: Any) -> None async

Record a turn_off call directly under its own method name.

Source code in src/hassette/test_utils/recording_api.py

async def turn_off(self, entity_id: str | StrEnum, domain: str | None = None, **data: Any) -> None:
    """Record a turn_off call directly under its own method name."""
    entity_id = str(entity_id)
    if domain is None:
        domain = entity_id.split(".", 1)[0]
    self._record_call(
        ApiCall(
            method="turn_off",
            args=(entity_id,),
            kwargs={"entity_id": entity_id, "domain": domain, **data},
        )
    )

toggle(entity_id: str | StrEnum, domain: str | None = None, **data: Any) -> None async

Record a toggle call directly under its own method name.

Source code in src/hassette/test_utils/recording_api.py

async def toggle(self, entity_id: str | StrEnum, domain: str | None = None, **data: Any) -> None:
    """Record a toggle call directly under its own method name."""
    entity_id = str(entity_id)
    if domain is None:
        domain = entity_id.split(".", 1)[0]
    self._record_call(
        ApiCall(
            method="toggle",
            args=(entity_id,),
            kwargs={"entity_id": entity_id, "domain": domain, **data},
        )
    )

call_service(domain: str, service: str, target: dict[str, str] | dict[str, list[str]] | None = None, return_response: bool | None = False, **data: Any) -> ServiceResponse | None async

Record a call_service call. Returns stub ServiceResponse when return_response=True.

Source code in src/hassette/test_utils/recording_api.py

async def call_service(
    self,
    domain: str,
    service: str,
    target: dict[str, str] | dict[str, list[str]] | None = None,
    return_response: bool | None = False,
    **data: Any,
) -> ServiceResponse | None:
    """Record a call_service call. Returns stub ServiceResponse when return_response=True."""
    self._record_call(
        ApiCall(
            method="call_service",
            args=(domain, service),
            kwargs={
                "domain": domain,
                "service": service,
                "target": copy.deepcopy(target),
                "return_response": return_response,
                **data,
            },
        )
    )
    if return_response:
        return ServiceResponse(context=Context(id=None, parent_id=None, user_id=None))
    return None

set_state(entity_id: str | StrEnum, state: Any, attributes: dict[str, Any] | None = None) -> dict async

Record a set_state call. Returns an empty dict stub.

Source code in src/hassette/test_utils/recording_api.py

async def set_state(
    self,
    entity_id: str | StrEnum,
    state: Any,
    attributes: dict[str, Any] | None = None,
) -> dict:
    """Record a set_state call. Returns an empty dict stub."""
    entity_id = str(entity_id)
    self._record_call(
        ApiCall(
            method="set_state",
            args=(entity_id, state),
            kwargs={
                "entity_id": entity_id,
                "state": state,
                "attributes": copy.deepcopy(attributes),
            },
        )
    )
    return {}

fire_event(event_type: str, event_data: dict[str, Any] | None = None) -> dict[str, Any] async

Record a fire_event call. Returns an empty dict stub.

Source code in src/hassette/test_utils/recording_api.py

async def fire_event(
    self,
    event_type: str,
    event_data: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Record a fire_event call. Returns an empty dict stub."""
    self._record_call(
        ApiCall(
            method="fire_event",
            args=(event_type,),
            kwargs={"event_type": event_type, "event_data": copy.deepcopy(event_data)},
        )
    )
    return {}

get_state(entity_id: str) -> BaseState async

Return the typed state for entity_id. Raises EntityNotFoundError if not seeded.

Source code in src/hassette/test_utils/recording_api.py

async def get_state(self, entity_id: str) -> BaseState:
    """Return the typed state for entity_id. Raises EntityNotFoundError if not seeded."""
    raw = self._get_raw_state(entity_id)
    return self._convert_state(raw, entity_id)

get_states() -> list[BaseState] async

Return typed states for all seeded entities.

Source code in src/hassette/test_utils/recording_api.py

async def get_states(self) -> list[BaseState]:
    """Return typed states for all seeded entities."""
    # Snapshot the dict to avoid RuntimeError from concurrent mutation.
    items = list(self._state_proxy.states.items())
    return [self._convert_state(raw, eid) for eid, raw in items]

get_entity(entity_id: str, model: type[BaseEntity]) -> BaseEntity async

Return a pydantic-validated entity wrapper for entity_id.

Matches the real Api.get_entity signature exactly — model is required and must be a :class:~hassette.models.entities.base.BaseEntity subclass. Callers that want registry-converted state without a specific entity model should call :meth:get_state instead.

Raises:

Type	Description
TypeError	If model is not a BaseEntity subclass.
EntityNotFoundError	If entity_id is not seeded.

Source code in src/hassette/test_utils/recording_api.py

async def get_entity(self, entity_id: str, model: type[BaseEntity]) -> BaseEntity:
    """Return a pydantic-validated entity wrapper for entity_id.

    Matches the real ``Api.get_entity`` signature exactly — ``model`` is required
    and must be a :class:`~hassette.models.entities.base.BaseEntity` subclass.
    Callers that want registry-converted state without a specific entity model
    should call :meth:`get_state` instead.

    Raises:
        TypeError: If ``model`` is not a ``BaseEntity`` subclass.
        EntityNotFoundError: If ``entity_id`` is not seeded.
    """
    if not issubclass(model, BaseEntity):  # runtime check — mirrors Api.get_entity
        raise TypeError(f"Model {model!r} is not a valid BaseEntity subclass")

    raw = self._get_raw_state(entity_id)
    return model.model_validate({"state": raw})

get_entity_or_none(entity_id: str, model: type[BaseEntity]) -> BaseEntity | None async

Return a pydantic-validated entity wrapper for entity_id, or None if not seeded.

Inlines the logic from :meth:get_entity using sync helpers only — no peer async def calls on self — to satisfy the authoring constraint required by the RecordingSyncFacade generator. Matches the real Api.get_entity_or_none signature; see :meth:get_entity for semantics.

Source code in src/hassette/test_utils/recording_api.py

async def get_entity_or_none(self, entity_id: str, model: type[BaseEntity]) -> BaseEntity | None:
    """Return a pydantic-validated entity wrapper for entity_id, or None if not seeded.

    Inlines the logic from :meth:`get_entity` using sync helpers only — no peer
    ``async def`` calls on ``self`` — to satisfy the authoring constraint required
    by the ``RecordingSyncFacade`` generator. Matches the real
    ``Api.get_entity_or_none`` signature; see :meth:`get_entity` for semantics.
    """
    if not issubclass(model, BaseEntity):  # runtime check — mirrors Api.get_entity
        raise TypeError(f"Model {model!r} is not a valid BaseEntity subclass")

    try:
        raw = self._get_raw_state(entity_id)
    except EntityNotFoundError:
        return None
    return model.model_validate({"state": raw})

entity_exists(entity_id: str) -> bool async

Return True if entity_id is seeded in the StateProxy.

Source code in src/hassette/test_utils/recording_api.py

async def entity_exists(self, entity_id: str) -> bool:
    """Return True if entity_id is seeded in the StateProxy."""
    return entity_id in self._state_proxy.states

get_state_or_none(entity_id: str) -> BaseState | None async

Return the typed state for entity_id, or None if not seeded.

Inlines the logic from :meth:get_state using sync helpers only — no peer async def calls on self — to satisfy the authoring constraint required by the RecordingSyncFacade generator.

Source code in src/hassette/test_utils/recording_api.py

async def get_state_or_none(self, entity_id: str) -> BaseState | None:
    """Return the typed state for entity_id, or None if not seeded.

    Inlines the logic from :meth:`get_state` using sync helpers only — no peer
    ``async def`` calls on ``self`` — to satisfy the authoring constraint required
    by the ``RecordingSyncFacade`` generator.
    """
    try:
        raw = self._get_raw_state(entity_id)
    except EntityNotFoundError:
        return None
    return self._convert_state(raw, entity_id)

get_state_raw(entity_id: str) -> dict async

Not implemented — raises NotImplementedError.

Source code in src/hassette/test_utils/recording_api.py

async def get_state_raw(self, entity_id: str) -> dict:
    """Not implemented — raises NotImplementedError."""
    not_implemented("get_state_raw")

get_states_raw() -> list[dict] async

Not implemented — raises NotImplementedError.

Source code in src/hassette/test_utils/recording_api.py

async def get_states_raw(self) -> list[dict]:
    """Not implemented — raises NotImplementedError."""
    not_implemented("get_states_raw")

get_history(entity_id: str, *args: Any, **kwargs: Any) -> list async

Not implemented — raises NotImplementedError.

Source code in src/hassette/test_utils/recording_api.py

async def get_history(self, entity_id: str, *args: Any, **kwargs: Any) -> list:
    """Not implemented — raises NotImplementedError."""
    not_implemented("get_history")

render_template(template: str, variables: dict | None = None) -> str async

Not implemented — raises NotImplementedError.

Source code in src/hassette/test_utils/recording_api.py

async def render_template(self, template: str, variables: dict | None = None) -> str:
    """Not implemented — raises NotImplementedError."""
    not_implemented("render_template")

ws_send_and_wait(**data: Any) -> Any async

Not implemented — raises NotImplementedError.

Source code in src/hassette/test_utils/recording_api.py

async def ws_send_and_wait(self, **data: Any) -> Any:
    """Not implemented — raises NotImplementedError."""
    not_implemented("ws_send_and_wait")

ws_send_json(**data: Any) -> None async

Not implemented — raises NotImplementedError.

Source code in src/hassette/test_utils/recording_api.py

async def ws_send_json(self, **data: Any) -> None:
    """Not implemented — raises NotImplementedError."""
    not_implemented("ws_send_json")

rest_request(method: str, url: str, **kwargs: Any) -> Any async

Not implemented — raises NotImplementedError.

Source code in src/hassette/test_utils/recording_api.py

async def rest_request(self, method: str, url: str, **kwargs: Any) -> Any:
    """Not implemented — raises NotImplementedError."""
    not_implemented("rest_request")

delete_entity(entity_id: str) -> None async

Not implemented — raises NotImplementedError.

Source code in src/hassette/test_utils/recording_api.py

async def delete_entity(self, entity_id: str) -> None:
    """Not implemented — raises NotImplementedError."""
    not_implemented("delete_entity")

__getattr__(name: str) -> Any

Raise NotImplementedError for public attributes not defined on RecordingApi.

Private/dunder attributes fall through to the default AttributeError so that Resource internals (e.g. _unique_name) and Python machinery work correctly.

State-conversion methods (get_state_value, get_attribute) get a tailored message directing users to await self.api.get_state(entity_id). All other unimplemented methods get the generic "Seed state" guidance.

Source code in src/hassette/test_utils/recording_api.py

def __getattr__(self, name: str) -> Any:
    """Raise NotImplementedError for public attributes not defined on RecordingApi.

    Private/dunder attributes fall through to the default AttributeError so that
    Resource internals (e.g. ``_unique_name``) and Python machinery work correctly.

    State-conversion methods (get_state_value, get_attribute)
    get a tailored message directing users to ``await self.api.get_state(entity_id)``.
    All other unimplemented methods get the generic "Seed state" guidance.
    """
    if name.startswith("_"):
        raise AttributeError(name)
    if name in self._STATE_CONVERSION_METHODS:
        raise NotImplementedError(
            f"RecordingApi.{name} is not implemented. "
            f"Call `await self.api.get_state(entity_id)` and read the returned state directly."
        )
    raise NotImplementedError(
        f"RecordingApi.{name}() is not implemented. "
        "Seed state via AppTestHarness.set_state() for read methods, "
        "or use a full integration test for methods requiring a live HA connection."
    )

get_calls(method: str | None = None) -> list[ApiCall]

Return all recorded calls, optionally filtered by method name.

Parameters:

Name	Type	Description	Default
method	str | None	If given, return only calls for this method name.	None

Returns:

Type	Description
list[ApiCall]	List of ApiCall records (a copy — callers may modify safely).

Source code in src/hassette/test_utils/recording_api.py

def get_calls(self, method: str | None = None) -> list[ApiCall]:
    """Return all recorded calls, optionally filtered by method name.

    Args:
        method: If given, return only calls for this method name.

    Returns:
        List of ApiCall records (a copy — callers may modify safely).
    """
    if method is None:
        return list(self.calls)
    return [c for c in self.calls if c.method == method]

assert_called(method: str, **kwargs: Any) -> None

Assert that method was called at least once with matching kwargs.

Performs partial (subset) matching: the call passes if all specified kwargs are present in the recorded call's kwargs with matching values. Extra kwargs in the recorded call are ignored. Positional arguments recorded in call.args are also checked via the recorded kwargs dict — write methods record their positional args as both args and kwargs so assertions like assert_called("turn_on", entity_id="light.kitchen") work.

This is a partial-match alias. See also :meth:assert_called_partial (identical semantics, explicit name) and :meth:assert_called_exact (no extra kwargs allowed in the recorded call).

Parameters:

Name	Type	Description	Default
method	str	Method name to check.	required
**kwargs	Any	Expected keyword arguments that must appear in at least one call.	{}

Raises:

Type	Description
AssertionError	If no call matches.

Source code in src/hassette/test_utils/recording_api.py

def assert_called(self, method: str, **kwargs: Any) -> None:
    """Assert that method was called at least once with matching kwargs.

    Performs **partial** (subset) matching: the call passes if all specified
    ``kwargs`` are present in the recorded call's kwargs with matching values.
    Extra kwargs in the recorded call are ignored. Positional arguments
    recorded in ``call.args`` are also checked via the recorded ``kwargs``
    dict — write methods record their positional args as both ``args`` and
    ``kwargs`` so assertions like
    ``assert_called("turn_on", entity_id="light.kitchen")`` work.

    This is a partial-match alias. See also :meth:`assert_called_partial`
    (identical semantics, explicit name) and :meth:`assert_called_exact`
    (no extra kwargs allowed in the recorded call).

    Args:
        method: Method name to check.
        **kwargs: Expected keyword arguments that must appear in at least one call.

    Raises:
        AssertionError: If no call matches.
    """
    self._validate_recorded_method(method)
    self._validate_expected_kwargs(method, kwargs)
    matching = self.get_calls(method)
    if not matching:
        raise AssertionError(f"Expected '{method}' to have been called, but it was never called.")

    if kwargs:
        for call in matching:
            if all(k in call.kwargs and call.kwargs[k] == v for k, v in kwargs.items()):
                return
        closest = self._describe_closest_call(matching, kwargs)
        raise AssertionError(
            f"'{method}' was called {len(matching)} time(s), but none matched kwargs {kwargs!r}. "
            f"{closest}. Calls recorded: {[{'args': c.args, 'kwargs': c.kwargs} for c in matching]}"
        )

assert_called_partial(method: str, **kwargs: Any) -> None

Assert that method was called at least once with matching kwargs (partial match).

Non-deprecated alias for :meth:assert_called. Performs partial (subset) matching: the call passes if all specified kwargs are present in the recorded call's kwargs with matching values. Extra kwargs in the recorded call are ignored.

Use this name when you want to make the partial-match intent explicit in test code. Both assert_called and assert_called_partial behave identically; they differ only in name clarity.

See also :meth:assert_called_exact for exact (no-extra-kwargs) matching.

Parameters:

Name	Type	Description	Default
method	str	Method name to check.	required
**kwargs	Any	Expected keyword arguments that must appear in at least one call.	{}

Raises:

Type	Description
AssertionError	If no call matches.

Source code in src/hassette/test_utils/recording_api.py

def assert_called_partial(self, method: str, **kwargs: Any) -> None:
    """Assert that method was called at least once with matching kwargs (partial match).

    Non-deprecated alias for :meth:`assert_called`. Performs **partial**
    (subset) matching: the call passes if all specified ``kwargs`` are
    present in the recorded call's kwargs with matching values. Extra kwargs
    in the recorded call are ignored.

    Use this name when you want to make the partial-match intent explicit in
    test code. Both ``assert_called`` and ``assert_called_partial`` behave
    identically; they differ only in name clarity.

    See also :meth:`assert_called_exact` for exact (no-extra-kwargs) matching.

    Args:
        method: Method name to check.
        **kwargs: Expected keyword arguments that must appear in at least one call.

    Raises:
        AssertionError: If no call matches.
    """
    self.assert_called(method, **kwargs)

assert_called_exact(method: str, **kwargs: Any) -> None

Assert that method was called at least once with exactly the specified kwargs.

Performs exact matching: the call passes only when the recorded call's kwargs dict is exactly equal to the provided kwargs — no extra keys are allowed. This is stricter than :meth:assert_called and :meth:assert_called_partial, which allow extra keys in the recorded call.

Use this when you need to verify that no unexpected kwargs were passed. For example, if a method should be called only with entity_id and nothing else, use assert_called_exact("turn_off", entity_id="light.x") rather than assert_called("turn_off", entity_id="light.x") — the latter would pass even if domain="light" was also recorded.

Parameters:

Name	Type	Description	Default
method	str	Method name to check.	required
**kwargs	Any	The exact keyword arguments expected in at least one call.	{}

Raises:

Type	Description
AssertionError	If no call was recorded with exactly the specified kwargs.

Example::

await api.turn_off("light.x")
# Passes — recorded kwargs are {"entity_id": "light.x", "domain": "light"}
api.assert_called("turn_off", entity_id="light.x")       # partial: OK
# Fails — extra "domain" key is present
api.assert_called_exact("turn_off", entity_id="light.x") # exact: fails
# Passes — matches exactly
api.assert_called_exact("turn_off", entity_id="light.x", domain="light")

Source code in src/hassette/test_utils/recording_api.py

def assert_called_exact(self, method: str, **kwargs: Any) -> None:
    """Assert that method was called at least once with exactly the specified kwargs.

    Performs **exact** matching: the call passes only when the recorded
    call's ``kwargs`` dict is exactly equal to the provided ``kwargs`` —
    no extra keys are allowed. This is stricter than :meth:`assert_called`
    and :meth:`assert_called_partial`, which allow extra keys in the
    recorded call.

    Use this when you need to verify that no unexpected kwargs were passed.
    For example, if a method should be called *only* with ``entity_id``
    and nothing else, use ``assert_called_exact("turn_off", entity_id="light.x")``
    rather than ``assert_called("turn_off", entity_id="light.x")`` — the latter
    would pass even if ``domain="light"`` was also recorded.

    Args:
        method: Method name to check.
        **kwargs: The exact keyword arguments expected in at least one call.

    Raises:
        AssertionError: If no call was recorded with exactly the specified kwargs.

    Example::

        await api.turn_off("light.x")
        # Passes — recorded kwargs are {"entity_id": "light.x", "domain": "light"}
        api.assert_called("turn_off", entity_id="light.x")       # partial: OK
        # Fails — extra "domain" key is present
        api.assert_called_exact("turn_off", entity_id="light.x") # exact: fails
        # Passes — matches exactly
        api.assert_called_exact("turn_off", entity_id="light.x", domain="light")
    """
    self._validate_recorded_method(method)
    self._validate_expected_kwargs(method, kwargs)
    matching = self.get_calls(method)
    if not matching:
        raise AssertionError(f"Expected '{method}' to have been called, but it was never called.")

    for call in matching:
        if call.kwargs == kwargs:
            return
    closest = self._describe_closest_call(matching, kwargs, exact=True)
    raise AssertionError(
        f"'{method}' was called {len(matching)} time(s), but none matched kwargs exactly {kwargs!r}. "
        f"{closest}. Calls recorded: {[{'args': c.args, 'kwargs': c.kwargs} for c in matching]}"
    )

assert_not_called(method: str, **kwargs: Any) -> None

Assert that method was never called.

Parameters:

Name	Type	Description	Default
method	str	Method name to check.	required
**kwargs	Any	If provided, only calls whose recorded kwargs match all of these key/value pairs count as a violation (partial match, consistent with assert_called). This lets you assert "turn_on was never called for light.bedroom" even when turn_on was called for other entities.	{}

Raises:

Type	Description
AssertionError	If a matching call was recorded.

Source code in src/hassette/test_utils/recording_api.py

def assert_not_called(self, method: str, **kwargs: Any) -> None:
    """Assert that method was never called.

    Args:
        method: Method name to check.
        **kwargs: If provided, only calls whose recorded kwargs match all of these
            key/value pairs count as a violation (partial match, consistent with
            assert_called). This lets you assert "turn_on was never called for
            light.bedroom" even when turn_on was called for other entities.

    Raises:
        AssertionError: If a matching call was recorded.
    """
    self._validate_recorded_method(method)
    self._validate_expected_kwargs(method, kwargs)
    matching = self.get_calls(method)
    if kwargs:
        matching = [c for c in matching if all(k in c.kwargs and c.kwargs[k] == v for k, v in kwargs.items())]
        if matching:
            raise AssertionError(
                f"Expected '{method}' not to have been called with kwargs {kwargs!r}, "
                f"but it was called {len(matching)} matching time(s). "
                f"Matching calls: {[{'args': c.args, 'kwargs': c.kwargs} for c in matching]}"
            )
        return
    if matching:
        raise AssertionError(
            f"Expected '{method}' not to have been called, but it was called {len(matching)} time(s)."
        )

assert_call_count(method: str, count: int, **kwargs: Any) -> None

Assert that method was called exactly count times.

Parameters:

Name	Type	Description	Default
method	str	Method name to check.	required
count	int	Expected number of calls (positional). With kwargs, only calls matching all the given keyword arguments are counted.	required
**kwargs	Any	If provided, only calls whose recorded kwargs match all of these key/value pairs are counted toward count (partial match, consistent with assert_called).	{}

Raises:

Type	Description
AssertionError	If the call count does not match.

Source code in src/hassette/test_utils/recording_api.py

def assert_call_count(self, method: str, count: int, **kwargs: Any) -> None:
    """Assert that method was called exactly count times.

    Args:
        method: Method name to check.
        count: Expected number of calls (positional). With kwargs, only calls
            matching all the given keyword arguments are counted.
        **kwargs: If provided, only calls whose recorded kwargs match all of these
            key/value pairs are counted toward ``count`` (partial match, consistent
            with assert_called).

    Raises:
        AssertionError: If the call count does not match.
    """
    self._validate_recorded_method(method)
    self._validate_expected_kwargs(method, kwargs)
    if count < 0:
        raise ValueError(f"assert_call_count() count must be non-negative, got {count}.")

    matching = self.get_calls(method)
    if kwargs:
        matching = [c for c in matching if all(k in c.kwargs and c.kwargs[k] == v for k, v in kwargs.items())]
    actual = len(matching)
    if actual != count:
        if kwargs:
            raise AssertionError(
                f"Expected '{method}' to have been called {count} time(s) with kwargs {kwargs!r}, "
                f"but it was called {actual} matching time(s)."
            )
        raise AssertionError(
            f"Expected '{method}' to have been called {count} time(s), but it was called {actual} time(s)."
        )

reset() -> None

Clear all recorded calls and reset self.helpers to empty-per-domain state.

Replaces the calls list with a new empty list rather than mutating the existing list in place. This preserves any snapshots callers hold (e.g., saved = api.calls before a simulate_* call) — they will still see the original calls after reset, as expected.

Replaces self.helpers with a fresh RecordingHelperClient rather than mutating its helper_definitions dict in place, for the same snapshot- preservation reason.

Source code in src/hassette/test_utils/recording_api.py

def reset(self) -> None:
    """Clear all recorded calls and reset self.helpers to empty-per-domain state.

    Replaces the calls list with a new empty list rather than mutating the
    existing list in place. This preserves any snapshots callers hold
    (e.g., ``saved = api.calls`` before a ``simulate_*`` call) — they
    will still see the original calls after reset, as expected.

    Replaces ``self.helpers`` with a fresh ``RecordingHelperClient`` rather than
    mutating its ``helper_definitions`` dict in place, for the same snapshot-
    preservation reason.
    """
    self.calls = []
    self.helpers = RecordingHelperClient(self)

make_test_config(*, data_dir: Path | str, **overrides: Any) -> HassetteConfig

Create a minimal :class:~hassette.config.config.HassetteConfig for testing.

No TOML file, no env file, no CLI args — only the provided overrides are read. All Pydantic validation still runs.

Defaults

• token: "test-token" (stored as SecretStr; read via config.token.get_secret_value())
• base_url: "http://test.invalid:8123" (unreachable by design)
• disable_state_proxy_polling: True
• apps: {"autodetect": False}
• web_api: {"run": False}
• run_app_precheck: False

Overrides are merged on top of these defaults before validation. Nested group overrides can be passed as dicts or model instances::

make_test_config(data_dir=tmp_path, database={"retention_days": 14})
make_test_config(data_dir=tmp_path, database=DatabaseConfig(retention_days=14))

Parameters:

Name	Type	Description	Default
data_dir	Path | str	Directory for Hassette data (caches, etc.). In pytest, pass tmp_path from the built-in tmp_path fixture:: def test_something(tmp_path): config = make_test_config(data_dir=tmp_path)	required
**overrides	Any	Any HassetteConfig field values to override. Nested group fields may be passed as dicts or model instances.	{}

Returns:

Type	Description
HassetteConfig	A validated :class:~hassette.config.config.HassetteConfig instance.

Example::

config = make_test_config(data_dir=tmp_path)
config = make_test_config(data_dir=tmp_path, base_url="http://192.168.1.1:8123")
config = make_test_config(data_dir=tmp_path, database={"retention_days": 14})

Source code in src/hassette/test_utils/config.py

def make_test_config(*, data_dir: Path | str, **overrides: Any) -> HassetteConfig:
    """Create a minimal :class:`~hassette.config.config.HassetteConfig` for testing.

    No TOML file, no env file, no CLI args — only the provided overrides are
    read. All Pydantic validation still runs.

    Defaults:
        - ``token``: ``"test-token"`` (stored as ``SecretStr``; read via
          ``config.token.get_secret_value()``)
        - ``base_url``: ``"http://test.invalid:8123"`` (unreachable by design)
        - ``disable_state_proxy_polling``: ``True``
        - ``apps``: ``{"autodetect": False}``
        - ``web_api``: ``{"run": False}``
        - ``run_app_precheck``: ``False``

    Overrides are merged on top of these defaults before validation. Nested
    group overrides can be passed as dicts or model instances::

        make_test_config(data_dir=tmp_path, database={"retention_days": 14})
        make_test_config(data_dir=tmp_path, database=DatabaseConfig(retention_days=14))

    Args:
        data_dir: Directory for Hassette data (caches, etc.). In pytest, pass
            ``tmp_path`` from the built-in ``tmp_path`` fixture::

                def test_something(tmp_path):
                    config = make_test_config(data_dir=tmp_path)

        **overrides: Any ``HassetteConfig`` field values to override. Nested
            group fields may be passed as dicts or model instances.

    Returns:
        A validated :class:`~hassette.config.config.HassetteConfig` instance.

    Example::

        config = make_test_config(data_dir=tmp_path)
        config = make_test_config(data_dir=tmp_path, base_url="http://192.168.1.1:8123")
        config = make_test_config(data_dir=tmp_path, database={"retention_days": 14})
    """
    defaults: dict[str, Any] = {
        "token": TEST_TOKEN,
        "base_url": TEST_BASE_URL,
        "data_dir": data_dir,
        "disable_state_proxy_polling": True,
        "apps": {"autodetect": False},
        "web_api": {"run": False},
        "run_app_precheck": False,
    }
    merged = {**defaults, **overrides}

    with _config_lock:
        cls, cell = get_hermetic_hassette_config_cls()
        cell[0] = merged
        return cls()

dummy_cache() -> DummyCache

A fresh DummyCache instance for injecting into an App's cache= constructor parameter.

Isolates cache state per test -- no temp directory management, no SQLite files.

Source code in src/hassette/test_utils/fixtures.py

@pytest.fixture
def dummy_cache() -> DummyCache:
    """A fresh `DummyCache` instance for injecting into an App's `cache=` constructor parameter.

    Isolates cache state per test -- no temp directory management, no SQLite files.
    """
    return DummyCache()

create_call_service_event(*, domain: str, service: str, service_data: dict[str, Any] | None = None) -> CallServiceEvent

Create a call service event for testing.

Source code in src/hassette/test_utils/helpers.py

def create_call_service_event(
    *,
    domain: str,
    service: str,
    service_data: dict[str, Any] | None = None,
) -> CallServiceEvent:
    """Create a call service event for testing."""
    event = create_hass_event(
        "call_service",
        {"domain": domain, "service": service, "service_data": service_data or {}},
    )
    assert isinstance(event, CallServiceEvent)
    return event

create_state_change_event(*, entity_id: str, old_value: Any, new_value: Any, old_attrs: dict[str, Any] | None = None, new_attrs: dict[str, Any] | None = None) -> RawStateChangeEvent

Create a state change event for testing.

Pass None for old_value or new_value to simulate entity creation or removal (produces None for that state dict, not {"state": None, ...}).

Source code in src/hassette/test_utils/helpers.py

def create_state_change_event(
    *,
    entity_id: str,
    old_value: Any,
    new_value: Any,
    old_attrs: dict[str, Any] | None = None,
    new_attrs: dict[str, Any] | None = None,
) -> RawStateChangeEvent:
    """Create a state change event for testing.

    Pass ``None`` for ``old_value`` or ``new_value`` to simulate entity creation or removal
    (produces ``None`` for that state dict, not ``{"state": None, ...}``).
    """
    old_state = make_state_dict(entity_id, str(old_value), attributes=old_attrs) if old_value is not None else None
    new_state = make_state_dict(entity_id, str(new_value), attributes=new_attrs) if new_value is not None else None
    event = create_hass_event(
        "state_changed",
        {"entity_id": entity_id, "old_state": old_state, "new_state": new_state},
    )
    assert isinstance(event, RawStateChangeEvent)
    return event

make_light_state_dict(entity_id: str = 'light.kitchen', state: str = 'on', brightness: int | None = None, color_temp: int | None = None, **kwargs: Any) -> dict[str, Any]

Factory for creating light state dictionary.

Parameters:

Name	Type	Description	Default
entity_id	str	The light entity ID	'light.kitchen'
state	str	"on" or "off"	'on'
brightness	int | None	Brightness value 0-255	None
color_temp	int | None	Color temperature in mireds	None
**kwargs	Any	Additional attributes or state dict fields	{}

Returns:

Type	Description
dict[str, Any]	Dictionary matching Home Assistant light state format

Source code in src/hassette/test_utils/helpers.py

def make_light_state_dict(
    entity_id: str = "light.kitchen",
    state: str = "on",
    brightness: int | None = None,
    color_temp: int | None = None,
    **kwargs: Any,
) -> dict[str, Any]:
    """Factory for creating light state dictionary.

    Args:
        entity_id: The light entity ID
        state: "on" or "off"
        brightness: Brightness value 0-255
        color_temp: Color temperature in mireds
        **kwargs: Additional attributes or state dict fields

    Returns:
        Dictionary matching Home Assistant light state format
    """
    attributes: dict[str, Any] = {"friendly_name": entity_id.split(".")[-1].replace("_", " ").title()}
    if brightness is not None:
        attributes["brightness"] = brightness
    if color_temp is not None:
        attributes["color_temp"] = color_temp

    state_kwargs, extra_attrs = split_state_kwargs(kwargs)
    attributes.update(extra_attrs)

    return make_state_dict(entity_id, state, attributes=attributes, **state_kwargs)

make_sensor_state_dict(entity_id: str = 'sensor.temperature', state: str = '25.5', unit_of_measurement: str | None = None, device_class: str | None = None, **kwargs: Any) -> dict[str, Any]

Factory for creating sensor state dictionary.

Parameters:

Name	Type	Description	Default
entity_id	str	The sensor entity ID	'sensor.temperature'
state	str	The sensor value as string	'25.5'
unit_of_measurement	str | None	Unit string (e.g., "°C", "%")	None
device_class	str | None	Device class (e.g., "temperature", "humidity")	None
**kwargs	Any	Additional attributes or state dict fields	{}

Returns:

Type	Description
dict[str, Any]	Dictionary matching Home Assistant sensor state format

Source code in src/hassette/test_utils/helpers.py

def make_sensor_state_dict(
    entity_id: str = "sensor.temperature",
    state: str = "25.5",
    unit_of_measurement: str | None = None,
    device_class: str | None = None,
    **kwargs: Any,
) -> dict[str, Any]:
    """Factory for creating sensor state dictionary.

    Args:
        entity_id: The sensor entity ID
        state: The sensor value as string
        unit_of_measurement: Unit string (e.g., "°C", "%")
        device_class: Device class (e.g., "temperature", "humidity")
        **kwargs: Additional attributes or state dict fields

    Returns:
        Dictionary matching Home Assistant sensor state format
    """
    attributes = {"friendly_name": entity_id.split(".")[-1].replace("_", " ").title()}
    if unit_of_measurement is not None:
        attributes["unit_of_measurement"] = unit_of_measurement
    if device_class is not None:
        attributes["device_class"] = device_class

    state_kwargs, extra_attrs = split_state_kwargs(kwargs)
    attributes.update(extra_attrs)

    return make_state_dict(entity_id, state, attributes=attributes, **state_kwargs)

make_state_dict(entity_id: str, state: str, attributes: dict[str, Any] | None = None, last_changed: str | None = None, last_updated: str | None = None, last_reported: str | None = None, context: dict[str, Any] | None = None) -> dict[str, Any]

Factory for creating state dictionary in Home Assistant format.

Parameters:

Name	Type	Description	Default
entity_id	str	The entity ID (e.g., "light.kitchen")	required
state	str	The state value (e.g., "on", "off", "25.5")	required
attributes	dict[str, Any] | None	Entity attributes dict	None
last_changed	str | None	ISO timestamp string	None
last_updated	str | None	ISO timestamp string	None
last_reported	str | None	ISO timestamp string	None
context	dict[str, Any] | None	Event context dict	None

Returns:

Type	Description
dict[str, Any]	Dictionary matching Home Assistant state format

Source code in src/hassette/test_utils/helpers.py

def make_state_dict(
    entity_id: str,
    state: str,
    attributes: dict[str, Any] | None = None,
    last_changed: str | None = None,
    last_updated: str | None = None,
    last_reported: str | None = None,
    context: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Factory for creating state dictionary in Home Assistant format.

    Args:
        entity_id: The entity ID (e.g., "light.kitchen")
        state: The state value (e.g., "on", "off", "25.5")
        attributes: Entity attributes dict
        last_changed: ISO timestamp string
        last_updated: ISO timestamp string
        last_reported: ISO timestamp string
        context: Event context dict

    Returns:
        Dictionary matching Home Assistant state format
    """
    now = date_utils.now().format_iso()
    result = {
        "entity_id": entity_id,
        "state": state,
        "attributes": attributes or {},
        "last_changed": last_changed or now,
        "last_updated": last_updated or now,
        "context": context or {"id": str(uuid4()), "parent_id": None, "user_id": None},
    }
    if last_reported is not None:
        result["last_reported"] = last_reported
    return result

make_switch_state_dict(entity_id: str = 'switch.outlet', state: str = 'on', **kwargs: Any) -> dict[str, Any]

Factory for creating switch state dictionary.

Parameters:

Name	Type	Description	Default
entity_id	str	The switch entity ID	'switch.outlet'
state	str	"on" or "off"	'on'
**kwargs	Any	Additional attributes or state dict fields	{}

Returns:

Type	Description
dict[str, Any]	Dictionary matching Home Assistant switch state format

Source code in src/hassette/test_utils/helpers.py

def make_switch_state_dict(entity_id: str = "switch.outlet", state: str = "on", **kwargs: Any) -> dict[str, Any]:
    """Factory for creating switch state dictionary.

    Args:
        entity_id: The switch entity ID
        state: "on" or "off"
        **kwargs: Additional attributes or state dict fields

    Returns:
        Dictionary matching Home Assistant switch state format
    """
    attributes = {"friendly_name": entity_id.split(".")[-1].replace("_", " ").title()}

    state_kwargs, extra_attrs = split_state_kwargs(kwargs)
    attributes.update(extra_attrs)

    return make_state_dict(entity_id, state, attributes=attributes, **state_kwargs)

make_typed_state(state_class: type[StateT], state_dict: dict[str, Any]) -> StateT

Convert a raw state dict to a typed state via the conversion pipeline.

Replaces direct XState.model_validate(dict) calls in tests; routes through the conversion entry point so tests exercise the same path as production.

Parameters:

Name	Type	Description	Default
state_class	type[StateT]	The target state model class (e.g., LightState, SensorState).	required
state_dict	dict[str, Any]	A raw state dict as produced by make_state_dict / make_*_state_dict.	required

Returns:

Type	Description
StateT	The typed state instance.

Source code in src/hassette/test_utils/helpers.py

def make_typed_state(state_class: type[StateT], state_dict: "dict[str, Any]") -> StateT:
    """Convert a raw state dict to a typed state via the conversion pipeline.

    Replaces direct ``XState.model_validate(dict)`` calls in tests; routes through
    the conversion entry point so tests exercise the same path as production.

    Args:
        state_class: The target state model class (e.g., LightState, SensorState).
        state_dict: A raw state dict as produced by make_state_dict / make_*_state_dict.

    Returns:
        The typed state instance.
    """
    entity_id: str = state_dict.get("entity_id", "<unknown>")
    result = STATE_REGISTRY.coerce_and_construct(state_class, cast("HassStateDict", state_dict), entity_id)
    assert isinstance(result, state_class)
    return result

make_mock_hassette(*, data_dir: Path | str | None = None, set_ready: bool = True, set_loop: bool = True, sealed: bool = True, **config_overrides: Any) -> AsyncMock

Create a fully-wired :class:unittest.mock.AsyncMock that stands in for Hassette.

The mock combines a real, Pydantic-validated :class:~hassette.config.config.HassetteConfig (via :func:~hassette.test_utils.config.make_test_config) with AsyncMock shells for all non-configuration attributes. This eliminates config drift across test files while keeping unit tests lightweight — no real Hassette __init__ side effects.

After wiring all standard attributes, :func:unittest.mock.seal is applied so that accessing any attribute not explicitly set here raises AttributeError. Tests that need additional attributes beyond the defaults pass sealed=False, set their extras, and optionally seal the mock themselves.

Parameters:

Name	Type	Description	Default
data_dir	Path | str | None	Directory for Hassette data. Defaults to tempfile.mkdtemp() so unit tests don't need tmp_path. Integration tests that need DB isolation should pass tmp_path or tmp_path_factory.mktemp().	None
set_ready	bool	If True (default), calls hassette.ready_event.set() so the mock appears ready immediately.	True
set_loop	bool	If True (default), sets hassette.loop to the running event loop via asyncio.get_running_loop(). Pass False for session-scoped or synchronous fixtures that run outside an async event loop.	True
sealed	bool	If True (default), calls :func:unittest.mock.seal after wiring all attributes. Pass False if the test needs to set additional attributes.	True
**config_overrides	Any	Any :class:~hassette.config.config.HassetteConfig field to override. Merged on top of make_test_config() defaults. Nested group fields may be passed as dicts:: make_mock_hassette(database={"retention_days": 14}) make_mock_hassette(strict_lifecycle=True)	{}

Returns:

Type	Description
AsyncMock	A sealed (by default) :class:~unittest.mock.AsyncMock with:
AsyncMock	.config: real :class:~hassette.config.config.HassetteConfig instance
AsyncMock	.ready_event, .shutdown_event: :class:asyncio.Event instances
AsyncMock	.event_streams_closed: False
AsyncMock	.loop_thread_id: current thread ident
AsyncMock	.loop: running event loop (or None if set_loop=False)
AsyncMock	.scheduler_service.register_removal_callback: :class:~unittest.mock.Mock
AsyncMock	.scheduler_service.deregister_removal_callback: :class:~unittest.mock.Mock
AsyncMock	.bus_service.remove_listeners_by_owner: :class:~unittest.mock.Mock
AsyncMock	.bus_service.get_listeners_by_owner: :class:~unittest.mock.Mock returning []
AsyncMock	.bus_service.register_removal_callback: :class:~unittest.mock.Mock
AsyncMock	.bus_service.deregister_removal_callback: :class:~unittest.mock.Mock
AsyncMock	.app_handler.get: :class:~unittest.mock.Mock returning None (no app running)
AsyncMock	._runtime_query_service: None (wired at runtime by the framework)
AsyncMock	.session_id: None
AsyncMock	.database_service: None
AsyncMock	.wait_for_ready: :class:~unittest.mock.AsyncMock returning True
AsyncMock	.children: []
AsyncMock	._sync_executor_service / .sync_executor_service: None (not wired; tests needing run_in_thread must inject a real SyncExecutorService into both attributes)
AsyncMock	._sync_executor / .sync_executor: None (not wired; tests needing run_in_thread must inject a real SyncExecutor into both attributes — _create_task_bucket reads the public sync_executor attribute)

Example::

async def test_something():
    hassette = make_mock_hassette()
    assert hassette.config.token.get_secret_value() == "test-token"

async def test_strict(tmp_path):
    hassette = make_mock_hassette(data_dir=tmp_path, strict_lifecycle=True)
    assert hassette.config.strict_lifecycle is True

Source code in src/hassette/test_utils/mock_hassette.py

def make_mock_hassette(
    *,
    data_dir: Path | str | None = None,
    set_ready: bool = True,
    set_loop: bool = True,
    sealed: bool = True,
    **config_overrides: Any,
) -> AsyncMock:
    """Create a fully-wired :class:`unittest.mock.AsyncMock` that stands in for Hassette.

    The mock combines a real, Pydantic-validated :class:`~hassette.config.config.HassetteConfig`
    (via :func:`~hassette.test_utils.config.make_test_config`) with ``AsyncMock`` shells for all
    non-configuration attributes. This eliminates config drift across test files while keeping
    unit tests lightweight — no real Hassette ``__init__`` side effects.

    After wiring all standard attributes, :func:`unittest.mock.seal` is applied so that
    accessing any attribute not explicitly set here raises ``AttributeError``. Tests that need
    additional attributes beyond the defaults pass ``sealed=False``, set their extras, and
    optionally seal the mock themselves.

    Args:
        data_dir: Directory for Hassette data. Defaults to ``tempfile.mkdtemp()`` so unit
            tests don't need ``tmp_path``. Integration tests that need DB isolation should
            pass ``tmp_path`` or ``tmp_path_factory.mktemp()``.
        set_ready: If ``True`` (default), calls ``hassette.ready_event.set()`` so the mock
            appears ready immediately.
        set_loop: If ``True`` (default), sets ``hassette.loop`` to the running event loop via
            ``asyncio.get_running_loop()``. Pass ``False`` for session-scoped or synchronous
            fixtures that run outside an async event loop.
        sealed: If ``True`` (default), calls :func:`unittest.mock.seal` after wiring all
            attributes. Pass ``False`` if the test needs to set additional attributes.
        **config_overrides: Any :class:`~hassette.config.config.HassetteConfig` field to
            override. Merged on top of ``make_test_config()`` defaults. Nested group fields
            may be passed as dicts::

                make_mock_hassette(database={"retention_days": 14})
                make_mock_hassette(strict_lifecycle=True)

    Returns:
        A sealed (by default) :class:`~unittest.mock.AsyncMock` with:

        - ``.config``: real :class:`~hassette.config.config.HassetteConfig` instance
        - ``.ready_event``, ``.shutdown_event``: :class:`asyncio.Event` instances
        - ``.event_streams_closed``: ``False``
        - ``.loop_thread_id``: current thread ident
        - ``.loop``: running event loop (or ``None`` if ``set_loop=False``)
        - ``.scheduler_service.register_removal_callback``: :class:`~unittest.mock.Mock`
        - ``.scheduler_service.deregister_removal_callback``: :class:`~unittest.mock.Mock`
        - ``.bus_service.remove_listeners_by_owner``: :class:`~unittest.mock.Mock`
        - ``.bus_service.get_listeners_by_owner``: :class:`~unittest.mock.Mock` returning ``[]``
        - ``.bus_service.register_removal_callback``: :class:`~unittest.mock.Mock`
        - ``.bus_service.deregister_removal_callback``: :class:`~unittest.mock.Mock`
        - ``.app_handler.get``: :class:`~unittest.mock.Mock` returning ``None`` (no app running)
        - ``._runtime_query_service``: ``None`` (wired at runtime by the framework)
        - ``.session_id``: ``None``
        - ``.database_service``: ``None``
        - ``.wait_for_ready``: :class:`~unittest.mock.AsyncMock` returning ``True``
        - ``.children``: ``[]``
        - ``._sync_executor_service`` / ``.sync_executor_service``: ``None`` (not wired;
            tests needing ``run_in_thread`` must inject a real ``SyncExecutorService``
            into both attributes)
        - ``._sync_executor`` / ``.sync_executor``: ``None`` (not wired; tests needing
            ``run_in_thread`` must inject a real ``SyncExecutor`` into both attributes —
            ``_create_task_bucket`` reads the public ``sync_executor`` attribute)

    Example::

        async def test_something():
            hassette = make_mock_hassette()
            assert hassette.config.token.get_secret_value() == "test-token"

        async def test_strict(tmp_path):
            hassette = make_mock_hassette(data_dir=tmp_path, strict_lifecycle=True)
            assert hassette.config.strict_lifecycle is True
    """
    if data_dir is None:
        data_dir = tempfile.mkdtemp()
        atexit.register(shutil.rmtree, data_dir, True)

    config = make_test_config(data_dir=data_dir, **config_overrides)

    hassette = AsyncMock()
    hassette.config = config

    # Readiness / shutdown signals
    ready_event = asyncio.Event()
    if set_ready:
        ready_event.set()
    hassette.ready_event = ready_event
    hassette.shutdown_event = asyncio.Event()

    # Fatal-exit state — matches a real fresh Hassette (no fatal reason recorded yet). Explicit so
    # code that branches on `fatal_shutdown_reason is not None` does not see MagicMock's auto-truthy
    # attribute (e.g. finalize_session persisting a spurious failure status). Set both the property
    # name (read path) and the backing field.
    hassette._fatal_shutdown_reason = None
    hassette.fatal_shutdown_reason = None

    # Event stream state
    hassette.event_streams_closed = False

    # Thread / loop identity — TaskBucket.spawn reads the public loop_thread_id accessor.
    hassette.loop_thread_id = threading.get_ident()
    if set_loop:
        try:
            hassette.loop = asyncio.get_running_loop()
        except RuntimeError:
            hassette.loop = None
    else:
        hassette.loop = None

    # Scheduler service stubs — production reads the public scheduler_service accessor.
    hassette.scheduler_service.register_removal_callback = Mock()
    hassette.scheduler_service.deregister_removal_callback = Mock()

    # Bus service stubs — production reads the public bus_service accessor.
    hassette.bus_service.remove_listeners_by_owner = Mock()
    hassette.bus_service.get_listeners_by_owner = Mock(return_value=[])
    hassette.bus_service.register_removal_callback = Mock()
    hassette.bus_service.deregister_removal_callback = Mock()

    # App handler stubs — get() is synchronous; return None (no app running by default)
    hassette.app_handler.get = Mock(return_value=None)

    # Runtime query service — None by default; set_runtime_query_service() wires it at runtime
    hassette._runtime_query_service = None

    # Database / session (wired by initialized_db after DB setup)
    hassette.session_id = None
    hassette.try_session_id = Mock(return_value=None)
    hassette.database_service = None

    # Async utilities
    hassette.wait_for_ready = AsyncMock(return_value=True)

    # Resource children
    hassette.children = []

    # SyncExecutorService — set both the backing field and the public accessor (same
    # pattern as fatal_shutdown_reason above) so tests that reference
    # `hassette.sync_executor_service`/`hassette._sync_executor_service` directly don't
    # raise AttributeError on sealed mocks. `_create_task_bucket` itself only reads
    # `hassette.sync_executor` (the capability class, wired below) — never this attribute.
    hassette._sync_executor_service = None
    hassette.sync_executor_service = None

    # SyncExecutor — the plain capability class (not the Service wrapper above).
    # _create_task_bucket reads `hassette.sync_executor` unconditionally, so sealed
    # mocks need it stubbed or every TaskBucket construction raises AttributeError.
    hassette._sync_executor = None
    hassette.sync_executor = None

    if sealed:
        seal(hassette)

    return hassette