> ## Documentation Index > Fetch the complete documentation index at: https://usefused.com/docs/llms.txt > Use this file to discover all available pages before exploring further. # harnest.testing > Public Python signatures, fields, and methods. [All Python modules](https://usefused.com/docs/harnest/reference/api/overview.md) ยท [Import and compatibility guidance](https://usefused.com/docs/harnest/build/python-imports.md) ## Smoke client operations | Method | Returns | | - | - | | `smoke.respond(...)` | One neutral JSON response | | `smoke.stream(...)` | Ordered SSE event objects | ## Exports | Name | Kind | | - | - | | [`AgentTestError`](#agenttesterror) | Class | | [`ChannelStoreConformanceMixin`](#channelstoreconformancemixin) | Class | | [`MemoryStoreConformanceMixin`](#memorystoreconformancemixin) | Class | | [`SmokeClient`](#smokeclient) | Class | | [`TaskStoreConformanceMixin`](#taskstoreconformancemixin) | Class | | [`run_agent_tests`](#run_agent_tests) | Function | ## AgentTestError An authored test suite does not follow the Harnest convention. > [accordion] Example > > ```python theme={null} > from harnest.testing import AgentTestError > > try: > ... > except AgentTestError as error: > ... > ``` ## ChannelStoreConformanceMixin Subclass with IsolatedAsyncioTestCase and an async make\_store factory. ```python theme={null} ChannelStoreConformanceMixin() ``` > [accordion] Example > > ```python theme={null} > from harnest.testing import ChannelStoreConformanceMixin > > mixin = ChannelStoreConformanceMixin() > ``` ### ChannelStoreConformanceMixin.make\_store ```python theme={null} async ChannelStoreConformanceMixin.make_store() -> Any ``` Override to return an unstarted provider implementing ChannelStore. Tests use a unique installation scope per case. Register backend cleanup with addAsyncCleanup; the mixin owns starting and closing it. > [accordion] Example > > ```python theme={null} > from harnest.testing import ChannelStoreConformanceMixin > > mixin: ChannelStoreConformanceMixin > result = await mixin.make_store() > ``` ### ChannelStoreConformanceMixin.asyncSetUp ```python theme={null} async ChannelStoreConformanceMixin.asyncSetUp() -> None ``` Give every check its own installation scope and provider lifecycle. > [accordion] Example > > ```python theme={null} > from harnest.testing import ChannelStoreConformanceMixin > > mixin: ChannelStoreConformanceMixin > await mixin.asyncSetUp() > ``` ### ChannelStoreConformanceMixin.channel\_event ```python theme={null} ChannelStoreConformanceMixin.channel_event(provider_event_id: str = 'evt', **changes: Any) -> ChannelEvent ``` Build deterministic safe event input independently of provider details. > [accordion] Parameters and example > > | Parameter | Type | Default | > | - | - | - | > | `provider_event_id` | `str` | `'evt'` | > | `**changes` | `Any` | Optional | > > ```python theme={null} > from harnest.testing import ChannelStoreConformanceMixin > > mixin: ChannelStoreConformanceMixin > event = mixin.channel_event() > ``` ### ChannelStoreConformanceMixin.channel\_reply ```python theme={null} ChannelStoreConformanceMixin.channel_reply(reply_id: str = 'reply', **changes: Any) -> ChannelReply ``` Build one queued reply targeting this scope's default conversation. > [accordion] Parameters and example > > | Parameter | Type | Default | > | - | - | - | > | `reply_id` | `str` | `'reply'` | > | `**changes` | `Any` | Optional | > > ```python theme={null} > from harnest.testing import ChannelStoreConformanceMixin > > mixin: ChannelStoreConformanceMixin > reply = mixin.channel_reply() > ``` ### ChannelStoreConformanceMixin.test\_admit\_is\_idempotent\_conflict\_checked\_and\_scope\_isolated ```python theme={null} async ChannelStoreConformanceMixin.test_admit_is_idempotent_conflict_checked_and_scope_isolated() -> None ``` Concurrent retries admit once and reject a changed redefinition. > [accordion] Example > > ```python theme={null} > from harnest.testing import ChannelStoreConformanceMixin > > mixin: ChannelStoreConformanceMixin > await mixin.test_admit_is_idempotent_conflict_checked_and_scope_isolated() > ``` ### ChannelStoreConformanceMixin.test\_get\_event\_reads\_only\_the\_admitted\_identity ```python theme={null} async ChannelStoreConformanceMixin.test_get_event_reads_only_the_admitted_identity() -> None ``` Reads are scoped to platform, installation, and provider event ID. > [accordion] Example > > ```python theme={null} > from harnest.testing import ChannelStoreConformanceMixin > > mixin: ChannelStoreConformanceMixin > await mixin.test_get_event_reads_only_the_admitted_identity() > ``` ### ChannelStoreConformanceMixin.test\_session\_binding\_is\_first\_write\_wins ```python theme={null} async ChannelStoreConformanceMixin.test_session_binding_is_first_write_wins() -> None ``` A conversation scope keeps its first bound session across retries. > [accordion] Example > > ```python theme={null} > from harnest.testing import ChannelStoreConformanceMixin > > mixin: ChannelStoreConformanceMixin > await mixin.test_session_binding_is_first_write_wins() > ``` ### ChannelStoreConformanceMixin.test\_reply\_lifecycle\_claims\_pending\_and\_finishes\_once ```python theme={null} async ChannelStoreConformanceMixin.test_reply_lifecycle_claims_pending_and_finishes_once() -> None ``` Replies are claimable while pending and terminal exactly once. > [accordion] Example > > ```python theme={null} > from harnest.testing import ChannelStoreConformanceMixin > > mixin: ChannelStoreConformanceMixin > await mixin.test_reply_lifecycle_claims_pending_and_finishes_once() > ``` ## MemoryStoreConformanceMixin Combine with IsolatedAsyncioTestCase and implement async make\_store(). ```python theme={null} MemoryStoreConformanceMixin() ``` > [accordion] Example > > ```python theme={null} > from harnest.testing import MemoryStoreConformanceMixin > > mixin = MemoryStoreConformanceMixin() > ``` ### MemoryStoreConformanceMixin.make\_store ```python theme={null} async MemoryStoreConformanceMixin.make_store() -> harnest.memory.MemoryStore ``` Return an unstarted provider; register any test-data cleanup yourself. > [accordion] Example > > ```python theme={null} > from harnest.testing import MemoryStoreConformanceMixin > > mixin: MemoryStoreConformanceMixin > store = await mixin.make_store() > ``` ### MemoryStoreConformanceMixin.asyncSetUp ```python theme={null} async MemoryStoreConformanceMixin.asyncSetUp() -> None ``` Own a fresh provider and unique application scope for each check. > [accordion] Example > > ```python theme={null} > from harnest.testing import MemoryStoreConformanceMixin > > mixin: MemoryStoreConformanceMixin > await mixin.asyncSetUp() > ``` ### MemoryStoreConformanceMixin.record ```python theme={null} MemoryStoreConformanceMixin.record(key: str = 'preference', **changes: Any) -> harnest.memory.MemoryRecord ``` Build a valid explicit write without coupling tests to a datastore. > [accordion] Parameters and example > > | Parameter | Type | Default | > | - | - | - | > | `key` | `str` | `'preference'` | > | `**changes` | `Any` | Optional | > > ```python theme={null} > from harnest.testing import MemoryStoreConformanceMixin > > mixin: MemoryStoreConformanceMixin > record = mixin.record() > ``` ### MemoryStoreConformanceMixin.test\_memory\_is\_scoped\_and\_explicit ```python theme={null} async MemoryStoreConformanceMixin.test_memory_is_scoped_and_explicit() -> None ``` Identical keys never cross application, user or namespace boundaries. > [accordion] Example > > ```python theme={null} > from harnest.testing import MemoryStoreConformanceMixin > > mixin: MemoryStoreConformanceMixin > await mixin.test_memory_is_scoped_and_explicit() > ``` ### MemoryStoreConformanceMixin.test\_memory\_results\_are\_detached ```python theme={null} async MemoryStoreConformanceMixin.test_memory_results_are_detached() -> None ``` Callers cannot silently mutate persisted metadata through returned objects. > [accordion] Example > > ```python theme={null} > from harnest.testing import MemoryStoreConformanceMixin > > mixin: MemoryStoreConformanceMixin > await mixin.test_memory_results_are_detached() > ``` ### MemoryStoreConformanceMixin.test\_memory\_json\_roundtrip\_is\_lossless ```python theme={null} async MemoryStoreConformanceMixin.test_memory_json_roundtrip_is_lossless() -> None ``` Provider scripting must preserve arrays, exact integers and floats. > [accordion] Example > > ```python theme={null} > from harnest.testing import MemoryStoreConformanceMixin > > mixin: MemoryStoreConformanceMixin > await mixin.test_memory_json_roundtrip_is_lossless() > ``` ### MemoryStoreConformanceMixin.test\_memory\_maintenance\_preserves\_other\_owners ```python theme={null} async MemoryStoreConformanceMixin.test_memory_maintenance_preserves_other_owners() -> None ``` Bulk erasure includes expired records but never another application/user. > [accordion] Example > > ```python theme={null} > from harnest.testing import MemoryStoreConformanceMixin > > mixin: MemoryStoreConformanceMixin > await mixin.test_memory_maintenance_preserves_other_owners() > ``` ### MemoryStoreConformanceMixin.test\_memory\_purge\_physically\_removes\_expired\_pages ```python theme={null} async MemoryStoreConformanceMixin.test_memory_purge_physically_removes_expired_pages() -> None ``` Zero-deletion pages continue and live data survives a bounded sweep. > [accordion] Example > > ```python theme={null} > from harnest.testing import MemoryStoreConformanceMixin > > mixin: MemoryStoreConformanceMixin > await mixin.test_memory_purge_physically_removes_expired_pages() > ``` ### MemoryStoreConformanceMixin.test\_memory\_revision\_fences\_concurrent\_writers ```python theme={null} async MemoryStoreConformanceMixin.test_memory_revision_fences_concurrent_writers() -> None ``` Exactly one writer can replace the same revision across concurrent calls. > [accordion] Example > > ```python theme={null} > from harnest.testing import MemoryStoreConformanceMixin > > mixin: MemoryStoreConformanceMixin > await mixin.test_memory_revision_fences_concurrent_writers() > ``` ### MemoryStoreConformanceMixin.test\_memory\_delete\_does\_not\_allow\_stale\_resurrection ```python theme={null} async MemoryStoreConformanceMixin.test_memory_delete_does_not_allow_stale_resurrection() -> None ``` Deletion removes search membership and invalidates earlier revisions. > [accordion] Example > > ```python theme={null} > from harnest.testing import MemoryStoreConformanceMixin > > mixin: MemoryStoreConformanceMixin > await mixin.test_memory_delete_does_not_allow_stale_resurrection() > ``` ### MemoryStoreConformanceMixin.test\_memory\_pages\_and\_literal\_search ```python theme={null} async MemoryStoreConformanceMixin.test_memory_pages_and_literal_search() -> None ``` Paginate datastore-filtered results without interpreting wildcard syntax. > [accordion] Example > > ```python theme={null} > from harnest.testing import MemoryStoreConformanceMixin > > mixin: MemoryStoreConformanceMixin > await mixin.test_memory_pages_and_literal_search() > ``` ### MemoryStoreConformanceMixin.test\_memory\_expiry\_hides\_records\_and\_fences\_updates ```python theme={null} async MemoryStoreConformanceMixin.test_memory_expiry_hides_records_and_fences_updates() -> None ``` Expiry applies to get, list, search and conditional mutation alike. > [accordion] Example > > ```python theme={null} > from harnest.testing import MemoryStoreConformanceMixin > > mixin: MemoryStoreConformanceMixin > await mixin.test_memory_expiry_hides_records_and_fences_updates() > ``` ### MemoryStoreConformanceMixin.test\_memory\_rejects\_invalid\_payloads\_and\_limits ```python theme={null} async MemoryStoreConformanceMixin.test_memory_rejects_invalid_payloads_and_limits() -> None ``` Validation never silently truncates private data or unbounded queries. > [accordion] Example > > ```python theme={null} > from harnest.testing import MemoryStoreConformanceMixin > > mixin: MemoryStoreConformanceMixin > await mixin.test_memory_rejects_invalid_payloads_and_limits() > ``` ## SmokeClient Small neutral-API helper injected into opted-in smoke tests. ```python theme={null} SmokeClient(client: Any) -> None ``` > [accordion] Parameters and example > > | Parameter | Type | Default | > | - | - | - | > | `client` | `Any` | Required | > > ```python theme={null} > from harnest.testing import SmokeClient > > client = SmokeClient(client=...) > ``` ### SmokeClient.respond ```python theme={null} SmokeClient.respond( input: str, *, session_id: str | None = None, metadata: Mapping[str, Any] | None = None, ) -> dict[str, Any] ``` Invoke the neutral response endpoint and return its JSON object. > [accordion] Parameters and example > > | Parameter | Type | Default | > | - | - | - | > | `input` | `str` | Required | > | `session_id` | `str \| None` | `None` | > | `metadata` | `Mapping[str, Any] \| None` | `None` | > > ```python theme={null} > from harnest.testing import SmokeClient > > client: SmokeClient > result = client.respond(input=...) > ``` ### SmokeClient.stream ```python theme={null} SmokeClient.stream( input: str, *, session_id: str | None = None, metadata: Mapping[str, Any] | None = None, ) -> list[dict[str, Any]] ``` Invoke the streaming endpoint and return parsed server-sent events. > [accordion] Parameters and example > > | Parameter | Type | Default | > | - | - | - | > | `input` | `str` | Required | > | `session_id` | `str \| None` | `None` | > | `metadata` | `Mapping[str, Any] \| None` | `None` | > > ```python theme={null} > from harnest.testing import SmokeClient > > client: SmokeClient > result = client.stream(input=...) > ``` ## TaskStoreConformanceMixin Subclass with IsolatedAsyncioTestCase and an async make\_store factory. ```python theme={null} TaskStoreConformanceMixin() ``` > [accordion] Example > > ```python theme={null} > from harnest.testing import TaskStoreConformanceMixin > > mixin = TaskStoreConformanceMixin() > ``` ### TaskStoreConformanceMixin.make\_store ```python theme={null} async TaskStoreConformanceMixin.make_store() -> Any ``` Override to return an unstarted provider implementing both contracts. Tests use a unique application scope per case. Register backend cleanup with addAsyncCleanup; the mixin owns starting and closing the provider. > [accordion] Example > > ```python theme={null} > from harnest.testing import TaskStoreConformanceMixin > > mixin: TaskStoreConformanceMixin > result = await mixin.make_store() > ``` ### TaskStoreConformanceMixin.asyncSetUp ```python theme={null} async TaskStoreConformanceMixin.asyncSetUp() -> None ``` Give every check its own application scope and provider lifecycle. > [accordion] Example > > ```python theme={null} > from harnest.testing import TaskStoreConformanceMixin > > mixin: TaskStoreConformanceMixin > await mixin.asyncSetUp() > ``` ### TaskStoreConformanceMixin.task\_record ```python theme={null} TaskStoreConformanceMixin.task_record(job_id: str = 'job', **changes: Any) -> TaskRecord ``` Build deterministic safe job input independently of provider details. > [accordion] Parameters and example > > | Parameter | Type | Default | > | - | - | - | > | `job_id` | `str` | `'job'` | > | `**changes` | `Any` | Optional | > > ```python theme={null} > from harnest.testing import TaskStoreConformanceMixin > > mixin: TaskStoreConformanceMixin > record = mixin.task_record() > ``` ### TaskStoreConformanceMixin.cron\_record ```python theme={null} TaskStoreConformanceMixin.cron_record(schedule_id: str = 'cron', **changes: Any) -> CronRecord ``` Build one due user schedule with stable ownership and definition. > [accordion] Parameters and example > > | Parameter | Type | Default | > | - | - | - | > | `schedule_id` | `str` | `'cron'` | > | `**changes` | `Any` | Optional | > > ```python theme={null} > from harnest.testing import TaskStoreConformanceMixin > > mixin: TaskStoreConformanceMixin > record = mixin.cron_record() > ``` ### TaskStoreConformanceMixin.claim ```python theme={null} async TaskStoreConformanceMixin.claim(**changes: Any) -> tuple[TaskRecord, ...] ``` Claim one bounded batch through the shared worker contract. > [accordion] Parameters and example > > | Parameter | Type | Default | > | - | - | - | > | `**changes` | `Any` | Optional | > > ```python theme={null} > from harnest.testing import TaskStoreConformanceMixin > > mixin: TaskStoreConformanceMixin > result = await mixin.claim() > ``` ### TaskStoreConformanceMixin.read\_task ```python theme={null} async TaskStoreConformanceMixin.read_task(job_id: str = 'job', user_id: str = 'alice') -> TaskRecord | None ``` Read a task as the selected owner. > [accordion] Parameters and example > > | Parameter | Type | Default | > | - | - | - | > | `job_id` | `str` | `'job'` | > | `user_id` | `str` | `'alice'` | > > ```python theme={null} > from harnest.testing import TaskStoreConformanceMixin > > mixin: TaskStoreConformanceMixin > record = await mixin.read_task() > ``` ### TaskStoreConformanceMixin.finish ```python theme={null} async TaskStoreConformanceMixin.finish(record: TaskRecord, **changes: Any) -> bool ``` Finish the exact lease with deterministic test timestamps. > [accordion] Parameters and example > > | Parameter | Type | Default | > | - | - | - | > | `record` | `TaskRecord` | Required | > | `**changes` | `Any` | Optional | > > ```python theme={null} > from harnest.testing import TaskStoreConformanceMixin > > mixin: TaskStoreConformanceMixin > result = await mixin.finish(record=...) > ``` ### TaskStoreConformanceMixin.test\_enqueue\_is\_atomic\_idempotent\_and\_owner\_scoped ```python theme={null} async TaskStoreConformanceMixin.test_enqueue_is_atomic_idempotent_and_owner_scoped() -> None ``` Concurrent retries produce one job without crossing task/user scope. > [accordion] Example > > ```python theme={null} > from harnest.testing import TaskStoreConformanceMixin > > mixin: TaskStoreConformanceMixin > await mixin.test_enqueue_is_atomic_idempotent_and_owner_scoped() > ``` ### TaskStoreConformanceMixin.test\_completed\_task\_retains\_idempotency\_but\_scrubs\_payload ```python theme={null} async TaskStoreConformanceMixin.test_completed_task_retains_idempotency_but_scrubs_payload() -> None ``` Private inputs disappear while retained fingerprints deduplicate retries. > [accordion] Example > > ```python theme={null} > from harnest.testing import TaskStoreConformanceMixin > > mixin: TaskStoreConformanceMixin > await mixin.test_completed_task_retains_idempotency_but_scrubs_payload() > ``` ### TaskStoreConformanceMixin.test\_claim\_filters\_due\_queue\_and\_application\_and\_never\_duplicates ```python theme={null} async TaskStoreConformanceMixin.test_claim_filters_due_queue_and_application_and_never_duplicates() -> None ``` Replicas atomically claim distinct jobs using datastore scope predicates. > [accordion] Example > > ```python theme={null} > from harnest.testing import TaskStoreConformanceMixin > > mixin: TaskStoreConformanceMixin > await mixin.test_claim_filters_due_queue_and_application_and_never_duplicates() > ``` ### TaskStoreConformanceMixin.test\_expired\_lease\_cannot\_finish\_and\_reclaim\_uses\_new\_token ```python theme={null} async TaskStoreConformanceMixin.test_expired_lease_cannot_finish_and_reclaim_uses_new_token() -> None ``` Fencing rejects old workers before and after a replacement claim. > [accordion] Example > > ```python theme={null} > from harnest.testing import TaskStoreConformanceMixin > > mixin: TaskStoreConformanceMixin > await mixin.test_expired_lease_cannot_finish_and_reclaim_uses_new_token() > ``` ### TaskStoreConformanceMixin.test\_renewal\_prevents\_early\_reclaim ```python theme={null} async TaskStoreConformanceMixin.test_renewal_prevents_early_reclaim() -> None ``` A valid heartbeat extends exclusive attempt ownership. > [accordion] Example > > ```python theme={null} > from harnest.testing import TaskStoreConformanceMixin > > mixin: TaskStoreConformanceMixin > await mixin.test_renewal_prevents_early_reclaim() > ``` ### TaskStoreConformanceMixin.test\_retry\_delay\_and\_exhausted\_crash\_recovery ```python theme={null} async TaskStoreConformanceMixin.test_retry_delay_and_exhausted_crash_recovery() -> None ``` Retries retain inputs and crashed final attempts reach a terminal state. > [accordion] Example > > ```python theme={null} > from harnest.testing import TaskStoreConformanceMixin > > mixin: TaskStoreConformanceMixin > await mixin.test_retry_delay_and_exhausted_crash_recovery() > ``` ### TaskStoreConformanceMixin.test\_cancellation\_is\_scoped\_terminal\_and\_fences\_running\_attempt ```python theme={null} async TaskStoreConformanceMixin.test_cancellation_is_scoped_terminal_and_fences_running_attempt() -> None ``` Cancellation cannot reveal another owner's work or accept late results. > [accordion] Example > > ```python theme={null} > from harnest.testing import TaskStoreConformanceMixin > > mixin: TaskStoreConformanceMixin > await mixin.test_cancellation_is_scoped_terminal_and_fences_running_attempt() > ``` ### TaskStoreConformanceMixin.test\_snapshots\_do\_not\_share\_mutable\_payload ```python theme={null} async TaskStoreConformanceMixin.test_snapshots_do_not_share_mutable_payload() -> None ``` Mutating submitted or returned objects cannot rewrite durable records. > [accordion] Example > > ```python theme={null} > from harnest.testing import TaskStoreConformanceMixin > > mixin: TaskStoreConformanceMixin > await mixin.test_snapshots_do_not_share_mutable_payload() > ``` ### TaskStoreConformanceMixin.test\_cron\_idempotency\_revision\_and\_terminal\_cancellation ```python theme={null} async TaskStoreConformanceMixin.test_cron_idempotency_revision_and_terminal_cancellation() -> None ``` Definition retry preserves inactive state and stale edits cannot win. > [accordion] Example > > ```python theme={null} > from harnest.testing import TaskStoreConformanceMixin > > mixin: TaskStoreConformanceMixin > await mixin.test_cron_idempotency_revision_and_terminal_cancellation() > ``` ### TaskStoreConformanceMixin.test\_cron\_lists\_enforce\_owner\_due\_state\_and\_keyset\_pagination ```python theme={null} async TaskStoreConformanceMixin.test_cron_lists_enforce_owner_due_state_and_keyset_pagination() -> None ``` Due pagination includes equal timestamps without leaking foreign scopes. > [accordion] Example > > ```python theme={null} > from harnest.testing import TaskStoreConformanceMixin > > mixin: TaskStoreConformanceMixin > await mixin.test_cron_lists_enforce_owner_due_state_and_keyset_pagination() > ``` ### TaskStoreConformanceMixin.test\_cron\_occurrence\_commit\_is\_atomic\_and\_exactly\_deduplicated ```python theme={null} async TaskStoreConformanceMixin.test_cron_occurrence_commit_is_atomic_and_exactly_deduplicated() -> None ``` Concurrent dispatchers advance once and commit exactly one runnable job. > [accordion] Example > > ```python theme={null} > from harnest.testing import TaskStoreConformanceMixin > > mixin: TaskStoreConformanceMixin > await mixin.test_cron_occurrence_commit_is_atomic_and_exactly_deduplicated() > ``` ### TaskStoreConformanceMixin.test\_cron\_max\_runs\_caps\_atomic\_handoffs ```python theme={null} async TaskStoreConformanceMixin.test_cron_max_runs_caps_atomic_handoffs() -> None ``` A schedule cannot enqueue more occurrences than its run cap. > [accordion] Example > > ```python theme={null} > from harnest.testing import TaskStoreConformanceMixin > > mixin: TaskStoreConformanceMixin > await mixin.test_cron_max_runs_caps_atomic_handoffs() > ``` ### TaskStoreConformanceMixin.test\_cron\_failure\_streak\_resets\_after\_success ```python theme={null} async TaskStoreConformanceMixin.test_cron_failure_streak_resets_after_success() -> None ``` Only consecutive terminal failures stop later occurrences. > [accordion] Example > > ```python theme={null} > from harnest.testing import TaskStoreConformanceMixin > > mixin: TaskStoreConformanceMixin > await mixin.test_cron_failure_streak_resets_after_success() > ``` ### TaskStoreConformanceMixin.test\_cancelled\_cron\_cannot\_commit\_and\_conflict\_rolls\_back\_cursor ```python theme={null} async TaskStoreConformanceMixin.test_cancelled_cron_cannot_commit_and_conflict_rolls_back_cursor() -> None ``` A cancelled/stale schedule or failed enqueue leaves no partial handoff. > [accordion] Example > > ```python theme={null} > from harnest.testing import TaskStoreConformanceMixin > > mixin: TaskStoreConformanceMixin > await mixin.test_cancelled_cron_cannot_commit_and_conflict_rolls_back_cursor() > ``` ### TaskStoreConformanceMixin.test\_occurrence\_cannot\_substitute\_schedule\_arguments ```python theme={null} async TaskStoreConformanceMixin.test_occurrence_cannot_substitute_schedule_arguments() -> None ``` Atomic handoff must preserve the validated schedule's authored input. > [accordion] Example > > ```python theme={null} > from harnest.testing import TaskStoreConformanceMixin > > mixin: TaskStoreConformanceMixin > await mixin.test_occurrence_cannot_substitute_schedule_arguments() > ``` ### TaskStoreConformanceMixin.test\_limits\_reject\_unbounded\_or\_ambiguous\_requests ```python theme={null} async TaskStoreConformanceMixin.test_limits_reject_unbounded_or_ambiguous_requests() -> None ``` Providers share the public bounded-page contract before datastore I/O. > [accordion] Example > > ```python theme={null} > from harnest.testing import TaskStoreConformanceMixin > > mixin: TaskStoreConformanceMixin > await mixin.test_limits_reject_unbounded_or_ambiguous_requests() > ``` ## run\_agent\_tests Test authored source with eval preparation progress and optional quiet output. ```python theme={null} run_agent_tests( source: str | Path, *, include_smoke: bool = False, include_evals: bool = False, eval_trajectory: str = 'business', no_output: bool = False, eval_output: Path | None = None, framework: str = 'adk', mode: str = 'managed', cli_enabled: bool = False, ) -> int ``` > [accordion] Parameters and example > > | Parameter | Type | Default | > | - | - | - | > | `source` | `str \| Path` | Required | > | `include_smoke` | `bool` | `False` | > | `include_evals` | `bool` | `False` | > | `eval_trajectory` | `str` | `'business'` | > | `no_output` | `bool` | `False` | > | `eval_output` | `Path \| None` | `None` | > | `framework` | `str` | `'adk'` | > | `mode` | `str` | `'managed'` | > | `cli_enabled` | `bool` | `False` | > > ```python theme={null} > from harnest.testing import run_agent_tests > > result = run_agent_tests(source=...) > ``` This documentation is built and hosted on [Mintlify](https://mintlify.com), a developer documentation platform.