What is ak-dev-testing-conventions?

A guide to the testing patterns and tools used to check Agent Kernel code. It explains how to run tests, test asynchronous code, replace outside dependencies in tests, and use the project's test framework.

Use it when adding features, debugging failures, checking runtime and streaming behavior, or running the test suite in continuous integration.

It helps contributors follow the repository's existing testing approach and investigate failing tests consistently.

ak-dev-testing-conventions — Skill by yaalalabs

Install

Getting it into your agent

One page per mod, every tool's command on it. A separate URL per tool would split the same page into five that compete with each other.

agentmods

npx agentmods add skills/yaalalabs/agent-kernel/ak-dev-testing-conventions

Any agent

npx skills add yaalalabs/agent-kernel --skill ak-dev-testing-conventions

Clone the repo

git clone --depth 1 https://github.com/yaalalabs/agent-kernel

Made for: Claude Code, Codex.

Per session 59 Skills are progressive disclosure: only the name and description are preloaded; the body loads when the skill is used.

When invoked 7,131 The whole file, excluding the scripts and references it only reads on demand.

Security scan A 0 findings. Scan, not verified.

Origin original No closer match found in the catalogue.

Token cost

What it costs to keep this loaded

Counted locally with the o200k_base tokenizer, which is exact for GPT models; Claude uses its own tokenizer and its counts differ. Treat this as one consistent yardstick across the catalogue rather than a bill. Prices are per million input tokens.

Model	Per session	Once invoked
Fable 5	$0.00059	$0.07131
Opus 5	$0.00030	$0.03565
Sonnet 5	$0.00012	$0.01426
Haiku 4.5	$0.00006	$0.00713

Measured 3d ago against content hash c24686580702, method: parsed. Prices are Anthropic first-party input rates as of 2026-08-30, from the pricing page.

Security

Grade A, and why

ak-dev-testing-conventions scanned grade A with 0 findings against 26 rules in 11 categories — prompt injection, anti-refusal, data exfiltration, privilege escalation, supply chain, agent snooping, system-prompt leakage, SSRF and excessive agency — measured 3d ago.

A static scan of the body, not an audit. Every finding is printed with the line that produced it so you can judge whether it matters here. A mod is markdown that instructs an agent; that is exactly why what it instructs is worth reading.

Nothing flagged

None of the 26 patterns this scan looks for appear in this file: no shell pipes, no recursive deletes, no credential paths, no hidden text, no instruction-override or anti-refusal phrasing, no agent-config snooping. That is not a guarantee, it is the absence of the things that are checkable.

.agents/skills/ak-dev-testing-conventions/SKILL.md · 399 lines

How it starts

The opening of the file, as written. The whole thing — 399 lines — stays where its author put it; the contents beside it link to each section on GitHub.

Testing Conventions

Running Tests

cd ak-py
uv run pytest                           # Run all tests with coverage
uv run pytest tests/test_runtime.py     # Run specific test file
uv run pytest -k "test_session"         # Run tests matching pattern
uv run pytest -x                        # Stop on first failure

Coverage and HTML reports are auto-generated per pyproject.toml:

[tool.pytest.ini_options]
addopts = "--cov=src --cov-report=term --cov-report=html --html=report.html"

Test File Organization

Tests live in ak-py/tests/ and follow the naming convention test_<module>.py:

Test File	Tests
`test_base.py`	Session, Agent, Runner abstractions
`test_runtime.py`	Runtime registration, execution, hooks
`test_stream_events.py`	`core/event.py`'s `StreamEvent` discriminated union: every member round-trips through JSON (`type` discriminator), rejects an unknown `type`, and stays JSON/pickle-safe (no framework-native fields)
`test_runtime_stream_events.py`	`Runtime.stream()`'s streaming contract (spec `docs/specs/523-ag-ui-support/`): legacy `str` yields normalised into a `MessageStart`/`TextDelta`/`MessageEnd` sequence, `PostHook.on_stream_chunk()` only sees `TextDelta`/`ReasoningDelta` content and a hook's edit is written back into the event, a hook returning `None` drops the whole chunk, `delta` is populated only for `TextDelta`, and the final chunk is a bare `StreamChunk(done=True)`
`test_module.py`	Module load/unload, wrapping
`test_session.py`	Session state, caches, context vars
`test_session_cache.py`	LRU SessionCache
`test_sessions_in_memory.py`	InMemorySessionStore
`test_sessions_redis.py`	RedisSessionStore missing-config error, shared RedisDriver retry exhaustion
`test_sessions_valkey.py`	ValkeySessionStore round trips (fake client), shared ValkeyDriver retry exhaustion
`test_sessions_dynamodb.py`	DynamoDBSessionStore Binary wrap/unwrap, missing-item skip (mocked driver)
`test_shared_drivers.py`	Shared DB drivers (`core/util/driver/`): retry scope, ping/reconnect, command surface, DynamoDB item-dict semantics
`test_multimodal_redis_store.py`	RedisAttachmentStore index TTL refresh, JSON round trip, pruning (mocked driver)
`test_multimodal_source_forms.py`	`MultimodalPreHook` attachment source-form classification (spec #523 §8): bare base64 and base64 `data:` URIs are described/stored/stripped; `http(s)://`/`s3://` and non-base64 `data:` URIs are retained undescribed; empty `data:` payloads are dropped
`test_config.py`	AKConfig loading, env vars
`test_test_config.py`	AKTestConfig (Test framework config) loading, defaults
`test_tool.py`	ToolContext, cache
`test_tool_openai.py`	OpenAI ToolBuilder
`test_tool_crewai.py`	CrewAI ToolBuilder
`test_tool_langgraph.py`	LangGraph ToolBuilder
`test_tool_adk.py`	Google ADK ToolBuilder
`test_tool_smolagents.py`	Smolagents ToolBuilder
`test_tool_pydanticai.py`	Pydantic AI ToolBuilder
`test_openai_runner.py`	OpenAIRunner execution, error handling
`test_crewai_runner.py`	CrewAIRunner execution (mocked Crew kickoff)
`test_smolagents_runner.py`	SmolagentsRunner execution, multimodal requests, error handling
`test_pydanticai_runner.py`	PydanticAIRunner execution, structured output, BinarySerde session round-trip, multimodal wiring
`test_langgraph_reasoning_live.py`	LangGraph reasoning against a REAL reasoning model, env-gated (`AK_TEST_REASONING_MODEL`; skipped in normal runs). Guards the premise the chunk-feeding unit tests cannot: that the model streams a summary at all (it must be asked — `reasoning={"summary": "auto"}`) and that LangChain surfaces it under `content_blocks`. Builds a bare `StateGraph`, because `langgraph.prebuilt` is unimportable against the pinned `langgraph`
`test_guardrail.py`	Guardrail factories, hooks
`test_api_http.py`	REST API handler
`test_chat_service_core.py`	ChatService execution core (`execute`/`execute_stream`): typed replies, prebuilt request lists, validation, error propagation, wrapper wire shapes
`test_chat_service_streaming.py`	ChatService SSE/stream chunk formatting
`test_slack_integration.py`	Slack handler on the ChatService core: request/identity mapping, attachment-only, error paths, chunking (pattern for integration handler tests)
`test_whatsapp_integration.py`	WhatsApp handler on the ChatService core: text/media paths, rejections before execute
`test_gmail_integration.py`	Gmail handler on the ChatService core: prompt assembly, session fallback, attachments, error paths
`test_thread_integration.py`	Thread integration: `ThreadRecorder` ordering/enforcement, `AgentThreadRequestHandler` recording + no-phantom-thread prechecks, stream accumulation, end-to-end read-back
`test_thread_router.py`	Thread read routes (`ThreadRESTRequestHandler`): pagination, `Authoriser` 401/403 semantics
`test_authoriser_shared.py`	Shared `Authoriser` in `agentkernel.auth`: package-export identity, guard that the thread package no longer exposes it, `AuthValidatorAuthoriser` adapter, `AuthorisedRESTRequestHandler` inheritance
`test_akagentrunner_stream.py`	Serverless `ServerlessStreamAgentRunner` (SQS streaming)
`test_serverless_agent_runner_schedule.py`	Serverless runners' trigger consumption: `request_id`/`user_id` body fallback, attribute precedence, missing-in-both error path
`test_akresponsehandler.py`	Serverless response handler (`CHAT_RESPONSE` / `STREAM_CHUNK` broadcast)
`test_ws_lambda_stream.py`	WebSocket Lambda router in `stream` mode
`test_cli_tester.py`	CLI test framework
`test_auth_handler.py`	Auth handler
`test_akauthorizer.py`	AWS Lambda authorizer
`test_lambda_router.py`	Lambda routing
`test_sqs_handler.py`	AWS SQSHandler config, client, message sending
`test_serverless_request_handle.py`	BaseRequest/BaseRunRequest parsing from serverless payloads
`test_firestore_database_id.py`	Shared `FirestoreDriver` (`core/util/driver/firestore.py`, explicit constructor params) named `database_id` configuration
`test_ak_logger.py`	AKLogger level resolution, configuration
`test_error_util.py`	`user_facing_error_message` error mapping
`test_thread_runner.py`	ThreadRunner task validation, failure/shutdown semantics
`test_ecs_sqs_consumer_parallel.py`	ECSSQSConsumer message processing + delete/retry semantics
`test_ecs_agent_runner_schedule.py`	ECS runner trigger consumption: `request_id`/`user_id` body fallback with attribute precedence, and `ChatService`'s status forwarded to the output queue instead of discarded
`test_ecs_output_consumer_status.py`	ECS output consumer persisting `status_code` on stored records (default 200, permanent failure 500)
`test_deployment_queue_contracts.py`	#495 public-interface cleanup: `pipeline.transport` (`QueueTransport`/`QueueMessage`) is the only public queue API; `RawQueueConsumer` + `SQSHandler`'s send models are internal; removed public names (`QueueHandler`, `QueueConsumer`, `deployment.common.queue_*`) raise `ImportError`
`test_pipeline_agent_runner.py`	`AgentRunner`/`StreamAgentRunner`: chat execution via `ChatService`, reply forwarding with `STATUS_CODE` attribute, per-chunk dedup suffixes, `run()` rejecting `in_memory` transport
`test_pipeline_agent_runner_schedule.py`	Pipeline runners' trigger consumption: `request_id`/`user_id` resolved from the message body, attribute precedence, and body-resolved metadata injected back into the attributes for output forwarding
`test_pipeline_bookkeeping.py`	Delivery bookkeeping for transports lacking native receive counts/dedup (spec #495 §6): `InMemoryBookkeepingStore`/`RedisLikeBookkeepingStore` attempt counters, retry-safe dedup claims, `BookkeepingStoreFactory` backend selection
`test_pipeline_sqs_transport.py`	`SQSTransport`: send/fetch/ack/nack/dead_letter, graceful shutdown handling, fetch-wait slicing, shared wire-format primitives
`test_pipeline_kafka_transport.py`	`KafkaTransport` against a fake in-memory Kafka cluster: send/fetch/ack/nack, dead-letter routing scoped by topic, delivery errors, consumer capacity check
`test_pipeline_nats_transport.py`	`NatsTransport` against a fake JetStream behind a real `_NatsLoop` bridge: subject/header construction, stable client-side (`crc32`) partition hashing, `num_delivered` mapping, nak redelivery, `term()` on permanent failure, one-in-flight-per-partition, stream-scoped dedup, `auto_provision` create-vs-verify, consumer capacity warning, and the full `QueueTransportContract` with no skips
`test_response_store_in_memory.py`	`InMemoryResponseStore`: `get_record` (`status_code` exposed), `add_chunk`/`stream` chunk-streaming for local SSE
`test_transport_contract.py`	The reusable `QueueTransportContract` (`pipeline/testing.py`) run against the `in_memory` transport
`test_transport_contract_live.py`	The same contract against REAL brokers, env-gated (`AK_TEST_NATS_URL` / `AK_TEST_KAFKA_BOOTSTRAP`; skipped in normal runs) with per-test unique streams/topics; run in CI by `test-reusable.yaml`'s `transport-integration-tests` job over the transport examples' compose stacks. Documents two live-only timing traps: the per-partition pull window must stay below `ack_wait`, and partition counts are chosen from the real partitioner mappings (crc32 / murmur2)
`test_pipeline_request_handler.py`	Pipeline `RequestHandler` over FastAPI `TestClient`: `rest_sync` parity (stored `status_code` honored), `rest_async` accept/poll, SSE streaming end-to-end, multipart-on-`in_memory` only
`test_pipeline_response_handler.py`	Pipeline `ResponseHandler` delivery paths: REST records, the `USER_ID`-presence WS routing marker, `STREAM_CHUNK`/`CHAT_RESPONSE` pushes, missing-attribute retries, permanent-failure frames/records
`test_pipeline_io_handler.py`	`IOHandler.run()` topology validation and fail-fasts (ASYNC-on-`in_memory` without a validator, broker WS modes without a push token, broker + non-shared response store), signal handlers, graceful-drain exit code
`test_pipeline_ws.py`	The gateway tier: `LocalConnectionRegistry`, `PodPushWebSocketHandler` store-lookup delivery + stale-connection cleanup, `/internal/push` auth, the native `/ws` route (1008 closes, chat enqueue attributes, custom routes), `WebSocketGateway` fail-fasts, single-process ASYNC/STREAM end-to-end over `in_memory`, and cross-"pod" delivery between two gateway apps sharing one connection store
`test_session_connection_store.py`	The `WSConnectionStore` contract over the in-memory/redis-like/DynamoDB implementations, plus `SessionStore.get_connection_store()` per backend (store-less backends raise actionably)
`test_sandbox.py`	Sandbox core: model/capabilities, error hierarchy, config, provider contract, manager + factory + embedded broker, agent surface (system tools + task-completion pre-hook), `agents` scoping
`test_sandbox_broker.py`	Broker flavors (embedded/thread) end-to-end, thread loop-identity contract, wait-policy promotion + late-completion recovery, suspend/resume completion ingestion
`test_sandbox_providers.py`	`local_subprocess` (real subprocess) + `docker` (mocked SDK) providers, run against the reusable `SandboxProviderContract`
`test_authoriser_shared.py`	Shared `Authoriser` in `agentkernel.auth`: `AuthValidatorAuthoriser` adaptation, export identity, and the guard that the thread package no longer re-exports it
`test_schedule_model.py`	`ScheduleSpec` one-of/timezone/session_mode validation, chat-envelope parsing, `ScheduledTask` JSON round trip (JSON primitives only)
`test_schedule_manager.py`	`ScheduleManager`: `get()` gating + singleton, provider/transport fail-fast, semantic validation matrix (including the named-agent precheck and the unnamed-agent exemption), create ordering + rollback, trigger-body freezing, ownership, amendment rules (occurrence rule replaced as a unit, untouched when the amendment names none of it), cancellation, occurrence recording (never raises)
`test_schedule_store.py`	Every `ScheduleStore` backend against the one contract the in_memory class pins — in_memory, redis/valkey through a fake redis-like client injected as `store._driver._client`, dynamodb through a fake driver whose `table.scan` replays `LastEvaluatedKey` pages — plus index cleanup on delete, TTL behaviour (none by default), and `ScheduleStoreBuilder` built-in/BYO/unknown-type/missing-extra resolution
`test_schedule_provider_local.py`	`LocalScheduleProvider`: next-fire computation, one-time vs re-armed occurrences, token substitution, body-only delivery into `InMemoryTransport`, pause/delete disarm, `ScheduleProviderFactory` resolution
`test_schedule_provider_eventbridge.py`	`EventBridgeScheduleProvider` against a mocked `boto3.client`: cron/at expression translation, exact registration/amendment/removal kwargs sent to the Scheduler API, error mapping to `ScheduleError`, `ScheduleProviderFactory` wiring
`test_schedule_router.py`	`ScheduleRESTRequestHandler`: 404 when unconfigured, the three 401 variants from the shared `AuthorisedRESTRequestHandler`, listings forced to the authorised user, 403-before-404 ordering, PUT amendment happy path + validation 400s, DELETE returning the cancelled task, mount-time validation of the configured backends (`get_router()` building the manager, an unusable provider failing the mount, mounting while unconfigured still allowed), and the guard that importing the package does not pull in FastAPI
`test_schedule_tools.py`	Schedule system tools: `SystemToolFactory` registration + `schedule.agents` scoping, prompt-suffix content, disabled short-circuit, acting-user read from the session volatile cache, and each tool's JSON contract including the no-identity and unknown-agent errors
`test_chat_service_schedule.py`	ChatService interception on all four entry points: 202 wire shapes, streaming terminal chunk, unconfigured 400, occurrence recording, no scheduling field leaking as `AgentRequestAny`
`test_serverless_status_propagation.py`	The status a chat run produced (e.g. 202 deferred-to-schedule, 4xx rejected) survives the serverless queue round trip: `ServerlessAgentRunner` forwards it as the `ATTR_STATUS_CODE` output-message attribute, `ResponseHandler` stores it on the response record, and the REST surface (`DefaultEndpointsHandler`, response-store polling) replays it to the caller
`test_pipeline_agent_runner_schedule.py`	Pipeline runners' trigger contract: `request_id`/`user_id` body fallback, attribute precedence, attribute injection for output forwarding
`test_ecs_agent_runner_schedule.py`	ECS runners' body fallback (`_get_record_attributes` with/without a parsed body) and `status_code` custom attribute
`test_ecs_output_consumer_status.py`	`ECSOutputConsumer` stores `status_code` (present / absent → 200 / permanent failure → 500)
`test_serverless_agent_runner_schedule.py`	Serverless runners' body fallback in both `_get_record_attributes` implementations
`test_factory.py`	Shared pluggable-backend helpers (`resolve_dotted`, `require_extra`, `AKConfigError`) in `core/util/factory.py`
`test_store_builders.py`	Session/thread/multimodal store builders: fail-loud on unknown type, BYO dotted-path subclass resolution
`test_trace.py`	Trace factory built-in resolution, BYO dotted path, unknown-type error

Read the full file on GitHub · 399 lines

Changes

What this file has done since we first saw it

Hashed on every crawl. A supply-chain change to an agent config is a question of when, not whether, so the history is kept rather than the latest state alone.

3d ago First seen · 399 lines · 59 tokens per session scan A c24686580702

Subscribe to this mod's changes

ak-dev-testing-conventions is a skill published in the GitHub repository yaalalabs/agent-kernel (145 stars, last pushed 5d ago), licensed Apache-2.0. It adds 59 tokens to every session and 7,131 once invoked, about $0.0003 per session on Opus 5. A static security scan graded it A with 0 findings. No closer match exists in the catalogue, so it is treated as the original; first seen 2026-08-30.

Testing a2a adk ai-agents aws azure cloud-native

notion

Notion workspace integration for searching pages, managing databases, creating postmortems, and exporting RCA findings.

Arvo-AI/aurora · 23 tokens

datadog

Datadog monitoring integration for querying logs, metrics, monitors, events, traces, hosts, and incidents during RCA investigations.

Arvo-AI/aurora · 28 tokens

ovh

OVHcloud infrastructure integration for managing instances, Kubernetes clusters, networks, and object storage via CLI and Terraform.

Arvo-AI/aurora · 24 tokens

scaleway

Scaleway cloud integration for managing instances, Kapsule Kubernetes clusters, object storage, and managed databases via CLI and Terraform.

Arvo-AI/aurora · 29 tokens

postmortem

Postmortem generation and management tools for writing, reading, and versioning incident postmortems.