Contents · 17 sections↓
- 01How do you test a scheduled AI agent?
- 02How testing an autonomous agent works
- 03When to write process tests (and when not to)
- 04What you need
- 05Step 1: Make the agent testable
- 06Step 2: Replace the model with a script
- 07Step 3: Start the process in a fixture and inject ticks
- 08Step 4: Assert on what a run did
- 09Step 5: Test hooks on their own and in the process
- 10Step 6: Freeze the clock for daily budgets
- 11Step 7: Test code that reads the journal
- 12The complete example
- 13Running it in production
- 14Common problems and fixes
- 15Limitations
- 16Frequently asked questions
- 17Next steps
To test a scheduled AI agent, you can't wait for 08:00, pay for a model call on every test, or hope the model makes the same tool calls twice. This guide is for developers who run agents on cron schedules, webhooks or queues and want a test suite that runs in seconds. It explains what to test in an autonomous agent and builds a pytest suite with Promptise Foundry. The suite covers ticks without a clock, a scripted model, hooks, dead letters, events and a daily budget tested with a frozen clock.
How do you test a scheduled AI agent?
Don't wait for the trigger. Start the agent's real process in the test and put the event its trigger would produce on the queue yourself. Replace the LLM with a scripted model that makes the same tool calls on every run. Wait until the run has finished, then assert on what changed outside the agent (the tools' side effects) and inside it (journal, state, status). Check the failure paths too: dead letters, refused calls, events, hooks. Freeze the clock for anything time-based, such as a daily budget. In Promptise Foundry that is process.inject(TriggerEvent(...)), a scripted chat model, a task_completed hook to wait on, and a file journal in pytest's tmp_path.
`inject()` replaces the clock for triggers. The event goes through the same queue, hooks and worker as a real tick.
A scripted model makes runs reproducible and free. Tools, approvals, budgets and hooks stay real.
Wait on `task_completed`, never on `sleep()`. A run happens in a background worker after inject() returns.
Test judgment separately. Scripted tests check the plumbing; evals and a nightly live test check the model.
[02]
How testing an autonomous agent works
A request-response agent is easy to test: call it, read the answer. An autonomous agent is harder for four reasons:
Time starts its runs. Cron triggers, file watchers and webhooks decide when the agent works, and daily limits reset at a fixed hour.
Runs are asynchronous. A trigger puts an event on a queue, and a worker runs the agent later. The test must know when the run is over.
The model is nondeterministic and costs money. Two runs of the same test can make different tool calls.
The risky code is the failure handling. Retries, dead letters, refused tool calls and blocked runs rarely happen in a manual try-out, so they go untested.
The answer is to keep everything real except the two things you can't control, time and the model, and to drive the agent through the same doors production uses:
test AgentProcess (real)
──── ───────────────────
inject(cron-shaped event) ───────▶ task_created hooks ──▶ queue ──┐
▼
worker: budget check, frozen clock
│
▼
scripted model ──▶ real MCP tools ──▶ side effects (tmp_path)
│
▼
journal (tmp_path), events, dead letters
│
await task_completed ◀────────────── task_completed hook ◀────────┘
assert on outbox, journal, dead_letters, events, status()Not every test needs the whole process. The table shows the layers:
Test | Model | What it catches | When it runs |
|---|---|---|---|
Unit test of a hook or a journal reader | None | Policy and reporting bugs | Every commit |
Process test (this guide) | Scripted | Wiring: tools, budgets, hooks, retries, dead letters | Every commit, seconds |
Live smoke test | Real | Prompt and tool-schema drift, provider changes | Nightly |
Evals | Real | Answer quality and tool choice over many cases | Before releases |
[03]
When to write process tests (and when not to)
Write them for every agent that runs unattended, and for every rule you would hate to see fail in production: "never send two reminders for one invoice", "stop after five emails", "do nothing during a freeze". They are fast, so they belong in the normal CI run.
Don't use them to judge the model. A scripted model always makes the calls you wrote, so a passing test says nothing about whether the real model would make them. That is what evaluating AI agents and a nightly live test are for. Don't test the scheduler's calendar math with them either: preview a cron expression's next fire times instead, as AI agent cron jobs shows.
[04]
What you need
Python 3.10 or later, Promptise Foundry 1.5.0 or later, pytest and pytest-asyncio.
No API key for the main suite. The optional live test uses OPENAI_API_KEY and gpt-5-mini.
About 20 minutes.
pip install "promptise>=1.5.0" pytest pytest-asyncioThe suite uses asyncio_mode = auto in pytest.ini, so async tests and fixtures need no decorators.
[05]
Step 1: Make the agent testable
The agent under test sends payment reminders every weekday at 08:00. Its MCP server (billing_server.py) has a read-only list_overdue tool and a send_reminder tool that appends to an outbox file named by the OUTBOX environment variable. The process config is built by a function, and everything that touches the outside world is a parameter:
"""The agent under test: sends payment reminders every weekday at 08:00."""
import asyncio
import sys
from pathlib import Path
from promptise.runtime import (
AgentProcess, BudgetConfig, HookBlocked, HookConfig, HookContext, JournalConfig,
ProcessConfig, ToolCostAnnotation, TriggerConfig,
)
HERE = Path(__file__).parent
def make_config(*, outbox: Path, journal_dir: Path, freeze_file: Path,
schedule: str = "0 8 * * 1-5") -> ProcessConfig:
"""Everything that touches the outside world is a parameter, so tests can redirect it."""
async def honor_freeze(ctx: HookContext) -> None:
if freeze_file.exists():
raise HookBlocked(f"reminders are frozen ({freeze_file.name} exists)")
return ProcessConfig(
model="openai:gpt-5-mini",
instructions="You send payment reminders. On every run, list the overdue invoices, "
"send one reminder per invoice, then say how many you sent.",
servers={"billing": {"command": sys.executable,
"args": [str(HERE / "billing_server.py")],
"env": {"OUTBOX": str(outbox)}}},
triggers=[TriggerConfig(type="cron", cron_expression=schedule)],
journal=JournalConfig(level="full", path=str(journal_dir)),
budget=BudgetConfig(
enabled=True,
max_runs_per_day=1, # one reminder run per budget day
daily_reset_hour_utc=6, # the budget day starts at 06:00 UTC
max_irreversible_per_run=5,
tool_costs={"send_reminder": ToolCostAnnotation(irreversible=True)},
),
hooks=[HookConfig(event="task_created", callback=honor_freeze, fail_closed=True)],
)
if __name__ == "__main__": # production: the real schedule, the real model
data = HERE / "data"
data.mkdir(exist_ok=True)
process = AgentProcess("reminders", make_config(
outbox=data / "outbox.jsonl", journal_dir=data / "journal",
freeze_file=data / "reminders.freeze"))
async def main() -> None:
await process.start()
await asyncio.Event().wait()
asyncio.run(main())The production entry point and the tests call the same make_config(). Tests pass paths under tmp_path and a schedule that can't fire. The config also carries the rules you want to test:
`max_runs_per_day=1` with `daily_reset_hour_utc=6`: one reminder run per budget day, which starts at 06:00 UTC.
`send_reminder` is irreversible, at most five per run: a reminder can't be unsent.
A `task_created` hook: while a freeze file exists, new events are refused before they reach the queue. Operators create the file during a billing incident.
A full journal: level="full" records every tool call and the final answer, which the tests read.
[06]
Step 2: Replace the model with a script
A scripted model plays back a list of turns. A turn is a final answer, a list of tool calls, or an exception:
"""A chat model that plays back a script: no network, no API key, the same turns every run."""
from __future__ import annotations
import itertools
from typing import Any
from langchain_core.language_models.chat_models import BaseChatModel
from langchain_core.messages import AIMessage, BaseMessage
from langchain_core.outputs import ChatGeneration, ChatResult
from pydantic import Field
_ids = itertools.count(1)
def call(name: str, **args: Any) -> list[tuple[str, dict[str, Any]]]:
"""A model turn that calls one tool."""
return [(name, args)]
class ScriptedModel(BaseChatModel):
"""Each turn takes the next step of `script`: an answer, tool calls, or an exception."""
script: list[Any] = Field(default_factory=list)
seen: list[list[BaseMessage]] = Field(default_factory=list)
@property
def _llm_type(self) -> str:
return "scripted"
def bind_tools(self, tools: Any, **kwargs: Any) -> ScriptedModel:
return self
def _generate(self, messages: list[BaseMessage], stop: Any = None,
run_manager: Any = None, **kwargs: Any) -> ChatResult:
self.seen.append(list(messages))
step = self.script.pop(0) if self.script else "Done."
if isinstance(step, Exception):
raise step
if isinstance(step, str):
message = AIMessage(content=step)
else:
message = AIMessage(content="", tool_calls=[
{"name": name, "args": args, "id": f"call_{next(_ids)}"} for name, args in step
])
return ChatResult(generations=[ChatGeneration(message=message)])When the script runs out, the model answers "Done.". bind_tools() returns the model itself, so the agent can bind its tools as it does with a real model. seen records what the model was shown on every turn, including tool results, so a test can check that a refusal reached it.
The process takes a model name ("openai:gpt-5-mini"), not a model object, and builds the agent when it starts. The model fixture in step 3 wraps promptise.agent.build_agent so every agent built during the test gets the scripted model. Everything else is the real build: MCP tools, the budget guard, approval gates and hooks. Promptise's own runtime tests patch the same function.
Run the suite without OPENAI_API_KEY in CI. A test that forgets the fixture then fails at process.start() with ModelSetupError: Cannot use model openai:'gpt-5-mini' (OpenAI) yet instead of quietly spending money.
[07]
Step 3: Start the process in a fixture and inject ticks
conftest.py builds the production config with temporary paths, starts the process, and gives each test a small harness:
"""Fixtures for testing the reminder agent without a clock, a network or an API key."""
import asyncio
import json
import os
from contextlib import asynccontextmanager
from dataclasses import dataclass, field
from pathlib import Path
import pytest
import promptise.agent
from promptise.events import AgentEvent, CallbackSink, EventNotifier
from promptise.runtime import AgentProcess, HookContext, HookEvent, TriggerEvent
from promptise.runtime.journal import FileJournal
from reminder_agent import make_config
from scripted_model import ScriptedModel
def cron_tick(when: str = "2026-10-21T08:00:00+00:00") -> TriggerEvent:
"""The event the cron trigger produces at `when`."""
return TriggerEvent(trigger_id="cron-test", trigger_type="cron", payload={
"scheduled_time": when, "cron_expression": "0 8 * * 1-5", "timezone": "UTC"})
@dataclass
class Agent:
"""A started reminder process plus what a test needs to drive and inspect it."""
process: AgentProcess
model: ScriptedModel | None
tmp_path: Path
notifier: EventNotifier
events: list[AgentEvent] = field(default_factory=list)
finished: asyncio.Queue = field(default_factory=asyncio.Queue)
async def tick(self, event: TriggerEvent | None = None, timeout: float = 30) -> dict:
"""Inject a cron tick and wait until its run is over; returns task_completed's data."""
await self.process.inject(event or cron_tick())
result = await asyncio.wait_for(self.finished.get(), timeout)
await asyncio.sleep(0.1) # let the worker release the event
await self.notifier.flush() # deliver the events emitted so far
return result
def outbox(self) -> list[str]:
path = self.tmp_path / "outbox.jsonl"
lines = path.read_text().splitlines() if path.exists() else []
return sorted(json.loads(line)["invoice_id"] for line in lines)
async def journal(self, entry_type: str) -> list:
return await FileJournal(str(self.tmp_path / "journal")).read(
"reminders", entry_type=entry_type)
def event_types(self) -> list[str]:
return [e.event_type for e in self.events]
@pytest.fixture
def model(monkeypatch) -> ScriptedModel:
"""Every agent built during the test uses this scripted model instead of the real one."""
scripted = ScriptedModel()
real_build = promptise.agent.build_agent
async def build_with_script(**kwargs):
kwargs["model"] = scripted
return await real_build(**kwargs)
monkeypatch.setattr(promptise.agent, "build_agent", build_with_script)
return scripted
@asynccontextmanager
async def started_agent(tmp_path: Path, model: ScriptedModel | None):
"""The production config, with its files in tmp_path and a schedule that won't fire."""
config = make_config(outbox=tmp_path / "outbox.jsonl", journal_dir=tmp_path / "journal",
freeze_file=tmp_path / "reminders.freeze", schedule="0 0 1 1 *")
events: list[AgentEvent] = []
notifier = EventNotifier(sinks=[CallbackSink(events.append)])
process = AgentProcess("reminders", config, event_notifier=notifier)
harness = Agent(process, model, tmp_path, notifier, events)
async def on_finished(ctx: HookContext) -> None:
await harness.finished.put(ctx.data)
process.hooks.register(HookEvent.TASK_COMPLETED, on_finished)
await process.start()
try:
yield harness
finally:
await process.stop()
await notifier.stop()
@pytest.fixture
async def agent(model, tmp_path):
async with started_agent(tmp_path, model) as harness:
yield harness
@pytest.fixture
async def live_agent(tmp_path):
"""The same agent with the real model. Costs money: only for the nightly smoke test."""
if not os.environ.get("OPENAI_API_KEY"):
pytest.skip("OPENAI_API_KEY is not set")
async with started_agent(tmp_path, None) as harness:
yield harnessThe harness methods do the waiting and reading for the tests:
`tick()` injects an event shaped like the cron trigger's own (scheduled_time, cron_expression, timezone) and waits for the task_completed hook of that run. The hook fires after the run and its follow-up steps, such as a budget pause, are done, so assertions see the final state. The timeout turns a hung run into a failed test.
`asyncio.sleep(0.1)` after the hook lets the worker release the finished event. In Promptise 1.5.0, a stop() that arrives while task_completed hooks are still running dead-letters that finished event as "process stopped". The pause keeps teardown from adding a false dead letter.
`notifier.flush()` delivers the events emitted so far to the CallbackSink, which fills events. Events are delivered by a background worker, so without the flush the list can lag behind.
`outbox()` and `journal()` read the side effects and the journal from tmp_path. The process doesn't expose its journal object, so the test reads the files with FileJournal, the same public API you use in production.
The schedule 0 0 1 1 * fires only at midnight on January 1, so no real tick lands in the middle of a test. One process per test keeps tests independent; starting the stdio MCP server is the slowest part, well under a second.
[08]
Step 4: Assert on what a run did
The first test scripts a normal run: list the overdue invoices, send two reminders in one turn, answer. The second makes the model fail:
"""What a tick does: tools called, answer journaled, failures dead-lettered."""
from scripted_model import call
async def test_tick_sends_one_reminder_per_overdue_invoice(agent):
agent.model.script = [
call("list_overdue"),
[("send_reminder", {"invoice_id": "INV-7"}), ("send_reminder", {"invoice_id": "INV-9"})],
"Sent 2 reminders.",
]
result = await agent.tick()
assert result["outcome"] == "success"
assert agent.outbox() == ["INV-7", "INV-9"]
tools = sorted(e.data["tool"] for e in await agent.journal("tool_call"))
assert tools == ["list_overdue", "send_reminder", "send_reminder"]
[answer] = await agent.journal("invocation_result")
assert answer.data["response"] == "Sent 2 reminders."
async def test_model_outage_dead_letters_the_tick(agent):
agent.model.script = [RuntimeError("provider unavailable")]
result = await agent.tick()
assert result["outcome"] == "failed"
[letter] = agent.process.dead_letters
assert letter["reason"] == "retries exhausted"
assert "provider unavailable" in letter["error"]
assert agent.outbox() == []test_runs.py::test_tick_sends_one_reminder_per_overdue_invoice PASSED [ 50%]
test_runs.py::test_model_outage_dead_letters_the_tick PASSED [100%]
============================== 2 passed in 1.52s ===============================The first test checks three different places, because each catches a different bug. The outbox proves the tool really ran with the right arguments. The journal's tool_call entries prove the agent called what you expected and nothing else. The invocation_result entry holds the final answer.
The second test covers the failure path. A model error raises inside the run. With trigger_delivery.max_retries at its default of 0, the event is not retried: task_completed reports outcome="failed" and the event lands in process.dead_letters with reason "retries exhausted" and the error text. Configure retries and the first outcome is "retrying" instead; the retries and dead letter guide covers that path.
[09]
Step 5: Test hooks on their own and in the process
Test a hook twice: once as a plain async function, once wired into the process:
"""The freeze hook: on its own, and wired into the process."""
import pytest
from conftest import cron_tick
from promptise.runtime import HookBlocked, HookContext, HookEvent
from reminder_agent import make_config
async def test_freeze_hook_on_its_own(tmp_path):
freeze = tmp_path / "reminders.freeze"
config = make_config(outbox=tmp_path / "o", journal_dir=tmp_path / "j", freeze_file=freeze)
hook = config.hooks[0].callback
ctx = HookContext(event=HookEvent.TASK_CREATED, process_id="reminders", data={})
await hook(ctx) # no freeze file: no objection
freeze.touch()
with pytest.raises(HookBlocked, match="frozen"):
await hook(ctx)
async def test_freeze_file_blocks_the_run_before_the_model(agent):
(agent.tmp_path / "reminders.freeze").touch()
await agent.process.inject(cron_tick()) # refused while queuing: nothing to wait for
[letter] = agent.process.dead_letters
assert letter["reason"] == "blocked by hook"
assert "frozen" in letter["error"]
assert agent.model.seen == [] # the model was never calledtest_hooks.py::test_freeze_hook_on_its_own PASSED [ 50%]
test_hooks.py::test_freeze_file_blocks_the_run_before_the_model PASSED [100%]
============================== 2 passed in 0.62s ===============================The unit test calls the hook with a HookContext it builds itself. That is the fastest way to cover a hook's logic, with every edge case you can think of.
The process test checks the wiring. A task_created hook runs inside inject(), before the event reaches the queue. When it raises HookBlocked, the event is dead-lettered with reason "blocked by hook" at once, and task_completed never fires because no task was created. That is why this test calls process.inject() directly instead of tick(), which would wait for a hook that never comes. model.seen == [] proves the model was never called. Blocking hooks are covered in depth in AI agent lifecycle hooks.
[10]
Step 6: Freeze the clock for daily budgets
Budgets read the time when a run starts, to decide whether the budget day has rolled over. To test that, these tests replace the budget module's clock with one they control:
"""Budget limits: a cap inside a run, and the budget day with a frozen clock."""
from datetime import datetime, timezone
import pytest
from promptise.runtime import HookContext, HookEvent
class Clock:
"""The time the budget sees. Tests set `clock.now`."""
def __init__(self, now: datetime) -> None:
self.now = now
@pytest.fixture
def clock(monkeypatch) -> Clock:
"""Freeze the budget's clock at 2026-10-21 07:00 UTC (asyncio keeps real time)."""
clock = Clock(datetime(2026, 10, 21, 7, 0, tzinfo=timezone.utc))
class FrozenDatetime(datetime):
@classmethod
def now(cls, tz=None):
return clock.now.astimezone(tz) if tz else clock.now.replace(tzinfo=None)
monkeypatch.setattr("promptise.runtime.budget.datetime", FrozenDatetime)
return clock
async def test_sixth_reminder_in_a_run_is_refused(agent):
calls = [("send_reminder", {"invoice_id": f"INV-{n}"}) for n in range(1, 7)]
agent.model.script = [calls, "Sent what the budget allowed."]
await agent.tick()
assert agent.outbox() == ["INV-1", "INV-2", "INV-3", "INV-4", "INV-5"]
tool_results = [m.content for m in agent.model.seen[-1] if m.type == "tool"]
assert any("was not executed" in str(text) for text in tool_results)
assert agent.process.status()["suspend_reason"] == "budget"
async def test_one_run_per_budget_day(clock, agent): # clock first: the budget starts frozen
denied = []
async def record(ctx: HookContext) -> None:
denied.append(ctx.data)
agent.process.hooks.register(HookEvent.PERMISSION_DENIED, record)
agent.model.script = ["Nothing overdue."]
await agent.tick() # 07:00 UTC, day 1
clock.now = datetime(2026, 10, 22, 5, 59, tzinfo=timezone.utc) # still day 1's budget
result = await agent.tick()
assert result["outcome"] == "success" # the event is consumed...
assert agent.process.status()["invocation_count"] == 1 # ...but the agent did not run
assert agent.process.status()["suspend_reason"] == "budget"
assert [d["limit_name"] for d in denied] == ["max_runs_per_day"]
assert "process.suspended" in agent.event_types()
async def test_budget_day_starts_at_six_utc(clock, agent):
agent.model.script = ["Nothing overdue.", "Nothing overdue."]
await agent.tick() # 07:00 UTC, day 1
clock.now = datetime(2026, 10, 22, 6, 0, tzinfo=timezone.utc)
await agent.tick()
assert agent.process.status()["invocation_count"] == 2
assert "budget.daily_reset" in agent.event_types()test_budget.py::test_sixth_reminder_in_a_run_is_refused PASSED [ 33%]
test_budget.py::test_one_run_per_budget_day PASSED [ 66%]
test_budget.py::test_budget_day_starts_at_six_utc PASSED [100%]
============================== 3 passed in 2.39s ===============================How the clock works: promptise.runtime.budget calls datetime.now(timezone.utc), and the fixture swaps that module's datetime for a subclass whose now() returns clock.now. Nothing else is patched, so asyncio and timeouts keep real time. Put clock before agent in the test's arguments: the budget records its start time when the process is created, and it must already see the frozen clock. If you prefer a library, the time_machine fixture from the time-machine package moves datetime.now() for the whole program, cron triggers and journal timestamps included, and asyncio keeps working because it uses the monotonic clock.
What the three tests pin down:
The cap inside a run. Six send_reminder calls in one turn, five sent. The sixth call's result, which the model saw, says the call "was not executed". The process is then suspended with reason budget.
A refused run. At 05:59 the next morning the budget day hasn't rolled over yet, so the second run is refused by max_runs_per_day. The permission_denied hook gets limit_name="max_runs_per_day", and the process is suspended. Note the outcome: the event is consumed with outcome="success" and is not dead-lettered, even though the agent didn't run. Assert on invocation_count or permission_denied, not on the outcome.
The reset hour. At 06:00 the budget day rolls over, the run is allowed, and budget.daily_reset is emitted.
[11]
Step 7: Test code that reads the journal
Code around an agent often reads its journal: an ops report, a dashboard, an alert. Here, ops_report.py has reminders_sent(journal), which returns the invoice IDs from the journal's send_reminder calls. Test code like that without a process, using InMemoryJournal:
"""Code that reads the journal, tested with an InMemoryJournal."""
from promptise.runtime.journal import InMemoryJournal, JournalEntry
from ops_report import reminders_sent
async def test_ops_report_reads_the_journal():
journal = InMemoryJournal()
for tool, args in [("list_overdue", {}), ("send_reminder", {"invoice_id": "INV-7"})]:
await journal.append(JournalEntry(process_id="reminders", entry_type="tool_call",
data={"tool": tool, "args": args}))
assert await reminders_sent(journal) == ["INV-7"]test_report.py::test_ops_report_reads_the_journal PASSED [100%]
============================== 1 passed in 0.01s ===============================InMemoryJournal implements the same append, read, checkpoint and last_checkpoint methods as FileJournal, so code written against one works with the other. A process can use one too, with JournalConfig(backend="memory"). That suits tests that only need a journal to exist, for example to call rewind(). The process keeps that journal object to itself, though, so when a test needs to read entries, a file journal under tmp_path is the better choice, as in step 3.
[12]
The complete example
reminder-agent/
├── reminder_agent.py # make_config(): the production process config (step 1)
├── billing_server.py # MCP tools: list_overdue, send_reminder -> $OUTBOX
├── scripted_model.py # a chat model that plays back a script (step 2)
├── conftest.py # model and agent fixtures, the tick() harness (step 3)
├── test_runs.py # side effects, journal, dead letters (step 4)
├── test_hooks.py # the freeze hook alone and wired in (step 5)
├── test_budget.py # per-run cap, frozen clock, reset hour (step 6)
├── ops_report.py # reminders_sent(): reads the journal for the ops report
├── test_report.py # that reader, tested with InMemoryJournal (step 7)
├── test_live.py # one smoke test with the real model
└── pytest.ini # asyncio_mode = auto, the "live" marker$ pytest -m "not live"
============================= test session starts ==============================
collected 9 items / 1 deselected / 8 selected
test_budget.py ... [ 37%]
test_hooks.py .. [ 62%]
test_report.py . [ 75%]
test_runs.py .. [100%]
======================= 8 passed, 1 deselected in 4.54s ========================Eight tests, under five seconds, no API key and no network.
[13]
Running it in production
Keep one live smoke test
A scripted suite can pass while the real model has stopped calling a renamed tool. Add one test that runs the same process with the real model, mark it, and run it nightly instead of on every commit:
"""A smoke test against the real model: run it nightly, not on every commit."""
import pytest
@pytest.mark.live
async def test_real_model_sends_both_reminders(live_agent):
result = await live_agent.tick(timeout=120)
assert result["outcome"] == "success"
assert live_agent.outbox() == ["INV-7", "INV-9"]$ pytest -m live -v
test_live.py::test_real_model_sends_both_reminders PASSED [100%]
======================= 1 passed, 8 deselected in 8.39s ========================The live_agent fixture in conftest.py skips the test when OPENAI_API_KEY isn't set, so a CI job without the key reports it as skipped:
$ env -u OPENAI_API_KEY pytest -m live -v
test_live.py::test_real_model_sends_both_reminders SKIPPED (OPENAI_A...) [100%]
SKIPPED [1] test_live.py:5: OPENAI_API_KEY is not set
======================= 1 skipped, 8 deselected in 0.01s =======================Use the same pattern for other triggers
Nothing in the harness is specific to cron. To test an agent that a webhook, a file watcher or an event bus wakes up, inject an event shaped like the one that trigger produces, and the rest of the test stays the same:
Trigger | trigger_type | payload the agent sees |
|---|---|---|
Cron | "cron" | scheduled_time, cron_expression, timezone |
Webhook | "webhook" | The JSON body of the request |
File watch | "file_watch" | path, filename, event_type ("created", "modified", ...), event_types |
Event bus | "event" | event_type, event_id, source, data |
Injected events skip the trigger itself, including its filter expression and, for webhooks, signature checks. Test those parts separately: send a signed and an unsigned request to the webhook in one test, and check a filter expression against sample payloads in another. A file-watch event passes through file_changed hooks on its way in, just as a task_created hook saw the frozen ticks in step 5.
Run the fast suite on every commit
Run pytest -m "not live" in CI with no model key in the environment. The suite then can't reach a provider, and a missing fixture fails fast. For quality checks on many cases, add evals to the same pipeline as described in AI agent evals in CI. For tests of prompts and single LLM calls, see testing LLM prompts and agents.
Test every rule you depend on
Write one test for each safety rule in the config, and make it fail first: comment out the rule and watch the test go red. Good candidates are caps on irreversible tools (AI agent irreversible actions), freeze switches, idempotency ("a second tick for the same slot sends nothing"), and recovery after a restart (crash recovery).
Keep tests isolated
One process per test, all files under tmp_path, no shared journal directory. A shared journal would make the second test recover the first test's conversation and budget counters, because recovery on start is on by default.
[14]
Common problems and fixes
Symptom | Cause | Fix |
|---|---|---|
ModelSetupError at process.start() | The test doesn't use the model fixture, and no key is set | Request model (or agent) in the test |
A test that asserts right after inject() fails sometimes | The run happens in a background worker | Wait for task_completed, as tick() does |
tick() times out | A task_created hook blocked the event, so no task ran | Call inject() and check dead_letters instead |
A finished event shows up in dead_letters after teardown | stop() arrived while task_completed hooks were running (1.5.0) | Yield briefly before stop(), and assert before teardown |
events is empty or misses the last event | Events are delivered in the background | await notifier.flush() before asserting |
The journal has no tool_call entries | The journal level is checkpoint | Use level="full" |
Budget tests pass or fail depending on the time of day | The budget clock isn't frozen, or it was frozen after the process was created | Request clock before agent |
A real tick fires during a test | The production schedule is due | Pass a schedule such as |
[15]
Limitations
Scripted models test plumbing, not judgment. A green suite doesn't mean the real model behaves.
The model swap patches an internal call. ProcessConfig.model takes a name, so the fixture patches promptise.agent.build_agent, as Promptise's own tests do. A future release may offer a public hook.
The clock fixture covers the budget only. Cron triggers and journal timestamps keep real time unless you use a library such as time-machine.
No public journal accessor. Read a file journal from tmp_path, or test journal readers with InMemoryJournal.
A run refused by `max_runs_per_day` reports `outcome="success"`. Assert on invocation_count or the permission_denied hook.
[16]
Frequently asked questions
How do I test a scheduled AI agent without waiting for the schedule?
Start the agent's real process in the test and inject the event its trigger would produce. In Promptise, await process.inject(TriggerEvent(trigger_type="cron", payload={...})) puts the tick on the same queue a cron trigger uses. Give the process a schedule that can't fire during the test, wait for the run's task_completed hook, then assert on its effects.
How do I unit test an AI agent without calling the LLM?
Replace the model with a scripted chat model that returns tool calls and answers from a list. The agent's tools, budget checks, approvals and hooks stay real, so you test everything except the model's judgment. In Promptise, wrap promptise.agent.build_agent in a pytest fixture so every agent built in the test uses the scripted model.
How do I wait for a background agent run to finish in a test?
Register a task_completed hook that puts its data on an asyncio.Queue, inject the event, and await the queue with a timeout. The hook fires after the run and its follow-up steps, such as budget actions, have finished. Avoid fixed sleeps: they make tests slow when they are long and flaky when they are short.
How do I test daily limits and other time-based behavior?
Freeze the clock the code under test reads. For Promptise budgets, replace datetime in promptise.runtime.budget with a subclass whose now() returns a time the test sets, then run ticks before and after the reset hour. Libraries such as time-machine move the clock for the whole program, and asyncio keeps working with either approach.
Should automated tests use the real LLM?
Most of them shouldn't. Real model calls are slow, cost money and vary between runs, which makes failures hard to read. Use a scripted model for the suite that runs on every commit, one or two live smoke tests nightly, and evals over many cases to measure the model's quality.
How do I test that a failed agent run is dead-lettered?
Make the scripted model raise an exception, inject a tick and wait for task_completed. With no retries configured, the outcome is "failed" and process.dead_letters holds the event with reason "retries exhausted" and the error text. With trigger_delivery.max_retries set, assert on the "retrying" outcome first.
[17]
Next steps
Build the scheduled agent itself: AI agent cron jobs.
Measure the model's quality: how to evaluate AI agents and AI agent evals in CI.
Handle failed runs: AI agent retries and dead letter queues.
Write the hooks you test here: AI agent lifecycle hooks.
Cap what a run may do: AI agent irreversible actions.
Reference: agent processes, lifecycle hooks, journal backends and autonomy budget in the Promptise docs.