Testing Your Apps

Hassette ships with hassette.test_utils — a set of utilities for testing your automations in isolation, without a running Home Assistant instance. You can simulate state changes, inspect API calls your app makes, and control time for scheduler tests.

The core idea: AppTestHarness runs your app against a test-grade Hassette environment with a RecordingApi in place of a live HA connection. When your app calls self.api.turn_on(), self.api.call_service(), or any other API method, RecordingApi records those calls instead of contacting Home Assistant — you then assert on the recorder via harness.api_recorder.

Installation

hassette.test_utils is part of the main hassette package — no extra install required. You only need to add your test runner:

pip install pytest pytest-asyncio

Or with uv:

uv add --dev pytest pytest-asyncio

Add this to your pyproject.toml to configure pytest-asyncio:

[tool.pytest.ini_options]
asyncio_mode = "auto"

With asyncio_mode = "auto", any async def test_* function is automatically treated as an async test — no @pytest.mark.asyncio decorator required. If you skip this config, your async tests will silently succeed without actually running — a silent false-green failure mode. The examples on this page assume asyncio_mode = "auto" is set.

Quick Start

Here's a complete test for an app that turns on a light when motion is detected:

Replace the placeholders with your own app

Replace MotionLights with your app class and motion_lights with your module path. The config keys (motion_entity, light_entity) should match the fields on your app's AppConfig subclass.

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",
        )

After async with, the app is fully initialized and ready to receive events. The harness tears everything down cleanly when the async with block exits.

The Test Harness

AppTestHarness wires your app class into a test-grade Hassette environment with a RecordingApi instead of a live HA connection.

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	Description
app_cls	Your App subclass to test.
config	Dict of config values. Keys must match the fields defined on your app's AppConfig subclass (see App Configuration). Invalid or missing fields raise AppConfigurationError during harness setup.
tmp_path	Optional directory for Hassette data files. Created and cleaned up automatically if omitted. In pytest, pass the built-in tmp_path fixture to share a directory across multiple harnesses in one test.

Properties

Once inside the async with block, the harness exposes:

Property	Type	Description
harness.app	App	The fully initialized app instance.
harness.bus	Bus	The test bus your app registered handlers on.
harness.scheduler	Scheduler	The test scheduler your app registered jobs on.
harness.api_recorder	RecordingApi	Records every API call your app makes. Use this for assertions.
harness.states	StateManager	The state manager your app reads from.

State Seeding

Before simulating events, seed the state of any entities your app reads. Use set_state() for a single entity or set_states() for multiple 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",
            }
        )

set_states() accepts either a plain state string or a (state, attributes) tuple.

set_state() does not fire bus events

set_state() is for pre-test setup only. It writes directly to the state proxy without publishing a state_changed event, so no handlers will fire. Do not use set_state() mid-test to simulate a state transition — use simulate_state_change() instead.

A second hazard: calling set_state() after a simulate_state_change() for the same entity will silently overwrite the simulated state with the seeded value, which can make subsequent reads return wrong values. Seed first, simulate second.

Simulating Events

If your handler reads entity state during handling (e.g., self.states.light.get("light.kitchen")), seed it first with harness.set_state(). Simulating an event does not update the state proxy automatically unless your handler writes back via the API.

State changes

simulate_state_change() publishes a state_changed event through the bus and waits for all triggered handlers to finish before returning.

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"},
        )

Attribute changes

simulate_attribute_change() simulates a change to a single attribute while keeping the state value the same.

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
        )

The generated event carries the entity's current cached state for the state field. If you haven't seeded the entity with set_state() first, that field defaults to "unknown" — which silently breaks any state-conditional predicates on the same entity. You can pass an explicit state= to avoid this, as shown above.

simulate_attribute_change can also fire state-change handlers

This method delegates to simulate_state_change under the hood. With the default changed=True, state-change handlers do not fire (the state value is unchanged). But if your app registers on_state_change with changed=False, that handler will fire — matching HA's real behavior where state_changed events fire even when only attributes change:

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)

Use harness.api_recorder.reset() between simulate calls, or get_calls() for targeted inspection, to isolate which handler made which API call.

Service call events

simulate_call_service() publishes a call_service event, useful for apps that listen for HA service calls.

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,
        )

Timeouts and slow handlers

All three simulate methods wait for dispatched handlers to finish before returning. The default timeout is 2 seconds. Override it with the timeout= parameter:

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,
        )

Task chains drain to completion

The drain is iterative: after the bus dispatch queue clears, any tasks spawned by self.task_bucket.spawn(...) inside a handler are awaited in turn, to arbitrary depth. simulate_* does not return until the full chain is settled. If a task raises or the drain times out, a DrainFailure subclass is raised — see DrainFailure Exception Hierarchy for the full exception hierarchy and catch patterns.

Typed dependency injection in handlers

Hassette handlers support typed dependency injection via D.* annotations. These work seamlessly with simulate_* — the harness dispatches the same event objects that production code receives, so DI resolution runs identically.

State change with D.StateNew — extract a typed state model from the event:

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")

Service call with D.Domain — extract the service domain from the event:

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() and its convenience wrappers (simulate_hassette_service_failed, simulate_hassette_service_crashed, simulate_hassette_service_started) let you test how your app responds to internal service lifecycle changes.

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",
        )

Asserting API Calls

harness.api_recorder records every call your app makes to self.api. Use it to assert that your app called the right services.

assert_called

Passes if the method was called at least once with kwargs that match all specified values (partial matching — additional kwargs in the recorded call are allowed).

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

These convenience methods record calls using their own method name — not call_service. Assert them directly:

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")

Use assert_called("call_service", ...) only for direct self.api.call_service(...) calls.

assert_not_called

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

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

Returns a list of ApiCall records, optionally filtered by method name. Each ApiCall has method, args, and kwargs attributes.

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

Clears all recorded calls. Useful when you want to assert on calls made after a specific point in your 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")

Configuration Errors

If the config dict you pass to AppTestHarness fails validation against your app's AppConfig class, the harness raises AppConfigurationError during setup — the async with body is never entered, so your test code inside the block does not run.

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

AppConfigurationError has two attributes: - app_cls — the App class whose config failed. - original_error — the underlying pydantic.ValidationError with full field-level detail.

Read the error message to find which field is missing or invalid, then fix the config dict in your test.

Harness Startup Failures

If the harness raises TimeoutError: Timed out waiting for <YourApp> RUNNING, the app's on_initialize() either raised an exception or took longer than 5 seconds to complete.

This is a bare TimeoutError, not DrainTimeout

Harness startup timeouts are distinct from drain timeouts. The startup wait still raises a plain TimeoutError — catch TimeoutError here, not DrainTimeout or DrainFailure. Drain-related failures only happen once the harness is running and you call simulate_*.

Check test output for log lines near the TimeoutError — exceptions raised during on_initialize() are caught and logged at WARNING level during harness cleanup, so the TimeoutError is the symptom, not the root cause.

Common triggers: - A required config field is present but its value causes a runtime error during initialization (distinct from a missing-field AppConfigurationError which fires before entry). - on_initialize() awaits something that never resolves, such as an external call that isn't mocked. - An await inside on_initialize() raises an exception that propagates out.

Next Steps

• Time Control: Freeze and advance time to test scheduler-driven behavior
• Concurrency & pytest-xdist: Understand the same-class lock and parallel test runners
• Factories & Internals: Event factories, state factories, make_test_config, and RecordingApi coverage boundary
• Bus: Event subscriptions and handler registration
• Scheduler: Timed and recurring job registration
• API: The self.api methods your app calls