Test Python AI Agent Before Production Scheduling

An agent that works in your terminal can still fail as a scheduled job. Your shell may provide a missing secret. An exception may produce the wrong exit code. In-memory state may resist serialization. The agent may finish without writing the file its consumer expects.

Before enabling cron, test the process contract—not just the model output. This guide builds a local-to-production test harness around four blocking failures: missing secrets, incorrect exit codes, unserializable state, and missing output artifacts.

Define a local-to-production test contract for your Python AI agent

A scheduled agent is a process with a contract. Define that contract before writing tests.

At minimum, specify:

• The Python entry point.
• Required environment variables.
• Which exit codes mean success or failure.
• The persistent state format.
• The exact output path.
• The output schema and required metadata.
• What gets written to stdout and stderr.

This complements the build and runtime boundaries described in the anatomy of an AI job deployment.

Consider a competitor-monitoring agent. It reads targets, calls an external API, stores a checkpoint, and writes a JSON report. For this example, use the following project-specific contract, including its chosen exit-code mapping:

Entry point:       python agent.py
Required secret:   AGENT_API_KEY
Input:             TARGETS_JSON
State path:        STATE_PATH
Artifact path:     OUTPUT_PATH
Success:           exit code 0 and valid artifact
Invalid config:    exit code 2 and useful stderr
Upstream failure:  exit code 3 and useful stderr
Unexpected error:  exit code 1 and useful stderr
State format:      JSON object
Artifact format:   JSON object with generated_at and items

These codes are choices made for this example, not universal meanings for exit codes. Implement the chosen contract explicitly instead of letting library exceptions define it. The production path must serialize and restore the state before writing its checkpoint so that the serialization boundary is enforced by the agent itself:

import json
import os
import sys
from datetime import datetime, timezone
from pathlib import Path

from state import restore_state, serialize_state


class ConfigurationError(Exception):
    pass


class UpstreamError(Exception):
    pass


def load_api_key() -> str:
    value = os.getenv("AGENT_API_KEY", "").strip()

    if not value:
        raise ConfigurationError("AGENT_API_KEY is required")

    # Example project-specific validation rule.
    if not value.startswith(("test_", "prod_")):
        raise ConfigurationError("AGENT_API_KEY has an invalid format")

    return value


def fetch_items(api_key: str, targets: list[str]) -> list[dict]:
    mode = os.getenv("API_MODE", "fake")

    if mode == "failure":
        raise UpstreamError("Upstream request failed")

    if mode == "timeout":
        raise TimeoutError("Upstream request timed out")

    if mode == "unexpected":
        raise RuntimeError("Unexpected adapter failure")

    return [{"target": target, "status": "checked"} for target in targets]


def save_json(path: Path, value: object) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(value, indent=2), encoding="utf-8")


def save_serialized_state(path: Path, payload: str) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(payload, encoding="utf-8")


def run() -> None:
    api_key = load_api_key()

    targets = json.loads(os.environ["TARGETS_JSON"])
    if not isinstance(targets, list) or not all(
        isinstance(target, str) for target in targets
    ):
        raise ConfigurationError("TARGETS_JSON must be a list of strings")

    items = fetch_items(api_key, targets)

    state = {
        "completed_targets": targets,
        "next_step": "publish",
    }

    test_mode = os.getenv("AGENT_TEST_MODE", "normal")
    if test_mode == "unserializable_state":
        state["runtime_object"] = object()

    serialized_state = serialize_state(state)
    restore_state(serialized_state)
    save_serialized_state(Path(os.environ["STATE_PATH"]), serialized_state)

    # Controlled subprocess mode used to prove that exit code 0 alone is
    # insufficient when the required output artifact is missing.
    if test_mode == "skip_output":
        return

    artifact = {
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "items": items,
    }
    save_json(Path(os.environ["OUTPUT_PATH"]), artifact)


def main() -> int:
    try:
        run()
        print("Agent completed")
        return 0
    except (ConfigurationError, KeyError, json.JSONDecodeError) as exc:
        print(f"Configuration error: {exc}", file=sys.stderr)
        return 2
    except (UpstreamError, TimeoutError) as exc:
        print(f"Runtime error: {exc}", file=sys.stderr)
        return 3
    except Exception as exc:
        print(f"Unhandled error: {type(exc).__name__}: {exc}", file=sys.stderr)
        return 1


if __name__ == "__main__":
    raise SystemExit(main())

The model response itself may vary. The process contract should not.

Build a pytest harness that matches the scheduled runtime

Import-based unit tests are useful, but they do not reproduce process behavior. A scheduler starts a fresh process. Your integration test should do the same.

Run the entry point with:

• A clean environment.
• An isolated temporary filesystem.
• Controlled external API behavior.
• Captured stdout and stderr.
• The same command used by the scheduler.

Create a subprocess helper in tests/test_agent_process.py:

import json
import os
import subprocess
import sys
from pathlib import Path

import pytest


PROJECT_ROOT = Path(__file__).parents[1]


def run_agent(
    tmp_path: Path,
    *,
    api_key: str | None = "test_local",
    api_mode: str = "fake",
    agent_mode: str = "normal",
    targets: object = None,
) -> tuple[subprocess.CompletedProcess[str], Path, Path]:
    output_path = tmp_path / "artifacts" / "report.json"
    state_path = tmp_path / "state" / "checkpoint.json"

    env = {
        "PATH": os.environ["PATH"],
        "PYTHONPATH": str(PROJECT_ROOT),
        "API_MODE": api_mode,
        "AGENT_TEST_MODE": agent_mode,
        "TARGETS_JSON": json.dumps(
            ["example.com"] if targets is None else targets
        ),
        "OUTPUT_PATH": str(output_path),
        "STATE_PATH": str(state_path),
    }

    if api_key is not None:
        env["AGENT_API_KEY"] = api_key

    result = subprocess.run(
        [sys.executable, str(PROJECT_ROOT / "agent.py")],
        cwd=PROJECT_ROOT,
        env=env,
        capture_output=True,
        text=True,
        timeout=30,
        check=False,
    )

    return result, state_path, output_path

In this example, the subprocess timeout is set to 30 seconds as a test-harness choice. It is not a general timeout recommendation; choose a value appropriate to your agent and test environment.

Using a temporary directory prevents an old report from making a failed test appear successful. A minimal environment also catches hidden dependencies on your interactive shell.

Assert stable properties rather than exact LLM wording. Check types, required fields, allowed values, paths, and process status. Exact text matching is appropriate for deterministic protocol fields, not generated prose.

Gate 1: fail fast when a required secret is absent

Test an unset secret, an empty value, and a malformed value. In this example, each case must stop before external work begins and return the configuration exit code 2 chosen in the contract above.

@pytest.mark.parametrize("api_key", [None, "", "wrong-format"])
def test_invalid_secret_blocks_execution(tmp_path, api_key):
    result, state_path, output_path = run_agent(
        tmp_path,
        api_key=api_key,
    )

    assert result.returncode == 2
    assert "Configuration error" in result.stderr
    assert not state_path.exists()
    assert not output_path.exists()

Do not assert the secret itself in logs. Assert that it is absent:

def test_secret_is_never_printed(tmp_path):
    secret = "test_do_not_log_this"

    result, _, _ = run_agent(
        tmp_path,
        api_key=secret,
        api_mode="failure",
    )

    assert result.returncode == 3
    combined_logs = result.stdout + result.stderr
    assert secret not in combined_logs

The same rule applies to exceptions, request headers, debug output, and serialized state. Inject the secret at runtime. Do not store it in the repository or include it in test fixtures committed to source control.

Your production identity should also receive only the permissions the agent needs. The practical checklist in AI agent security for production Python covers secrets injection, isolation, and least-privilege access.

Gate 2: verify process success and failure exit codes

Schedulers can use process status to determine how a run should be handled. A stack trace with the example's success code 0 would be a false success under this contract. A completed report with a nonzero exit code would be a false failure.

For this example, assert the exact codes selected in the contract: 0 for success, 2 for invalid configuration, 3 for an upstream failure, and 1 for an unexpected exception. Other projects may choose a different mapping, which their tests and scheduler configuration should reflect.

def test_success_returns_zero(tmp_path):
    result, _, output_path = run_agent(tmp_path)

    assert result.returncode == 0
    assert output_path.exists()
    assert "Agent completed" in result.stdout


@pytest.mark.parametrize("api_mode", ["failure", "timeout"])
def test_upstream_failures_return_three(tmp_path, api_mode):
    result, _, output_path = run_agent(
        tmp_path,
        api_mode=api_mode,
    )

    assert result.returncode == 3
    assert "Runtime error" in result.stderr
    assert not output_path.exists()


def test_invalid_input_returns_two(tmp_path):
    result, _, output_path = run_agent(
        tmp_path,
        targets={"unexpected": "object"},
    )

    assert result.returncode == 2
    assert "Configuration error" in result.stderr
    assert not output_path.exists()


def test_unexpected_exception_returns_one(tmp_path):
    result, state_path, output_path = run_agent(
        tmp_path,
        api_mode="unexpected",
    )

    assert result.returncode == 1
    assert "Unhandled error: RuntimeError" in result.stderr
    assert not state_path.exists()
    assert not output_path.exists()

Keep stderr actionable. Include the failure category and operation. Exclude secrets and full sensitive payloads. In this implementation, main() owns the project-specific mapping from exceptions to process exit codes so that every code path follows the same chosen policy.

Gate 3: prove agent state can be serialized and restored

Agents often keep live objects in memory: API clients, generators, callbacks, file handles, or framework-specific message classes. Those objects may work during one process but fail when you checkpoint state for a later run.

Make serialization part of the state boundary in state.py:

import json
from typing import Any


def serialize_state(state: dict[str, Any]) -> str:
    return json.dumps(state)


def restore_state(payload: str) -> dict[str, Any]:
    value = json.loads(payload)

    if not isinstance(value, dict):
        raise ValueError("State must be a JSON object")

    if "next_step" not in value:
        raise ValueError("State is missing next_step")

    return value

Test both valid and invalid state in tests/test_state.py:

import pytest

from state import restore_state, serialize_state


def test_state_round_trip():
    original = {
        "completed_targets": ["example.com"],
        "next_step": "publish",
    }

    restored = restore_state(serialize_state(original))

    assert restored == original
    assert restored["next_step"] == "publish"


@pytest.mark.parametrize(
    "invalid_value",
    [
        lambda: None,
        iter(["example.com"]),
        open,
    ],
)
def test_nonserializable_state_is_rejected(invalid_value):
    with pytest.raises(TypeError):
        serialize_state(
            {
                "next_step": "publish",
                "runtime_object": invalid_value,
            }
        )

The isolated unit test verifies the serializer, but the production subprocess must also prove that agent.py applies it before writing the checkpoint. The controlled unserializable_state mode inserts a live object into the actual agent state. Serialization then fails on the production execution path, maps to this example contract's unexpected-error code 1, and prevents both checkpoint and artifact writes:

def test_agent_rejects_nonserializable_state_in_subprocess(tmp_path):
    result, state_path, output_path = run_agent(
        tmp_path,
        agent_mode="unserializable_state",
    )

    assert result.returncode == 1
    assert "Unhandled error: TypeError" in result.stderr
    assert not state_path.exists()
    assert not output_path.exists()

Persist data needed to reconstruct a dependency, not the dependency itself. Store an API request ID instead of a live client. Store a list or cursor instead of a generator. Store a filename instead of an open handle.

After restoration, verify that the next workflow step receives the same required inputs and does not repeat completed side effects. For longer jobs, this aligns with the checkpointing approach in resumable AI agent task design.

Gate 4: validate the output artifact before enabling the schedule

A successful process is not enough under this example's contract. The consumer also needs a valid artifact at the agreed path.

Centralize the complete success contract so tests cannot accidentally treat the chosen success code 0 as sufficient:

def assert_success_contract(result, output_path):
    assert result.returncode == 0
    assert output_path.is_file(), "required output artifact is missing"
    assert output_path.stat().st_size > 0

Test the artifact's existence, content, schema, metadata, and readability:

def test_output_artifact_contract(tmp_path):
    result, state_path, output_path = run_agent(tmp_path)

    assert_success_contract(result, output_path)

    report = json.loads(output_path.read_text(encoding="utf-8"))

    assert isinstance(report, dict)
    assert isinstance(report["generated_at"], str)
    assert isinstance(report["items"], list)
    assert report["items"]

    for item in report["items"]:
        assert set(item) == {"target", "status"}
        assert isinstance(item["target"], str)
        assert item["status"] == "checked"

    checkpoint = json.loads(state_path.read_text(encoding="utf-8"))
    assert checkpoint["next_step"] == "publish"

Also exercise a real failure of the success contract. The controlled skip_output mode lets the agent finish with this example's success code 0 without writing OUTPUT_PATH. The test verifies that the complete contract rejects that run despite the successful process status:

def test_success_contract_rejects_zero_exit_without_artifact(tmp_path):
    result, state_path, output_path = run_agent(
        tmp_path,
        agent_mode="skip_output",
    )

    assert result.returncode == 0
    assert state_path.is_file()
    assert not output_path.exists()

    with pytest.raises(
        AssertionError,
        match="required output artifact is missing",
    ):
        assert_success_contract(result, output_path)

This regression test keeps the suite green while proving that the gate itself fails for a false-success process. Any scheduler smoke-test wrapper using this example contract should apply the same combined rule: success requires both exit code 0 and a valid artifact.

Run the complete suite from a clean checkout:

python -m venv .venv
. .venv/bin/activate
python -m pip install pytest
pytest -q

Then follow a controlled release sequence:

1. Commit the tested code and dependency definitions.
2. Let the deployment pipeline build that exact commit.
3. Inject production secrets through the runtime.
4. Run one unscheduled production smoke test.
5. Inspect its logs, contract-specific exit code, state, and artifact.
6. Enable the schedule only after every gate passes.

Do not “fix” a production smoke test by editing the deployed environment manually. Change the code or configuration in source control, rerun the local suite, and redeploy. That keeps the tested commit aligned with the running job. See the GitHub-to-production deployment pipeline for the surrounding validation and build flow.

Conclusion

Testing model quality is only part of testing an AI agent. In this example, the scheduled Python process must also reject missing secrets, report failures through the exact exit codes chosen for its contract, persist restorable state, and produce the artifact promised to downstream consumers.

Encode those requirements in pytest. Run the agent as a clean subprocess. Then deploy the same commit and execute an unscheduled smoke test before turning on cron.

A managed agent runtime such as HollowHost can handle deployment, secrets injection, logs, and scheduling. Your test contract still matters. It defines what the platform should run—and what counts as a successful agent job.