Test Harness Reference

AppTestHarness wires an App subclass into Hassette's test infrastructure without a live Home Assistant connection. It exposes the app's bus, scheduler, seeded state, and a RecordingApi that records every API call.

Basic Pattern

A typical test creates a harness, simulates an event, and asserts on the API calls the app made:

from hassette.test_utils import AppTestHarness

from my_apps.motion_lights import MotionLights


async def test_light_turns_on_when_motion_detected():
    async with AppTestHarness(
        MotionLights,
        config={"motion_entity": "binary_sensor.hallway", "light_entity": "light.hallway"},
    ) as harness:
        await harness.simulate_state_change("binary_sensor.hallway", old_value="off", new_value="on")
        harness.api_recorder.assert_called(
            "turn_on",
            entity_id="light.hallway",
        )

The async with block initializes the app (running on_initialize), then tears it down on exit. simulate_state_change publishes an event through the bus and waits for all handlers to finish. api_recorder.assert_called verifies the app called the expected service.

The sections below cover each piece of this pattern in detail.

Prerequisites

The harness ships in the hassette[test] extra. Write Your First Test covers installation and pyproject.toml setup.

Seeding State

set_state() pre-populates a single entity's state into the state proxy (the in-process state store that app code reads via self.states). set_states() pre-populates multiple entities at once.

from hassette.test_utils import AppTestHarness

from my_apps.thermostat import ThermostatApp


async def test_thermostat_state_seeding():
    async with AppTestHarness(ThermostatApp, config={}) as harness:
        # Seed a single entity
        await harness.set_state("sensor.temperature", "20.5", unit_of_measurement="°C")

        # Seed multiple entities at once
        await harness.set_states(
            {
                "sensor.temperature": ("20.5", {"unit_of_measurement": "°C"}),
                "sensor.humidity": "55",
                "climate.living_room": "heat",
            }
        )

Both methods write directly to the state proxy. No bus events fire. Handlers do not run.

set_states() accepts a plain state string or a (state, attrs) tuple per entity.

Seed state before simulating events

set_state() does not fire bus events. It must precede simulate_state_change() for the same entity, not follow it. A later set_state() silently overwrites the state the simulation wrote.

Simulating Events

Every simulate_* method sends an event through the bus and waits for all triggered handlers to complete before returning.

State Changes

simulate_state_change() publishes a state_changed event and drains all handlers.

from hassette.test_utils import AppTestHarness

from my_apps.motion_lights import MotionLights


async def test_simulate_state_changes():
    async with AppTestHarness(MotionLights, config={}) as harness:
        # Basic state change
        await harness.simulate_state_change(
            "binary_sensor.motion",
            old_value="off",
            new_value="on",
        )

        # With attributes
        await harness.simulate_state_change(
            "sensor.temperature",
            old_value="20.0",
            new_value="21.5",
            old_attrs={"unit_of_measurement": "°C"},
            new_attrs={"unit_of_measurement": "°C"},
        )

Typed dependency injection via D.StateNew[T] (hassette.dependencies) delivers the new state as a typed object:

from hassette import D, states
from hassette.app import App, AppConfig
from hassette.test_utils import AppTestHarness


class SecurityConfig(AppConfig):
    door_entity: str


class SecurityApp(App[SecurityConfig]):
    async def on_initialize(self):
        await self.bus.on_state_change(
            self.app_config.door_entity,
            changed_to="on",
            handler=self.on_door_opened,
            name="door_opened",
        )

    async def on_door_opened(self, new_state: D.StateNew[states.BinarySensorState]):
        device_class = new_state.attributes.device_class
        if device_class == "door":
            await self.api.call_service("notify", "send_message", message="Door opened")


async def test_typed_state_change_handler():
    async with AppTestHarness(SecurityApp, config={"door_entity": "binary_sensor.front_door"}) as harness:
        await harness.set_state("binary_sensor.front_door", "off", device_class="door")
        await harness.simulate_state_change("binary_sensor.front_door", old_value="off", new_value="on")
        harness.api_recorder.assert_called("call_service", domain="notify", service="send_message", message="Door opened")

When old_attrs or new_attrs is omitted, simulate_state_change() merges attributes from the state proxy automatically. Attributes seeded via set_state() appear in the event without being passed again.

Attribute Changes

simulate_attribute_change() changes one attribute while keeping the entity's state value. It delegates to simulate_state_change() internally.

from hassette.test_utils import AppTestHarness

from my_apps.light_app import LightApp


async def test_simulate_attribute_change():
    async with AppTestHarness(LightApp, config={}) as harness:
        await harness.simulate_attribute_change(
            "light.kitchen",
            "brightness",
            old_value=128,
            new_value=255,
        )

        # Seed state first to avoid the "unknown" fallback in predicates
        await harness.set_state("light.kitchen", "on", brightness=128)
        # ...or pass state= explicitly for a one-off:
        await harness.simulate_attribute_change(
            "light.kitchen",
            "brightness",
            old_value=128,
            new_value=255,
            state="on",  # avoids the "unknown" fallback
        )

State value resolution order: the explicit state= argument, the value cached in the state proxy, then "unknown" as a fallback. Seeding state first avoids the fallback.

Attribute changes can fire state-change handlers

on_state_change handlers registered with changed=False fire on any state_changed event, including attribute-only changes. When an app registers such a handler, simulate_attribute_change() fires both it and any on_attribute_change handler for the same entity.

from hassette import App, AppConfig
from hassette.test_utils import AppTestHarness


class SensorApp(App[AppConfig]):
    async def on_initialize(self):
        # on_state_change with changed=False fires even when only attributes change
        await self.bus.on_state_change("sensor.temp", changed=False, handler=self.on_temp_state, name="temp_state")
        await self.bus.on_attribute_change("sensor.temp", "temperature", handler=self.on_temp_attr, name="temp_attr")

    async def on_temp_state(self):
        await self.api.turn_on("light.indicator")

    async def on_temp_attr(self):
        await self.api.turn_on("light.indicator")


async def test_both_handlers_fire():
    async with AppTestHarness(SensorApp, config={}) as harness:
        # simulate_attribute_change fires BOTH handlers because on_state_change
        # was registered with changed=False (fires for any state_changed event).
        # With the default changed=True, only the attribute handler would fire.
        await harness.simulate_attribute_change("sensor.temp", "temperature", old_value=20, new_value=21)

        # Both handlers ran
        harness.api_recorder.assert_call_count("turn_on", 2)

Service Calls

simulate_call_service() publishes a call_service event and drains all handlers.

from hassette.test_utils import AppTestHarness

from my_apps.light_app import LightApp


async def test_simulate_call_service():
    async with AppTestHarness(LightApp, config={}) as harness:
        await harness.simulate_call_service(
            "light",
            "turn_on",
            entity_id="light.kitchen",
            brightness=200,
        )

D.Domain (hassette.dependencies) injects the service domain into handlers the same way D.StateNew works for state changes:

from hassette import D
from hassette.app import App, AppConfig
from hassette.test_utils import AppTestHarness


class AuditConfig(AppConfig):
    pass


class AuditApp(App[AuditConfig]):
    async def on_initialize(self):
        await self.bus.on_call_service(domain="light", handler=self.on_light_service, name="light_service")

    async def on_light_service(self, domain: D.Domain):
        await self.api.call_service("notify", "log", message=f"Service called on {domain}")


async def test_typed_call_service_handler():
    async with AppTestHarness(AuditApp, config={}) as harness:
        await harness.simulate_call_service("light", "turn_on", entity_id="light.kitchen")
        harness.api_recorder.assert_called("call_service", domain="notify", service="log", message="Service called on light")

Hassette Service Events

simulate_hassette_service_status() fires a Hassette-internal service lifecycle event. Convenience wrappers cover the common cases:

Method	Status	ready
simulate_hassette_service_ready(resource_name)	RUNNING	True
simulate_hassette_service_started(resource_name)	RUNNING	False
simulate_hassette_service_failed(resource_name)	FAILED	False
simulate_hassette_service_crashed(resource_name)	CRASHED	False

from hassette.app import App, AppConfig
from hassette.test_utils import AppTestHarness
from hassette.types.enums import ResourceStatus


class WatchdogConfig(AppConfig):
    pass


class WatchdogApp(App[WatchdogConfig]):
    async def on_initialize(self):
        await self.bus.on_hassette_service_failed(handler=self.on_service_failed, name="service_watchdog")

    async def on_service_failed(self) -> None:
        await self.api.call_service("notify", "send_message", message="Service failed")


async def test_service_failure_triggers_notification():
    async with AppTestHarness(WatchdogApp, config={}) as harness:
        await harness.simulate_hassette_service_failed("WebSocketService")
        harness.api_recorder.assert_called(
            "call_service",
            domain="notify",
            service="send_message",
            message="Service failed",
        )


async def test_granular_service_status():
    """You can also simulate specific status transitions."""
    async with AppTestHarness(WatchdogApp, config={}) as harness:
        await harness.simulate_hassette_service_status(
            "SchedulerService",
            ResourceStatus.FAILED,
            previous_status=ResourceStatus.RUNNING,
            exception=ConnectionError("connection lost"),
        )
        harness.api_recorder.assert_called(
            "call_service",
            domain="notify",
            service="send_message",
            message="Service failed",
        )

simulate_hassette_service_status() accepts previous_status, exception, and role for cases the convenience wrappers do not cover.

Timeouts

All simulate_* methods default to a 2-second drain timeout. The timeout= parameter overrides this per call.

from hassette.test_utils import AppTestHarness

from my_apps.slow_app import SlowApp


async def test_with_custom_timeout():
    async with AppTestHarness(SlowApp, config={}) as harness:
        await harness.simulate_state_change(
            "sensor.slow_device",
            old_value="off",
            new_value="on",
            timeout=5.0,
        )

The drain mechanism waits until both the bus dispatch queue and the app's task bucket are quiescent.

Exception	Meaning
DrainTimeout	Handlers did not finish within the deadline
DrainError	One or more handlers raised an exception
DrainFailure	Base class; catches both of the above

When tasks include debounce handlers, the timeout= value should exceed the debounce window. Concurrency covers task chain draining in detail.

Asserting API Calls

harness.api_recorder exposes a RecordingApi that records every call the app makes through self.api: turn_on, turn_off, call_service, set_state, fire_event, and all helper CRUD methods.

assert_called

assert_called(method, **kwargs) passes when at least one recorded call matches every specified key-value pair. Extra kwargs in the recorded call are allowed (partial match).

from hassette.test_utils import AppTestHarness

from my_apps.motion_lights import MotionLights


async def test_assert_called_examples():
    async with AppTestHarness(MotionLights, config={}) as harness:
        await harness.simulate_state_change("binary_sensor.motion", old_value="off", new_value="on")

        # Assert turn_on was called for a specific entity
        harness.api_recorder.assert_called(
            "turn_on",
            entity_id="light.kitchen",
            domain="light",
        )

        # Assert fire_event was called with a specific event type
        harness.api_recorder.assert_called("fire_event", event_type="my_custom_event")

        # Assert call_service was called directly (for services without a named wrapper)
        harness.api_recorder.assert_called(
            "call_service",
            domain="light",
            service="set_color_temp",
            target={"entity_id": "light.kitchen"},
        )

turn_on, turn_off, and toggle_service record under their own names, not under call_service:

from hassette.test_utils import AppTestHarness

from my_apps.motion_lights import MotionLights


async def test_turn_on_off_recording():
    async with AppTestHarness(MotionLights, config={}) as harness:
        await harness.simulate_state_change("binary_sensor.motion", old_value="off", new_value="on")

        # Your app calls: await self.api.turn_on("light.kitchen", domain="light")
        harness.api_recorder.assert_called("turn_on", entity_id="light.kitchen", domain="light")

        # Your app calls: await self.api.turn_off("light.kitchen", domain="light")
        harness.api_recorder.assert_called("turn_off", entity_id="light.kitchen", domain="light")

assert_not_called

assert_not_called(method, **kwargs) raises AssertionError when a matching call exists. With kwargs, only calls whose recorded kwargs include all given pairs count as a violation.

from hassette.test_utils import AppTestHarness

from my_apps.motion_lights import MotionLights


async def test_assert_not_called():
    async with AppTestHarness(MotionLights, config={}) as harness:
        harness.api_recorder.assert_not_called("call_service")

assert_call_count

assert_call_count(method, count, **kwargs) raises AssertionError when the method was not called exactly count times. With kwargs, only matching calls are counted.

from hassette.test_utils import AppTestHarness

from my_apps.motion_lights import MotionLights


async def test_assert_call_count():
    async with AppTestHarness(MotionLights, config={}) as harness:
        await harness.simulate_state_change("binary_sensor.motion", old_value="off", new_value="on")
        harness.api_recorder.assert_call_count("call_service", 2)

get_calls

get_calls(method) returns a list of ApiCall records for the named method. Each ApiCall has method, args, and kwargs fields. Omitting method returns all recorded calls.

from hassette.test_utils import AppTestHarness

from my_apps.motion_lights import MotionLights


async def test_get_calls():
    async with AppTestHarness(MotionLights, config={}) as harness:
        await harness.simulate_state_change("binary_sensor.motion", old_value="off", new_value="on")

        calls = harness.api_recorder.get_calls("call_service")
        for call in calls:
            print(call.kwargs)  # e.g. {"domain": "light", "service": "turn_on", ...}

reset

reset() clears all recorded calls and resets helper definitions. Mid-test isolation is the primary use case: asserting separately on two distinct phases within one test.

from hassette.test_utils import AppTestHarness

from my_apps.motion_lights import MotionLights


async def test_recorder_reset():
    async with AppTestHarness(MotionLights, config={}) as harness:
        await harness.simulate_state_change("binary_sensor.motion", old_value="off", new_value="on")
        harness.api_recorder.reset()  # ignore calls from the above simulate

        await harness.simulate_state_change("binary_sensor.motion", old_value="on", new_value="off")
        harness.api_recorder.assert_called("turn_off", entity_id="light.hallway", domain="light")

reset() replaces the calls list with a new empty list. Any snapshot taken before the reset (e.g., saved = harness.api_recorder.calls) retains the original calls.

Testing Configuration Errors

AppConfigurationError raises during async with AppTestHarness(...) entry when the config dict fails validation. The async with body never runs.

import pytest

from hassette.test_utils import AppConfigurationError, AppTestHarness

from my_apps.motion_lights import MotionLights


async def test_missing_config_raises():
    with pytest.raises(AppConfigurationError) as exc_info:
        async with AppTestHarness(MotionLights, config={}) as harness:
            pass

    # The error message includes the field name and validation failure reason
    print(exc_info.value)
    # AppConfigurationError for MotionLights: 1 validation error — field 'motion_entity': Field required

Attribute	Type	Description
app_cls	type[App]	The App class whose config failed
original_error	pydantic.ValidationError	The underlying Pydantic error

Testing Startup Failures

When on_initialize() raises, the harness startup times out with a plain TimeoutError. This is distinct from DrainTimeout, which only surfaces from simulate_* methods. The exception from on_initialize() appears in the log output.

Harness Constructor and Properties

Constructor

from hassette import App
from hassette.test_utils import AppTestHarness

AppTestHarness(
    app_cls=App,       # Replace with your App subclass
    config={},         # Dict matching your app's AppConfig fields
    tmp_path=None,     # Optional: Path | None
)

Parameter	Type	Description
app_cls	type[App]	The App subclass to test
config	dict[str, Any]	Config values validated against the app's AppConfig
tmp_path	Path \| None	Directory for Hassette data. Auto-created and cleaned up if omitted.

Properties

Property	Type	Description
harness.app	App	Fully initialized app instance
harness.bus	Bus	Test bus owned by the app
harness.scheduler	Scheduler	Test scheduler owned by the app
harness.api_recorder	RecordingApi	Records every API call the app makes
harness.states	StateManager	State manager owned by the app

Next Steps

• Time Control: freeze time and trigger scheduled jobs
• Concurrency & pytest-xdist: parallel test execution and drain failure details
• Factories: build custom state dicts and event objects