Skip to content

Time Control

The time control API on AppTestHarness freezes the harness clock and advances it manually. This makes scheduler-driven behavior deterministic in tests — without it, tests that wait on scheduled jobs depend on real wall-clock time. The examples assert with harness.api_recorder, the recorder that tracks every HA API call (covered in Test Harness).

whenever is installed automatically

Code examples on this page import from whenever, Hassette's date/time library. It ships as a direct dependency of hassette, so no separate install is needed.

The canonical sequence is: freeze, advance, trigger.

from whenever import Instant

from hassette.test_utils import AppTestHarness

from my_apps.reminder import ReminderApp


async def test_reminder_fires_after_one_hour():
    async with AppTestHarness(ReminderApp, config={}) as harness:
        # 1. Freeze time at a known point
        start = Instant.from_utc(2024, 1, 15, 9, 0, 0)  # 2024-01-15 09:00 UTC
        harness.freeze_time(start)

        # 2. The app registered its job in on_initialize when the harness started

        # 3. Advance the frozen clock
        harness.advance_time(hours=1)

        # 4. Fire any jobs whose due time is now <= frozen clock
        count = await harness.trigger_due_jobs()
        assert count == 1

        # 5. Assert your app made the expected API call
        harness.api_recorder.assert_called("fire_event", event_type="reminder_fired")

freeze_time pins the clock at a known point. advance_time moves it forward. trigger_due_jobs fires every job whose scheduled time is at or before the frozen clock.

The three steps are separate because advancing time and dispatching jobs are distinct operations. A test that advances the clock by 30 minutes may only care about the first job. Remaining due jobs stay untriggered until an explicit trigger_due_jobs call. This separation gives each test precise control over which jobs fire and when.

freeze_time(instant)

freeze_time patches hassette.utils.date_utils.now to return a fixed time. The instant parameter accepts an Instant or ZonedDateTime from whenever.

from whenever import Instant, ZonedDateTime

from hassette.test_utils import AppTestHarness

from my_apps.reminder import ReminderApp


async def test_freeze_time_variants():
    async with AppTestHarness(ReminderApp, config={}) as harness:
        # From a UTC instant (most portable)
        harness.freeze_time(Instant.from_utc(2024, 6, 1, 8, 0, 0))

        # From a ZonedDateTime (when local time matters)
        harness.freeze_time(ZonedDateTime(2024, 6, 1, 8, 0, 0, tz="America/Chicago"))

Calling freeze_time again replaces the frozen time. The old patchers stop and new ones start. The clock unfreezes automatically when the harness async with block exits.

advance_time(*, seconds, minutes, hours)

advance_time moves the frozen clock forward by the given delta. The seconds, minutes, and hours keywords combine in a single call.

from whenever import Instant

from hassette.test_utils import AppTestHarness

from my_apps.reminder import ReminderApp


async def test_advance_time_variants():
    async with AppTestHarness(ReminderApp, config={}) as harness:
        harness.freeze_time(Instant.from_utc(2024, 1, 15, 9, 0, 0))

        harness.advance_time(seconds=30)
        harness.advance_time(minutes=5)
        harness.advance_time(hours=1)
        harness.advance_time(hours=1, minutes=30)  # combined

advance_time does not trigger jobs

Advancing the clock does not dispatch any scheduled jobs. trigger_due_jobs() must be called explicitly afterward. Without it, jobs accumulate silently and side-effect assertions fail.

trigger_due_jobs()

trigger_due_jobs fires all jobs whose scheduled time is at or before the current frozen clock. It returns the number of jobs dispatched and completed.

from whenever import Instant

from hassette.test_utils import AppTestHarness

from my_apps.reminder import ReminderApp


async def test_trigger_due_jobs():
    async with AppTestHarness(ReminderApp, config={}) as harness:
        harness.freeze_time(Instant.from_utc(2024, 1, 15, 9, 0, 0))
        harness.advance_time(hours=1)

        count = await harness.trigger_due_jobs()
        assert count == 1

trigger_due_jobs operates on a snapshot of due jobs taken at the moment of the call. Jobs re-enqueued during dispatch (repeating jobs) are not included in that snapshot and are not re-triggered in the same call. This prevents infinite loops when the clock is frozen.

If dispatched jobs send events through the bus, downstream handler tasks are spawned but not drained by trigger_due_jobs. await harness.drain_task_bucket() waits for those handler tasks to complete before assertions run — see Test Harness for the full method reference.

Next Steps