Ak Dev Testing Conventions

Test Patterns

Dummy Implementations for Unit Testing

Create minimal implementations of abstract classes:

from agentkernel.core.base import Agent, Runner, Session
from agentkernel.core.model import AgentReplyText, AgentRequest, AgentRequestText


class DummyRunner(Runner):
    async def run(self, agent, session, requests):
        prompt = requests[0].prompt if isinstance(requests[0], AgentRequestText) else ""
        return AgentReplyText(response=f"ok:{prompt}")

    async def stream(self, agent, session, requests):
        # Runner.stream() is abstract — implement even in test doubles, and yield `StreamEvent`
        # members (never a bare `str` — `StreamChunk.event` rejects it with a `ValidationError`).
        # Raise NotImplementedError() (with a trailing `yield`) if the test doesn't exercise streaming,
        # or yield MessageStart/TextDelta/MessageEnd (own the message's boundaries yourself) to test
        # Runtime.stream() / AgentService.stream_multi().
        raise NotImplementedError()
        yield


class DummyAgent(Agent):
    def __init__(self, name="test-agent"):
        runner = DummyRunner("DummyRunner")
        super().__init__(name, runner)

    def get_description(self) -> str:
        return "Test agent"

    def get_a2a_card(self):
        return None

A double that does exercise streaming yields events, not strings. Runner.stream() is typed AsyncGenerator[StreamEvent, None], and StreamChunk.event is a discriminated union — so a bare str raises a ValidationError the moment Runtime.stream() wraps it. A double owns its own boundaries:

from agentkernel.core.event import MessageEnd, MessageStart, TextDelta


class StreamingDummyRunner(Runner):
    async def run(self, agent, session, requests):
        return AgentReplyText(response="ok")

    async def stream(self, agent, session, requests):
        yield MessageStart(message_id="m-1")
        for token in ("Hel", "lo"):
            yield TextDelta(message_id="m-1", content=token)
        yield MessageEnd(message_id="m-1")

Every event reaches PostHook.on_stream_event(), but only TextDelta is projected into StreamChunk.delta — so a test that accumulates the reply must filter on chunk.delta is not None rather than slice by position. (delta is a model field, so the attribute always exists; it is None on every non-text frame. Key presence is the wire-format rule, which applies to the serialised SSE frames, not to the objects a test sees.)

A test double for a hook implements on_stream_event, and a double that returns a list ends the chain for that event — so a two-hook test asserting what the second hook saw is the way to pin that rule. See RecordingHook and HaltingHook in tests/test_runtime_stream_events.py.

Async Test Patterns

Use @pytest.mark.asyncio for async tests:

import pytest

@pytest.mark.asyncio
async def test_runtime_run():
    runtime = Runtime(InMemorySessionStore())
    agent = DummyAgent()
    runtime.register(agent)
    session = runtime.sessions().new("test-session")

    result = await runtime.run(agent, session, [AgentRequestText(prompt="hello")])
    assert result.response == "ok:hello"

Monkeypatching Config

Use monkeypatch to override AKConfig for tests:

def test_redis_session_store(monkeypatch):
    class FakeCfg:
        class session:
            type = "redis"
            cache = None
            class redis:
                url = "redis://localhost:6379"
                ttl = 60
                prefix = "ak:test:"

    monkeypatch.setattr("agentkernel.core.config.AKConfig.get", classmethod(lambda cls: FakeCfg))
    store = SessionStoreBuilder.build()
    assert isinstance(store, RedisSessionStore)

Session Context Tests

Test the async context manager pattern:

@pytest.mark.asyncio
async def test_session_context():
    session = Session("test-id")

    async with session:
        current = Session.current()
        assert current is session
        assert current.id == "test-id"

    # Outside context, no current session
    assert Session.current() is None

Testing Volatile vs Non-Volatile Caches

@pytest.mark.asyncio
async def test_volatile_cache_cleared():
    session = Session("test-id")

    async with session:
        session.get_volatile_cache().set("key", "value")
        assert session.get_volatile_cache().get("key") == "value"

    # Volatile cache is cleared after Runtime.run() completes
    # Non-volatile cache persists

Testing Hooks

@pytest.mark.asyncio
async def test_pre_hook_modifies_request():
    class TestPreHook(PreHook):
        async def on_run(self, session, agent, requests):
            for req in requests:
                if isinstance(req, AgentRequestText):
                    req.prompt = req.prompt.upper()
            return requests
        def name(self): return "test_hook"

    agent = DummyAgent()
    agent.pre_hooks.append(TestPreHook())
    # ... run through Runtime and verify modified input


@pytest.mark.asyncio
async def test_pre_hook_halts_execution():
    class BlockingHook(PreHook):
        async def on_run(self, session, agent, requests):
            return AgentReplyText(response="blocked", prompt="")
        def name(self): return "blocking_hook"

    # When a PreHook returns AgentReply, Runtime.run() returns it immediately
    # without calling the agent's runner

Built-in Test Framework

Agent Kernel provides a Test class (ak-py/src/agentkernel/test/) for integration testing. This framework is used in examples and can be used for testing deployed agents as well.

from agentkernel.test import Test

# In test files
@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def test_client():
    test = Test("demo.py")       # Path to agent definition file
    await test.start()
    try:
        yield test
    finally:
        await test.stop()

@pytest.mark.order(1)
async def test_agent_response(test_client):
    await test_client.send("Who won the 1996 cricket world cup?")
    await test_client.expect(["Sri Lanka won the 1996 cricket world cup."])

Test Modes

Configured via test-config.yaml — a separate, un-nested file resolved from the cwd (or AK_TEST_CONFIG_PATH_OVERRIDE), loaded only when the test harness runs. It is not part of config.yaml; a leftover test: section there is ignored:

mode: score    # score | llm | fallback (default: fallback)
evaluator: deepeval   # built-in short name ('deepeval', 'opik' or 'jev'), or a dotted path to your own AKEvaluator subclass
llm:
  model: gpt-4o-mini
  provider: openai

• score: Deterministic, offline scoring via the configured AKEvaluator's evaluate_by_score — the built-in DeepevalAKEvaluator uses Scorer.quasi_exact_match_score (normalised whole-string equality, no LLM call); the built-in OpikAKEvaluator uses Opik's LevenshteinRatio instead (graded fuzzy-similarity, not exact-match); the built-in JevAKEvaluator has no score mode (AKMetricNotSupported), so evaluator: jev needs mode: llm
• llm: LLM-as-judge scoring via evaluate_by_llm — deepeval and opik use a GEval metric against the expected answer(s) (ground truth), while jev asks one TypeSafe Noul yes/no question (its probability is the score; comparison text is sent to api.typesafe.ai): DeepevalAKEvaluator via DeepEval's GEval and an LLMTestCase, OpikAKEvaluator via Opik's GEval and a single packed output string. Evaluation backends are pluggable (agentkernel.test.core.evaluator.AKEvaluator); see ak-py/src/agentkernel/test/test.py
• fallback: Tries score first, falls back to llm if score fails

evaluator is resolved by Test._resolve_evaluator (agentkernel/test/test.py), the same bring-your-own-dotted-path pattern resolve_dotted/require_extra (core/util/factory.py) use for session stores, sandbox providers, and trace backends — a custom evaluator subclasses AKEvaluator (agentkernel.test.core.evaluator) and implements evaluate_by_score/ evaluate_by_llm, each returning an AKEvaluationResult. See examples/cli/custom-evaluator/ for a full worked example, and docs/specs/555-pluggable-test-evaluators/ for the design/spec behind this interface.

Test.compare/Test.expect also take return_metrics: bool = False — when True, they return the AKEvaluationResult instead of raising AssertionError on a failing comparison (other errors, e.g. AKEvaluationError/AKMetricNotSupported, still propagate).

Test.compare() for API Tests

For HTTP API tests, use Test.compare():

response = await http_client.send("What is 2+2?")
Test.compare(response, ["4", "The answer is 4"])

HTTP API Integration Tests

Pattern for testing deployed agents:

class APITestClient:
    def __init__(self, url):
        self.url = url
        self.session_id = str(uuid.uuid4())

    async def send(self, prompt, endpoint=""):
        payload = {
            "prompt": prompt,
            "session_id": self.session_id,
            "agent": "triage"
        }
        async with httpx.AsyncClient(timeout=30.0) as client:
            resp = await client.post(f"{self.url}{endpoint}", json=payload)
            resp.raise_for_status()
            return resp.json().get("result", "")

@pytest_asyncio.fixture(scope="session", loop_scope="session")
async def http_client():
    endpoint = os.getenv("AK_TEST_ENDPOINT")
    yield APITestClient(endpoint)

CI/CD Workflows

• test.yaml: Triggers on pull requests, pushes to develop, and manual dispatch; has an update-lock-files job (dispatch-only) and a run-tests job that delegates to test-reusable.yaml
• test-reusable.yaml: Reusable workflow (workflow_call) containing the actual test jobs, including the uv run pytest invocation and the transport-integration-tests job, which starts real NATS and Kafka containers from the transport examples' compose files (broker services only, up -d --wait nats|kafka) and runs tests/test_transport_contract_live.py against them on every PR. build-ak-py hands ak-py/dist to unit-tests and e2e-tests as a run-scoped artifact (actions/upload-artifact/download-artifact), deliberately not actions/cache: the workflow is called with an arbitrary checkout_ref from privileged contexts (workflow_dispatch on develop, pull_request_target for labelled fork PRs), where a cache write is a poisoning vector the default branch would later restore (CodeQL actions/cache-poisoning). Don't "optimise" it back into a cache
• chart-test.yaml: The #495 Helm chart gates, path-filtered on the chart, the k8s example, and agentkernel/pipeline/: a lint job (ct lint via ak-deployment/ak-k8s/ci/ct.yaml, plus explicit helm template renders of every flavor values file and optional tier) and a kind-smoke matrix over the dev/baremetal/eks flavors, each installing the chart with the example images (the dev flavor through examples/k8s/openai-queue-mode/deploy/deploy.sh local, the example's own install script, so the user-facing path is what CI exercises against the checkout's chart) and driving one real chat request through NATS (baremetal/eks install the Gateway API CRDs and a live NACK controller so autoProvision: false verifies operator-reconciled objects; the CI overlays in ak-deployment/ak-k8s/ci/ swap only storage class and sizing). Skipped for fork PRs (needs OPENAI_API_KEY)
• test-trusted-pr.yaml: Runs test-reusable.yaml with secrets for fork PRs that have been reviewed and labeled safe-to-test (pull_request_target)
• test-github-app.yaml: Manual dispatch only; verifies the GitHub App secrets (APP_ID/APP_PRIVATE_KEY) are configured correctly
• integration-test.yaml: "Nightly" (tier nightly) integration tests against deployed environments; scheduled weekly on Sundays at 5:30 PM UTC (cron: '30 17 * * 0'), plus manual dispatch
• integration-test-weekly.yaml: Weekly integration tests against deployed environments (cron currently commented out; manual dispatch, with option to keep cloud resources on failure), plus two additional manual-dispatch-only jobs gated on inputs.provision_e2e_messaging: e2e-messaging-deploy (builds and Terraform-applies the e2e/app messaging harness to AWS ECS, then waits for the service to stabilize) and e2e-messaging-test (probes the deployed webhook, registers the Telegram webhook, then runs the e2e/tests pytest suite against Slack/Telegram/WhatsApp/Messenger/Instagram/Gmail). See e2e/README.md for the full harness design.
• code-quality.yml: Runs linting checks (see code-quality skill)
• pr-title-check.yaml: Fails the PR unless its title follows Conventional Commits (type: description / type(scope): description, types in CONTRIBUTING.md). Runs on pull_request open/edit/synchronize/reopen/ready-for-review so a result exists for every head SHA (required checks are recorded per SHA)
• copilot-review-request.yaml: Requests a Copilot code review on open/reopen/ready-for-review using the COPILOT_REVIEW_PAT secret (pull_request_target, API-only, no checkout): an org-scoped fine-grained PAT (resource owner yaalalabs, this repo, Pull requests: Read and write) created by a licensed maintainer. It is separate from COPILOT_REQUEST_TOKEN because the account-level Copilot Requests permission only exists on user-owned tokens, which cannot hold org repository permissions. Exists because the develop ruleset's copilot_code_review rule only fires for authors who hold a Copilot license. Only collaborator-authored PRs are requested automatically: the script calls repos.checkCollaborator (needs only the Metadata read every fine-grained PAT carries) rather than trusting the payload's author_association, because each review is a premium request billed to the PAT owner. Also skips PRs where Copilot is already requested and bot-authored PRs. workflow_dispatch takes a PR number, bypasses the collaborator check, and is how a maintainer requests a review on an outside contributor's PR after reading it (or backfills one)
• reviewed-label-reset.yaml: Removes the maintainer-applied Reviewed label whenever new commits are pushed (pull_request_target: synchronize, API-only, no checkout) so the PR reappears in the -label:Reviewed review queue

Every workflow declares an explicit least-privilege permissions block (CodeQL actions/missing-workflow-permissions); jobs that write repo content (commits, branches, PRs) do so with the GitHub App token minted in-job, not with GITHUB_TOKEN. Metadata-only writes such as the label removal in reviewed-label-reset.yaml use a job-scoped GITHUB_TOKEN (pull-requests: write) instead of minting an App token. When adding a workflow, start from permissions: {} (or contents: read if it checks out) and widen only where a step provably needs it. Workflows on pull_request_target (test-trusted-pr.yaml, copilot-review-request.yaml, reviewed-label-reset.yaml) must never check out or execute PR code unless gated behind a maintainer-applied label, as test-trusted-pr.yaml is.

scripts/update_examples_version.py keeps the agentkernel pin in sync across both examples/** and e2e/app (via --examples-dir e2e/app): publish.yaml's "Update e2e messaging harness with new version" step bumps the pin with --skip-lock right after a production publish (the just-published version isn't resolvable on PyPI yet, so the lock refresh is deferred), and test.yaml's "Update e2e messaging harness lock file" step later runs --force-lock --examples-dir e2e/app to regenerate e2e/app/uv.lock and commits it alongside examples/**/uv.lock. If you add a new pinned dependency consumer under e2e/app, make sure it's covered by this same version-bump/lock-refresh pair instead of drifting out of sync with the published package.

Both integration workflows restore the branch-built agentkernel wheel from the ak-py-${{ github.sha }} cache with fail-on-cache-miss: true — the job fails loudly instead of silently falling back to the published PyPI wheel if the build/cache step didn't run first. .github/scripts/run_single_test.py's test_aws_deployment() then runs ./build.sh local in the example directory (force-reinstalling the local wheel with --no-cache-dir before packaging/deploying) and invokes the test client with uv run --no-sync pytest ... so uv doesn't re-sync the venv from uv.lock and revert the local wheel back to the PyPI version. When adding a new example to integration-test-config.yaml, make sure its build.sh/deploy.sh local branch force-reinstalls agentkernel from ../../../ak-py/dist with --no-cache-dir, matching this pattern — otherwise the test can silently exercise a stale published version instead of the branch's code. The Kubernetes examples apply the same split to the chart: deploy/deploy.sh installs the published OCI chart at the pinned release version, deploy/deploy.sh local installs ak-deployment/ak-k8s/chart from the checkout, and both examples/sandbox/broker-nats/app_test.py and chart-test.yaml's dev-flavor smoke call the local form.

integration-test-weekly.yaml's AWS matrix jobs reuse the shared VPC/subnets/security groups created by the aws-serverless base deployment (examples/aws-serverless/openai, itself queue_mode = true with a request handler, agent runner, and response handler tier) instead of each creating its own: get-base-outputs runs .github/scripts/get_base_outputs.py --base-path examples/aws-serverless/openai --deploy-dir deploy to read vpc_id/private_subnet_ids plus the three per-tier *_security_group_id terraform outputs, and each matrix job's Deploy/Destroy steps forward them as run_single_test.py --vpc-id --private-subnet-ids --request-handler-security-group-id (every aws-serverless job — the request handler tier is also shared by the authorizer and WebSocket connection handler Lambdas) and additionally --agent-runner-security-group-id --response-handler-security-group-id (only scalable-openai/schedule-openai, the two other weekly-matrix examples with those tiers). These become TF_VAR_request_handler_security_group_id/TF_VAR_agent_runner_security_group_id/TF_VAR_response_handler_security_group_id, which the example's own deploy/variables.tf (default null) and deploy/main.tf (security_group_id = var.*_security_group_id) must declare and forward into the corresponding serverless_agents module object (request_handler/agent_runner/response_handler), per docs/specs/716-reuse-sg-in-integration-test-pipeline/. When adding a new aws-serverless example to the weekly matrix, wire its deploy/variables.tf/main.tf the same way for whichever tiers it has, or the job will keep creating (and paying for/leaking) its own SGs instead of reusing the base's.

integration-test-weekly.yaml's deploy/test/destroy are separate workflow steps (not one bash block) so each shows up as its own status: the Test step only runs if: steps.deploy.outcome == 'success', and Destroy runs if: !cancelled() && !(keep_resources_on_failure && (deploy or test failed)). .github/scripts/generate_test_matrix.py assigns each gcp-* matrix entry a deploy_stagger (90s apart) so parallel GCP jobs don't provision VPC connectors on the same network simultaneously; the workflow sleeps that many seconds before deploying. Known infra-flakiness mitigations to preserve when touching this area, since they were added specifically to fix recurring weekly e2e failures:

• AWS containerized examples' deploy.sh call wait_for_ecs_stable after terraform apply (reads region/prefix from terraform.tfvars, then aws ecs wait services-stable) so the test doesn't hit the app before the ECS service has finished rolling out.
• run_single_test.py's destroy_aws_resources pre-deletes Lambda functions on the example's security groups and runs a background thread that periodically deletes their now-detached ENIs, so terraform destroy isn't blocked waiting ~20 min for AWS to release Hyperplane ENIs.
• run_single_test.py's deploy_gcp_resources/destroy_gcp_resources retry the deploy up to 3 times and sweep ERROR-state VPC Access Connectors between attempts (sweep_gcp_error_connectors, matched by network name from terraform.tfvars).
• Azure serverless/containerized deploys accept an AK_PRE_DEPLOY_AUTH_CMD env var (set in the workflows to refresh the OIDC-based az login) and re-run it before each long-running az step in linux_function.tf's deploy_function_code, since the OIDC token can expire mid-deploy.

If a weekly/nightly integration run fails with a symptom matching one of these (ECS "still rolling out", GCP VPC_ACCESS_CONNECTOR_ERROR, AWS destroy timing out on ENIs, or Azure CLI auth expiring), look here before re-adding ad-hoc sleeps or retries.

Best Practices

1. Use ordered tests (@pytest.mark.order(n)) when testing conversational flows where follow-up questions depend on prior context
2. Use session-scoped fixtures for test clients that are expensive to create
3. Mock external services (LLM APIs, cloud services) in unit tests — only hit real APIs in integration tests
4. Test both success and failure paths — especially for hooks and guardrails
5. Use DummyAgent/DummyRunner to isolate the component under test from framework-specific behavior
6. Test session persistence — verify that state survives across multiple Runtime.run() calls within the same session