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 from hassette.test_utils._internal for backward compatibility with hassette's own internal test suite. 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

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

Mutates class-level attributes (app_manifest) with save/restore under a narrow per-class asyncio.Lock. Safe for sequential tests, xdist workers, and concurrent use via asyncio.gather for the same App class.

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

class AppTestHarness(SimulationMixin, TimeControlMixin):
    """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:
        Mutates class-level attributes (app_manifest) with save/restore under a
        narrow per-class asyncio.Lock. Safe for sequential tests, xdist workers,
        and concurrent use via asyncio.gather for the same App class.
    """

    # Class-level sentinel for "not set" — distinguishes None from "attribute absent"
    _UNSET: ClassVar[object] = object()

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

        Args:
            app_cls: The App subclass to instantiate and test.
            config: Dict of config values to validate against app_cls.app_config_cls.
            tmp_path: Optional directory for Hassette data. Auto-created and cleaned
                up if not provided.
        """
        self._app_cls = app_cls
        self._config_dict = config
        self._tmp_path = tmp_path

        # Set during __aenter__
        self._exit_stack: AsyncExitStack | None = None
        self._harness: HassetteHarness | None = None
        self._app: App | 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

    async def __aenter__(self) -> "AppTestHarness":
        """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
        harness.state_proxy.mark_ready(reason="AppTestHarness: mark ready for test")

        # Step 9: Synthesize manifest under narrow per-class lock.
        # The lock serializes both hermetic config validation and the
        # class_manifest_state read-modify-write so concurrent harnesses for
        # the same class share one manifest lifecycle — only the last to exit
        # restores the original.
        async with get_class_lock(self._app_cls):
            validated_config = make_hermetic_config(self._app_cls, app_config_cls, self._config_dict)
            state = class_manifest_state.get(self._app_cls)
            if state is None:
                original_manifest = getattr(self._app_cls, "app_manifest", self._UNSET)
                manifest = synthesize_manifest(self._app_cls)
                self._app_cls.app_manifest = manifest
                class_manifest_state[self._app_cls] = (1, original_manifest)
            else:
                count, original_manifest = state
                class_manifest_state[self._app_cls] = (count + 1, original_manifest)

        exit_stack.push_async_callback(self._restore_manifest)

        # Step 10: Instantiate the app with RecordingApi injected via constructor
        app = self._app_cls(
            hassette=harness.hassette,  # pyright: ignore[reportArgumentType]
            app_config=validated_config,
            index=0,
            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
        app.start()
        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)

    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 e:
            LOGGER.warning("Failed to clean up tmpdir %s: %s", data_dir, e)

    async def _restore_manifest(self) -> None:
        """Decrement the manifest reference count; restore original when count reaches 0."""
        async with get_class_lock(self._app_cls):
            state = class_manifest_state.get(self._app_cls)
            if state is None:
                LOGGER.warning(
                    "_restore_manifest: class_manifest_state entry missing for %s",
                    self._app_cls.__name__,
                )
                return
            count, original = state
            if count > 1:
                class_manifest_state[self._app_cls] = (count - 1, original)
                return
            del class_manifest_state[self._app_cls]
            if original is self._UNSET:
                with contextlib.suppress(AttributeError):
                    del self._app_cls.app_manifest
                if "app_manifest" in self._app_cls.__dict__:
                    LOGGER.warning(
                        "_restore_manifest: could not delete app_manifest from %s.__dict__",
                        self._app_cls.__name__,
                    )
            else:
                self._app_cls.app_manifest = original

    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) -> App:
        """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 e:
            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 e
        if record.id in self.api_recorder.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.helper_definitions[domain][record.id] = record.model_copy(deep=True)  # pyright: ignore[reportAttributeAccessIssue]

    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: App 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[App], config: dict[str, Any], *, tmp_path: Path | None = None) -> None

Store args. No resource allocation.

Parameters:

Name	Type	Description	Default
app_cls	type[App]	The App subclass to instantiate and test.	required
config	dict[str, Any]	Dict of config values to validate against app_cls.app_config_cls.	required
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[App],
    config: dict[str, Any],
    *,
    tmp_path: Path | None = None,
) -> None:
    """Store args. No resource allocation.

    Args:
        app_cls: The App subclass to instantiate and test.
        config: Dict of config values to validate against app_cls.app_config_cls.
        tmp_path: Optional directory for Hassette data. Auto-created and cleaned
            up if not provided.
    """
    self._app_cls = app_cls
    self._config_dict = config
    self._tmp_path = tmp_path

    # Set during __aenter__
    self._exit_stack: AsyncExitStack | None = None
    self._harness: HassetteHarness | None = None
    self._app: App | 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

__aenter__() -> AppTestHarness 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":
    """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 e:
        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 e
    if record.id in self.api_recorder.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.helper_definitions[domain][record.id] = record.model_copy(deep=True)  # pyright: ignore[reportAttributeAccessIssue]

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 self.mark_ready() — 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 self.mark_ready() — 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]
    helper_definitions: dict[str, dict[str, Any]]
    # 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_state_value_typed",
            "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.helper_definitions = {d: {} for d in SUPPORTED_HELPER_DOMAINS}
        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()."""
        self.mark_ready(reason="RecordingApi initialized")

    def _new_helper_id(self, domain: str, name: str) -> str:
        """Generate a unique helper id for domain, mirroring HA's IDManager.generate_id.

        Private sync helper called by create_* methods. The sync facade generator
        rewrites ``self._new_helper_id(...)`` → ``self._parent._new_helper_id(...)``
        so body-copied create methods in RecordingSyncFacade call this correctly.

        Emits a DEBUG log when the returned id was auto-suffixed due to a
        collision — otherwise a test author who expected ``vacation_mode`` but
        got ``vacation_mode_2`` has no log signal explaining why.
        """
        existing_ids = set(self.helper_definitions[domain].keys())
        generated = generate_helper_id(existing_ids, name)
        base_slug = slugify_helper_name(name)
        if generated != base_slug:
            self.logger.debug(
                "RecordingApi %s: name %r -> id %r (base slug %r was already taken; auto-suffixed)",
                domain,
                name,
                generated,
                base_slug,
            )
        return generated

    # Signatures must exactly match hassette.api.Api.

    async def turn_on(self, entity_id: str | StrEnum, domain: str = "homeassistant", **data: Any) -> None:
        """Record a turn_on call directly under its own method name."""
        entity_id = str(entity_id)
        self.calls.append(
            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 = "homeassistant") -> None:
        """Record a turn_off call directly under its own method name."""
        entity_id = str(entity_id)
        self.calls.append(
            ApiCall(
                method="turn_off",
                args=(entity_id,),
                kwargs={"entity_id": entity_id, "domain": domain},
            )
        )

    async def toggle_service(self, entity_id: str | StrEnum, domain: str = "homeassistant") -> None:
        """Record a toggle_service call directly under its own method name."""
        entity_id = str(entity_id)
        self.calls.append(
            ApiCall(
                method="toggle_service",
                args=(entity_id,),
                kwargs={"entity_id": entity_id, "domain": domain},
            )
        )

    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.calls.append(
            ApiCall(
                method="call_service",
                args=(domain, service),
                kwargs={
                    "domain": domain,
                    "service": service,
                    # Deep-copy target at record time so later caller mutations — including
                    # mutations to nested lists like `{"entity_id": [...]}`, which HA entity
                    # targets frequently contain — do not alter the recorded assertion
                    # surface (immutability principle).
                    "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.calls.append(
            ApiCall(
                method="set_state",
                args=(entity_id, state),
                # Deep-copy attributes at record time so later caller mutations —
                # including mutations to nested structures — do not alter the recorded
                # assertion surface (immutability principle).
                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.calls.append(
            ApiCall(
                method="fire_event",
                args=(event_type,),
                # Deep-copy event_data at record time so later caller mutations —
                # including mutations to nested structures — do not alter the recorded
                # assertion surface (immutability principle).
                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")

    # Signatures match hassette.api.Api exactly.
    #
    # Generic core methods (_list_helper, _create_helper, _update_helper,
    # _delete_helper) implement the shared logic dispatching via
    # RECORD_TYPE_TO_DOMAIN. The 32 per-domain methods below are thin typed
    # delegations that call the generic core; no logic lives in them.

    def _list_helper(self, record_type: type) -> list[Any]:
        """Generic list helper — returns shallow or deep copies of stored records.

        Dispatches via ``RECORD_TYPE_TO_DOMAIN`` to determine the domain and
        whether ``deep=True`` copies are needed.

        Args:
            record_type: The Record class (e.g. ``InputBooleanRecord``).

        Returns:
            List of model copies for the domain.
        """
        domain, deep_copy = RECORD_TYPE_TO_DOMAIN[record_type]
        return [r.model_copy(deep=deep_copy) for r in self.helper_definitions[domain].values()]

    def _create_helper(self, record_type: type, method_name: str, params: Any) -> Any:
        """Generic create helper — records an ApiCall and inserts a new record.

        Dispatches via ``RECORD_TYPE_TO_DOMAIN`` to determine the domain and
        copy depth.  Auto-suffixes the generated id on collision (mirrors HA's
        IDManager.generate_id).

        Args:
            record_type: The Record class to instantiate.
            method_name: The API method name to record (e.g. ``"create_input_boolean"``).
            params: The Create*Params model instance.

        Returns:
            A copy of the newly created record.
        """
        domain, deep_copy = RECORD_TYPE_TO_DOMAIN[record_type]
        self.calls.append(
            ApiCall(
                method=method_name,
                args=(),
                kwargs=params.model_dump(exclude_unset=True),
            )
        )
        generated_id = self._new_helper_id(domain, params.name)
        record = record_type(id=generated_id, **params.model_dump(exclude_unset=True))
        self.helper_definitions[domain][record.id] = record
        return record.model_copy(deep=deep_copy)

    def _update_helper(self, record_type: type, method_name: str, helper_id: str, params: Any) -> Any:
        """Generic update helper — records an ApiCall and mutates the stored record.

        Dispatches via ``RECORD_TYPE_TO_DOMAIN`` to determine the domain and
        copy depth.

        Args:
            record_type: The Record class.
            method_name: The API method name to record (e.g. ``"update_input_boolean"``).
            helper_id: The helper id to update.
            params: The Update*Params model instance.

        Returns:
            A copy of the updated record.

        Raises:
            FailedMessageError: With code='not_found' if helper_id is not seeded.
        """
        domain, deep_copy = RECORD_TYPE_TO_DOMAIN[record_type]
        self.calls.append(
            ApiCall(
                method=method_name,
                args=(helper_id,),
                kwargs={"helper_id": helper_id, **params.model_dump(exclude_unset=True)},
            )
        )
        if helper_id not in self.helper_definitions[domain]:
            raise FailedMessageError(
                f"{domain} helper {helper_id!r} not found. Seed it via harness.seed_helper() first.",
                code="not_found",
            )
        existing = self.helper_definitions[domain][helper_id]
        updated = existing.model_copy(update=params.model_dump(exclude_unset=True))
        self.helper_definitions[domain][helper_id] = updated
        return updated.model_copy(deep=deep_copy)

    def _delete_helper(self, record_type: type, method_name: str, helper_id: str) -> None:
        """Generic delete helper — records an ApiCall and removes the stored record.

        Dispatches via ``RECORD_TYPE_TO_DOMAIN`` to determine the domain.

        Args:
            record_type: The Record class.
            method_name: The API method name to record (e.g. ``"delete_input_boolean"``).
            helper_id: The helper id to delete.

        Raises:
            FailedMessageError: With code='not_found' if helper_id is not seeded.
        """
        domain, _deep_copy = RECORD_TYPE_TO_DOMAIN[record_type]
        self.calls.append(
            ApiCall(
                method=method_name,
                args=(helper_id,),
                kwargs={"helper_id": helper_id},
            )
        )
        if helper_id not in self.helper_definitions[domain]:
            raise FailedMessageError(
                f"{domain} helper {helper_id!r} not found.",
                code="not_found",
            )
        del self.helper_definitions[domain][helper_id]

    async def list_input_booleans(self) -> list[InputBooleanRecord]:
        """Return all seeded input_boolean helpers. Delegates to _list_helper."""
        return cast("list[InputBooleanRecord]", self._list_helper(InputBooleanRecord))

    async def create_input_boolean(self, params: CreateInputBooleanParams) -> InputBooleanRecord:
        """Record the call and add a record to helper_definitions. Delegates to _create_helper."""
        return cast("InputBooleanRecord", self._create_helper(InputBooleanRecord, "create_input_boolean", params))

    async def update_input_boolean(self, helper_id: str, params: UpdateInputBooleanParams) -> InputBooleanRecord:
        """Record the call and mutate the seeded record. Delegates to _update_helper."""
        return cast(
            "InputBooleanRecord", self._update_helper(InputBooleanRecord, "update_input_boolean", helper_id, params)
        )

    async def delete_input_boolean(self, helper_id: str) -> None:
        """Record the call and remove the seeded record. Delegates to _delete_helper."""
        self._delete_helper(InputBooleanRecord, "delete_input_boolean", helper_id)

    async def list_input_numbers(self) -> list[InputNumberRecord]:
        """Return all seeded input_number helpers. Delegates to _list_helper."""
        return cast("list[InputNumberRecord]", self._list_helper(InputNumberRecord))

    async def create_input_number(self, params: CreateInputNumberParams) -> InputNumberRecord:
        """Record the call and add a record to helper_definitions. Delegates to _create_helper."""
        return cast("InputNumberRecord", self._create_helper(InputNumberRecord, "create_input_number", params))

    async def update_input_number(self, helper_id: str, params: UpdateInputNumberParams) -> InputNumberRecord:
        """Record the call and mutate the seeded record. Delegates to _update_helper."""
        return cast(
            "InputNumberRecord", self._update_helper(InputNumberRecord, "update_input_number", helper_id, params)
        )

    async def delete_input_number(self, helper_id: str) -> None:
        """Record the call and remove the seeded record. Delegates to _delete_helper."""
        self._delete_helper(InputNumberRecord, "delete_input_number", helper_id)

    async def list_input_texts(self) -> list[InputTextRecord]:
        """Return all seeded input_text helpers. Delegates to _list_helper."""
        return cast("list[InputTextRecord]", self._list_helper(InputTextRecord))

    async def create_input_text(self, params: CreateInputTextParams) -> InputTextRecord:
        """Record the call and add a record to helper_definitions. Delegates to _create_helper."""
        return cast("InputTextRecord", self._create_helper(InputTextRecord, "create_input_text", params))

    async def update_input_text(self, helper_id: str, params: UpdateInputTextParams) -> InputTextRecord:
        """Record the call and mutate the seeded record. Delegates to _update_helper."""
        return cast("InputTextRecord", self._update_helper(InputTextRecord, "update_input_text", helper_id, params))

    async def delete_input_text(self, helper_id: str) -> None:
        """Record the call and remove the seeded record. Delegates to _delete_helper."""
        self._delete_helper(InputTextRecord, "delete_input_text", helper_id)

    async def list_input_selects(self) -> list[InputSelectRecord]:
        """Return all seeded input_select helpers as deep-isolated copies.

        Delegates to _list_helper. Uses ``model_copy(deep=True)`` because
        ``InputSelectRecord.options`` is a ``list[str]`` — the ``deep_copy=True``
        flag in ``RECORD_TYPE_TO_DOMAIN`` ensures the list is not aliased.
        """
        return cast("list[InputSelectRecord]", self._list_helper(InputSelectRecord))

    async def create_input_select(self, params: CreateInputSelectParams) -> InputSelectRecord:
        """Record the call and add a record to helper_definitions. Delegates to _create_helper."""
        return cast("InputSelectRecord", self._create_helper(InputSelectRecord, "create_input_select", params))

    async def update_input_select(self, helper_id: str, params: UpdateInputSelectParams) -> InputSelectRecord:
        """Record the call and mutate the seeded record. Delegates to _update_helper."""
        return cast(
            "InputSelectRecord", self._update_helper(InputSelectRecord, "update_input_select", helper_id, params)
        )

    async def delete_input_select(self, helper_id: str) -> None:
        """Record the call and remove the seeded record. Delegates to _delete_helper."""
        self._delete_helper(InputSelectRecord, "delete_input_select", helper_id)

    async def list_input_datetimes(self) -> list[InputDatetimeRecord]:
        """Return all seeded input_datetime helpers. Delegates to _list_helper."""
        return cast("list[InputDatetimeRecord]", self._list_helper(InputDatetimeRecord))

    async def create_input_datetime(self, params: CreateInputDatetimeParams) -> InputDatetimeRecord:
        """Record the call and add a record to helper_definitions. Delegates to _create_helper."""
        return cast("InputDatetimeRecord", self._create_helper(InputDatetimeRecord, "create_input_datetime", params))

    async def update_input_datetime(self, helper_id: str, params: UpdateInputDatetimeParams) -> InputDatetimeRecord:
        """Record the call and mutate the seeded record. Delegates to _update_helper."""
        return cast(
            "InputDatetimeRecord", self._update_helper(InputDatetimeRecord, "update_input_datetime", helper_id, params)
        )

    async def delete_input_datetime(self, helper_id: str) -> None:
        """Record the call and remove the seeded record. Delegates to _delete_helper."""
        self._delete_helper(InputDatetimeRecord, "delete_input_datetime", helper_id)

    async def list_input_buttons(self) -> list[InputButtonRecord]:
        """Return all seeded input_button helpers. Delegates to _list_helper."""
        return cast("list[InputButtonRecord]", self._list_helper(InputButtonRecord))

    async def create_input_button(self, params: CreateInputButtonParams) -> InputButtonRecord:
        """Record the call and add a record to helper_definitions. Delegates to _create_helper."""
        return cast("InputButtonRecord", self._create_helper(InputButtonRecord, "create_input_button", params))

    async def update_input_button(self, helper_id: str, params: UpdateInputButtonParams) -> InputButtonRecord:
        """Record the call and mutate the seeded record. Delegates to _update_helper."""
        return cast(
            "InputButtonRecord", self._update_helper(InputButtonRecord, "update_input_button", helper_id, params)
        )

    async def delete_input_button(self, helper_id: str) -> None:
        """Record the call and remove the seeded record. Delegates to _delete_helper."""
        self._delete_helper(InputButtonRecord, "delete_input_button", helper_id)

    async def list_counters(self) -> list[CounterRecord]:
        """Return all seeded counter helpers. Delegates to _list_helper."""
        return cast("list[CounterRecord]", self._list_helper(CounterRecord))

    async def create_counter(self, params: CreateCounterParams) -> CounterRecord:
        """Record the call and add a record to helper_definitions. Delegates to _create_helper."""
        return cast("CounterRecord", self._create_helper(CounterRecord, "create_counter", params))

    async def update_counter(self, helper_id: str, params: UpdateCounterParams) -> CounterRecord:
        """Record the call and mutate the seeded record. Delegates to _update_helper."""
        return cast("CounterRecord", self._update_helper(CounterRecord, "update_counter", helper_id, params))

    async def delete_counter(self, helper_id: str) -> None:
        """Record the call and remove the seeded record. Delegates to _delete_helper."""
        self._delete_helper(CounterRecord, "delete_counter", helper_id)

    async def list_timers(self) -> list[TimerRecord]:
        """Return all seeded timer helpers. Delegates to _list_helper."""
        return cast("list[TimerRecord]", self._list_helper(TimerRecord))

    async def create_timer(self, params: CreateTimerParams) -> TimerRecord:
        """Record the call and add a record to helper_definitions. Delegates to _create_helper."""
        return cast("TimerRecord", self._create_helper(TimerRecord, "create_timer", params))

    async def update_timer(self, helper_id: str, params: UpdateTimerParams) -> TimerRecord:
        """Record the call and mutate the seeded record. Delegates to _update_helper."""
        return cast("TimerRecord", self._update_helper(TimerRecord, "update_timer", helper_id, params))

    async def delete_timer(self, helper_id: str) -> None:
        """Record the call and remove the seeded record. Delegates to _delete_helper."""
        self._delete_helper(TimerRecord, "delete_timer", helper_id)

    async def increment_counter(self, entity_id: str) -> None:
        """Record an increment_counter call directly (not via call_service)."""
        self.calls.append(
            ApiCall(
                method="increment_counter",
                args=(entity_id,),
                kwargs={"entity_id": entity_id},
            )
        )

    async def decrement_counter(self, entity_id: str) -> None:
        """Record a decrement_counter call directly (not via call_service)."""
        self.calls.append(
            ApiCall(
                method="decrement_counter",
                args=(entity_id,),
                kwargs={"entity_id": entity_id},
            )
        )

    async def reset_counter(self, entity_id: str) -> None:
        """Record a reset_counter call directly (not via call_service)."""
        self.calls.append(
            ApiCall(
                method="reset_counter",
                args=(entity_id,),
                kwargs={"entity_id": entity_id},
            )
        )

    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_state_value_typed, 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.
        """
        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:
                # Check that all expected kwargs appear in the call's recorded kwargs.
                # Write methods record positional args in both call.args and call.kwargs
                # so kwargs-based assertions work uniformly for all methods.
                if all(k in call.kwargs and call.kwargs[k] == v for k, v in kwargs.items()):
                    return
            raise AssertionError(
                f"'{method}' was called {len(matching)} time(s), but none matched kwargs {kwargs!r}. "
                f"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="homeassistant"`` 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": "homeassistant"}
            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="homeassistant")
        """
        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
        raise AssertionError(
            f"'{method}' was called {len(matching)} time(s), but none matched kwargs exactly {kwargs!r}. "
            f"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.
        """
        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.
        """
        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 helper_definitions 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.
        """
        self.calls = []
        self.helper_definitions = {d: {} for d in SUPPORTED_HELPER_DOMAINS}

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()."""
    self.mark_ready(reason="RecordingApi initialized")

turn_on(entity_id: str | StrEnum, domain: str = 'homeassistant', **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 = "homeassistant", **data: Any) -> None:
    """Record a turn_on call directly under its own method name."""
    entity_id = str(entity_id)
    self.calls.append(
        ApiCall(
            method="turn_on",
            args=(entity_id,),
            kwargs={"entity_id": entity_id, "domain": domain, **data},
        )
    )

turn_off(entity_id: str | StrEnum, domain: str = 'homeassistant') -> 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 = "homeassistant") -> None:
    """Record a turn_off call directly under its own method name."""
    entity_id = str(entity_id)
    self.calls.append(
        ApiCall(
            method="turn_off",
            args=(entity_id,),
            kwargs={"entity_id": entity_id, "domain": domain},
        )
    )

toggle_service(entity_id: str | StrEnum, domain: str = 'homeassistant') -> None async

Record a toggle_service call directly under its own method name.

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

async def toggle_service(self, entity_id: str | StrEnum, domain: str = "homeassistant") -> None:
    """Record a toggle_service call directly under its own method name."""
    entity_id = str(entity_id)
    self.calls.append(
        ApiCall(
            method="toggle_service",
            args=(entity_id,),
            kwargs={"entity_id": entity_id, "domain": domain},
        )
    )

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.calls.append(
        ApiCall(
            method="call_service",
            args=(domain, service),
            kwargs={
                "domain": domain,
                "service": service,
                # Deep-copy target at record time so later caller mutations — including
                # mutations to nested lists like `{"entity_id": [...]}`, which HA entity
                # targets frequently contain — do not alter the recorded assertion
                # surface (immutability principle).
                "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.calls.append(
        ApiCall(
            method="set_state",
            args=(entity_id, state),
            # Deep-copy attributes at record time so later caller mutations —
            # including mutations to nested structures — do not alter the recorded
            # assertion surface (immutability principle).
            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.calls.append(
        ApiCall(
            method="fire_event",
            args=(event_type,),
            # Deep-copy event_data at record time so later caller mutations —
            # including mutations to nested structures — do not alter the recorded
            # assertion surface (immutability principle).
            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")

list_input_booleans() -> list[InputBooleanRecord] async

Return all seeded input_boolean helpers. Delegates to _list_helper.

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

async def list_input_booleans(self) -> list[InputBooleanRecord]:
    """Return all seeded input_boolean helpers. Delegates to _list_helper."""
    return cast("list[InputBooleanRecord]", self._list_helper(InputBooleanRecord))

create_input_boolean(params: CreateInputBooleanParams) -> InputBooleanRecord async

Record the call and add a record to helper_definitions. Delegates to _create_helper.

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

async def create_input_boolean(self, params: CreateInputBooleanParams) -> InputBooleanRecord:
    """Record the call and add a record to helper_definitions. Delegates to _create_helper."""
    return cast("InputBooleanRecord", self._create_helper(InputBooleanRecord, "create_input_boolean", params))

update_input_boolean(helper_id: str, params: UpdateInputBooleanParams) -> InputBooleanRecord async

Record the call and mutate the seeded record. Delegates to _update_helper.

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

async def update_input_boolean(self, helper_id: str, params: UpdateInputBooleanParams) -> InputBooleanRecord:
    """Record the call and mutate the seeded record. Delegates to _update_helper."""
    return cast(
        "InputBooleanRecord", self._update_helper(InputBooleanRecord, "update_input_boolean", helper_id, params)
    )

delete_input_boolean(helper_id: str) -> None async

Record the call and remove the seeded record. Delegates to _delete_helper.

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

async def delete_input_boolean(self, helper_id: str) -> None:
    """Record the call and remove the seeded record. Delegates to _delete_helper."""
    self._delete_helper(InputBooleanRecord, "delete_input_boolean", helper_id)

list_input_numbers() -> list[InputNumberRecord] async

Return all seeded input_number helpers. Delegates to _list_helper.

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

async def list_input_numbers(self) -> list[InputNumberRecord]:
    """Return all seeded input_number helpers. Delegates to _list_helper."""
    return cast("list[InputNumberRecord]", self._list_helper(InputNumberRecord))

create_input_number(params: CreateInputNumberParams) -> InputNumberRecord async

Record the call and add a record to helper_definitions. Delegates to _create_helper.

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

async def create_input_number(self, params: CreateInputNumberParams) -> InputNumberRecord:
    """Record the call and add a record to helper_definitions. Delegates to _create_helper."""
    return cast("InputNumberRecord", self._create_helper(InputNumberRecord, "create_input_number", params))

update_input_number(helper_id: str, params: UpdateInputNumberParams) -> InputNumberRecord async

Record the call and mutate the seeded record. Delegates to _update_helper.

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

async def update_input_number(self, helper_id: str, params: UpdateInputNumberParams) -> InputNumberRecord:
    """Record the call and mutate the seeded record. Delegates to _update_helper."""
    return cast(
        "InputNumberRecord", self._update_helper(InputNumberRecord, "update_input_number", helper_id, params)
    )

delete_input_number(helper_id: str) -> None async

Record the call and remove the seeded record. Delegates to _delete_helper.

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

async def delete_input_number(self, helper_id: str) -> None:
    """Record the call and remove the seeded record. Delegates to _delete_helper."""
    self._delete_helper(InputNumberRecord, "delete_input_number", helper_id)

list_input_texts() -> list[InputTextRecord] async

Return all seeded input_text helpers. Delegates to _list_helper.

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

async def list_input_texts(self) -> list[InputTextRecord]:
    """Return all seeded input_text helpers. Delegates to _list_helper."""
    return cast("list[InputTextRecord]", self._list_helper(InputTextRecord))

create_input_text(params: CreateInputTextParams) -> InputTextRecord async

Record the call and add a record to helper_definitions. Delegates to _create_helper.

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

async def create_input_text(self, params: CreateInputTextParams) -> InputTextRecord:
    """Record the call and add a record to helper_definitions. Delegates to _create_helper."""
    return cast("InputTextRecord", self._create_helper(InputTextRecord, "create_input_text", params))

update_input_text(helper_id: str, params: UpdateInputTextParams) -> InputTextRecord async

Record the call and mutate the seeded record. Delegates to _update_helper.

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

async def update_input_text(self, helper_id: str, params: UpdateInputTextParams) -> InputTextRecord:
    """Record the call and mutate the seeded record. Delegates to _update_helper."""
    return cast("InputTextRecord", self._update_helper(InputTextRecord, "update_input_text", helper_id, params))

delete_input_text(helper_id: str) -> None async

Record the call and remove the seeded record. Delegates to _delete_helper.

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

async def delete_input_text(self, helper_id: str) -> None:
    """Record the call and remove the seeded record. Delegates to _delete_helper."""
    self._delete_helper(InputTextRecord, "delete_input_text", helper_id)

list_input_selects() -> list[InputSelectRecord] async

Return all seeded input_select helpers as deep-isolated copies.

Delegates to _list_helper. Uses model_copy(deep=True) because InputSelectRecord.options is a list[str] — the deep_copy=True flag in RECORD_TYPE_TO_DOMAIN ensures the list is not aliased.

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

async def list_input_selects(self) -> list[InputSelectRecord]:
    """Return all seeded input_select helpers as deep-isolated copies.

    Delegates to _list_helper. Uses ``model_copy(deep=True)`` because
    ``InputSelectRecord.options`` is a ``list[str]`` — the ``deep_copy=True``
    flag in ``RECORD_TYPE_TO_DOMAIN`` ensures the list is not aliased.
    """
    return cast("list[InputSelectRecord]", self._list_helper(InputSelectRecord))

create_input_select(params: CreateInputSelectParams) -> InputSelectRecord async

Record the call and add a record to helper_definitions. Delegates to _create_helper.

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

async def create_input_select(self, params: CreateInputSelectParams) -> InputSelectRecord:
    """Record the call and add a record to helper_definitions. Delegates to _create_helper."""
    return cast("InputSelectRecord", self._create_helper(InputSelectRecord, "create_input_select", params))

update_input_select(helper_id: str, params: UpdateInputSelectParams) -> InputSelectRecord async

Record the call and mutate the seeded record. Delegates to _update_helper.

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

async def update_input_select(self, helper_id: str, params: UpdateInputSelectParams) -> InputSelectRecord:
    """Record the call and mutate the seeded record. Delegates to _update_helper."""
    return cast(
        "InputSelectRecord", self._update_helper(InputSelectRecord, "update_input_select", helper_id, params)
    )

delete_input_select(helper_id: str) -> None async

Record the call and remove the seeded record. Delegates to _delete_helper.

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

async def delete_input_select(self, helper_id: str) -> None:
    """Record the call and remove the seeded record. Delegates to _delete_helper."""
    self._delete_helper(InputSelectRecord, "delete_input_select", helper_id)

list_input_datetimes() -> list[InputDatetimeRecord] async

Return all seeded input_datetime helpers. Delegates to _list_helper.

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

async def list_input_datetimes(self) -> list[InputDatetimeRecord]:
    """Return all seeded input_datetime helpers. Delegates to _list_helper."""
    return cast("list[InputDatetimeRecord]", self._list_helper(InputDatetimeRecord))

create_input_datetime(params: CreateInputDatetimeParams) -> InputDatetimeRecord async

Record the call and add a record to helper_definitions. Delegates to _create_helper.

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

async def create_input_datetime(self, params: CreateInputDatetimeParams) -> InputDatetimeRecord:
    """Record the call and add a record to helper_definitions. Delegates to _create_helper."""
    return cast("InputDatetimeRecord", self._create_helper(InputDatetimeRecord, "create_input_datetime", params))

update_input_datetime(helper_id: str, params: UpdateInputDatetimeParams) -> InputDatetimeRecord async

Record the call and mutate the seeded record. Delegates to _update_helper.

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

async def update_input_datetime(self, helper_id: str, params: UpdateInputDatetimeParams) -> InputDatetimeRecord:
    """Record the call and mutate the seeded record. Delegates to _update_helper."""
    return cast(
        "InputDatetimeRecord", self._update_helper(InputDatetimeRecord, "update_input_datetime", helper_id, params)
    )

delete_input_datetime(helper_id: str) -> None async

Record the call and remove the seeded record. Delegates to _delete_helper.

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

async def delete_input_datetime(self, helper_id: str) -> None:
    """Record the call and remove the seeded record. Delegates to _delete_helper."""
    self._delete_helper(InputDatetimeRecord, "delete_input_datetime", helper_id)

list_input_buttons() -> list[InputButtonRecord] async

Return all seeded input_button helpers. Delegates to _list_helper.

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

async def list_input_buttons(self) -> list[InputButtonRecord]:
    """Return all seeded input_button helpers. Delegates to _list_helper."""
    return cast("list[InputButtonRecord]", self._list_helper(InputButtonRecord))

create_input_button(params: CreateInputButtonParams) -> InputButtonRecord async

Record the call and add a record to helper_definitions. Delegates to _create_helper.

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

async def create_input_button(self, params: CreateInputButtonParams) -> InputButtonRecord:
    """Record the call and add a record to helper_definitions. Delegates to _create_helper."""
    return cast("InputButtonRecord", self._create_helper(InputButtonRecord, "create_input_button", params))

update_input_button(helper_id: str, params: UpdateInputButtonParams) -> InputButtonRecord async

Record the call and mutate the seeded record. Delegates to _update_helper.

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

async def update_input_button(self, helper_id: str, params: UpdateInputButtonParams) -> InputButtonRecord:
    """Record the call and mutate the seeded record. Delegates to _update_helper."""
    return cast(
        "InputButtonRecord", self._update_helper(InputButtonRecord, "update_input_button", helper_id, params)
    )

delete_input_button(helper_id: str) -> None async

Record the call and remove the seeded record. Delegates to _delete_helper.

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

async def delete_input_button(self, helper_id: str) -> None:
    """Record the call and remove the seeded record. Delegates to _delete_helper."""
    self._delete_helper(InputButtonRecord, "delete_input_button", helper_id)

list_counters() -> list[CounterRecord] async

Return all seeded counter helpers. Delegates to _list_helper.

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

async def list_counters(self) -> list[CounterRecord]:
    """Return all seeded counter helpers. Delegates to _list_helper."""
    return cast("list[CounterRecord]", self._list_helper(CounterRecord))

create_counter(params: CreateCounterParams) -> CounterRecord async

Record the call and add a record to helper_definitions. Delegates to _create_helper.

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

async def create_counter(self, params: CreateCounterParams) -> CounterRecord:
    """Record the call and add a record to helper_definitions. Delegates to _create_helper."""
    return cast("CounterRecord", self._create_helper(CounterRecord, "create_counter", params))

update_counter(helper_id: str, params: UpdateCounterParams) -> CounterRecord async

Record the call and mutate the seeded record. Delegates to _update_helper.

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

async def update_counter(self, helper_id: str, params: UpdateCounterParams) -> CounterRecord:
    """Record the call and mutate the seeded record. Delegates to _update_helper."""
    return cast("CounterRecord", self._update_helper(CounterRecord, "update_counter", helper_id, params))

delete_counter(helper_id: str) -> None async

Record the call and remove the seeded record. Delegates to _delete_helper.

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

async def delete_counter(self, helper_id: str) -> None:
    """Record the call and remove the seeded record. Delegates to _delete_helper."""
    self._delete_helper(CounterRecord, "delete_counter", helper_id)

list_timers() -> list[TimerRecord] async

Return all seeded timer helpers. Delegates to _list_helper.

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

async def list_timers(self) -> list[TimerRecord]:
    """Return all seeded timer helpers. Delegates to _list_helper."""
    return cast("list[TimerRecord]", self._list_helper(TimerRecord))

create_timer(params: CreateTimerParams) -> TimerRecord async

Record the call and add a record to helper_definitions. Delegates to _create_helper.

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

async def create_timer(self, params: CreateTimerParams) -> TimerRecord:
    """Record the call and add a record to helper_definitions. Delegates to _create_helper."""
    return cast("TimerRecord", self._create_helper(TimerRecord, "create_timer", params))

update_timer(helper_id: str, params: UpdateTimerParams) -> TimerRecord async

Record the call and mutate the seeded record. Delegates to _update_helper.

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

async def update_timer(self, helper_id: str, params: UpdateTimerParams) -> TimerRecord:
    """Record the call and mutate the seeded record. Delegates to _update_helper."""
    return cast("TimerRecord", self._update_helper(TimerRecord, "update_timer", helper_id, params))

delete_timer(helper_id: str) -> None async

Record the call and remove the seeded record. Delegates to _delete_helper.

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

async def delete_timer(self, helper_id: str) -> None:
    """Record the call and remove the seeded record. Delegates to _delete_helper."""
    self._delete_helper(TimerRecord, "delete_timer", helper_id)

increment_counter(entity_id: str) -> None async

Record an increment_counter call directly (not via call_service).

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

async def increment_counter(self, entity_id: str) -> None:
    """Record an increment_counter call directly (not via call_service)."""
    self.calls.append(
        ApiCall(
            method="increment_counter",
            args=(entity_id,),
            kwargs={"entity_id": entity_id},
        )
    )

decrement_counter(entity_id: str) -> None async

Record a decrement_counter call directly (not via call_service).

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

async def decrement_counter(self, entity_id: str) -> None:
    """Record a decrement_counter call directly (not via call_service)."""
    self.calls.append(
        ApiCall(
            method="decrement_counter",
            args=(entity_id,),
            kwargs={"entity_id": entity_id},
        )
    )

reset_counter(entity_id: str) -> None async

Record a reset_counter call directly (not via call_service).

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

async def reset_counter(self, entity_id: str) -> None:
    """Record a reset_counter call directly (not via call_service)."""
    self.calls.append(
        ApiCall(
            method="reset_counter",
            args=(entity_id,),
            kwargs={"entity_id": entity_id},
        )
    )

__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_state_value_typed, 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_state_value_typed, 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.
    """
    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:
            # Check that all expected kwargs appear in the call's recorded kwargs.
            # Write methods record positional args in both call.args and call.kwargs
            # so kwargs-based assertions work uniformly for all methods.
            if all(k in call.kwargs and call.kwargs[k] == v for k, v in kwargs.items()):
                return
        raise AssertionError(
            f"'{method}' was called {len(matching)} time(s), but none matched kwargs {kwargs!r}. "
            f"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="homeassistant" 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": "homeassistant"}
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="homeassistant")

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="homeassistant"`` 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": "homeassistant"}
        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="homeassistant")
    """
    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
    raise AssertionError(
        f"'{method}' was called {len(matching)} time(s), but none matched kwargs exactly {kwargs!r}. "
        f"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.
    """
    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.
    """
    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 helper_definitions 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.

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

def reset(self) -> None:
    """Clear all recorded calls and reset helper_definitions 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.
    """
    self.calls = []
    self.helper_definitions = {d: {} for d in SUPPORTED_HELPER_DOMAINS}

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"
• 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"``
        - ``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()

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, 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
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,
    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
        context: Event context dict

    Returns:
        Dictionary matching Home Assistant state format
    """
    now = _date_utils.now().format_iso()
    return {
        "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},
    }

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_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	.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: []

Example::

async def test_something():
    hassette = make_mock_hassette()
    assert hassette.config.token == "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 ``[]``
        - ``.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``: ``[]``

    Example::

        async def test_something():
            hassette = make_mock_hassette()
            assert hassette.config.token == "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
    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
    hassette._scheduler_service.register_removal_callback = Mock()
    hassette._scheduler_service.deregister_removal_callback = Mock()

    # Bus service stubs
    hassette._bus_service.remove_listeners_by_owner = Mock()
    hassette._bus_service.get_listeners_by_owner = Mock(return_value=[])

    # 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.database_service = None

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

    # Resource children
    hassette.children = []

    if sealed:
        seal(hassette)

    return hassette