Testing API — Vega SDK Reference

Import

python

from vega_capability_sdk.testing import (
    ConformanceFailure,
    ConformanceReport,
    InMemoryArtifactStore,
    ProgressCollector,
    ProviderContractHarness,
    TestClock,
    fake_execution_context,
    fake_initialization_context,
)

TestClock

python

@dataclass(slots=True)
class TestClock:
    current: datetime = datetime(2020, 1, 1, tzinfo=UTC)

    def now(self) -> datetime: ...
    def advance(self, delta: timedelta) -> datetime: ...

current must be timezone-aware and is normalized to UTC at construction.now() returns it. advance() increments it by delta and returns the new value. Construction raises ValueError for a naive current.advance() applies the supplied timedelta immediately and does not sleep.

InMemoryArtifactStore

python

class InMemoryArtifactStore:
    def create(self, *, mime_type: str, logical_role: str) -> ArtifactWriter: ...
    def read(self, reference: ArtifactReference) -> bytes: ...

In-memory implementation of the Artifact service for tests. Its writer accepts bytes untilfinalize(), then returns an ArtifactReference whose URI starts withmemory://. read() returns the finalized bytes for a reference. Writing or finalizing a closed writer raises RuntimeError. Exiting a writer context without finalizing, or due to an exception, aborts it. read() raises KeyErrorwhen the reference ID was not finalized in this store.

ProgressCollector

python

class ProgressCollector:
    events: list[dict[str, Any]]
    async def __call__(self, event: dict[str, Any]) -> None: ...

Callable progress callback that appends every received event to events in order. The collector stores the event object it receives; it does not copy, freeze, or validate the event.

Fake Context Factories

fake_initialization_context

python

def fake_initialization_context(
    *,
    capability_id: str = "test.capability",
    provider_id: str = "test.provider",
    configuration: dict[str, Any] | None = None,
    secrets: dict[str, str] | None = None,
    clock: TestClock | None = None,
    artifacts: InMemoryArtifactStore | None = None,
) -> ProviderInitializationContext: ...

Returns a context with deterministic test robot identity, a configuration view, a secret accessor, bound structured logger, a TestClock, and an in-memory Artifact store unless alternatives are supplied. Identity fields are test-installation,test-robot, and test placeholders. The factory does not accept overrides for robot identity, metadata, or logger.

fake_execution_context

python

def fake_execution_context(
    *,
    execution_id: str = "test-execution",
    collector: ProgressCollector | None = None,
    clock: TestClock | None = None,
    artifacts: InMemoryArtifactStore | None = None,
) -> CapabilityExecutionContext: ...

Creates a context with a new CancellationToken, a ProgressReporter using the supplied or new collector, a deterministic trace ID and span ID, a bound logger, and test clock/artifact services. Supply a collector explicitly when tests need to inspect progress.

Conformance Harness

ConformanceFailure — Frozen dataclass with check: str andmessage: str.

ConformanceReport

python

@dataclass(frozen=True, slots=True)
class ConformanceReport:
    failures: tuple[ConformanceFailure, ...] = ()

    @property
    def passed(self) -> bool: ...
    def raise_for_failures(self) -> None: ...

passed is True when there are no failures.raise_for_failures() raises AssertionError with a semicolon-separated summary if any failure exists.

ProviderContractHarness

python

class ProviderContractHarness:
    def __init__(self, *, timeout_seconds: float = 5.0) -> None: ...
    async def run(
        self,
        provider: CapabilityProvider,
        request: CapabilityExecutionRequest,
        *,
        action: ActionDefinition | None = None,
    ) -> ConformanceReport: ...

Runs initialization, health check, execution, and shutdown with a timeout for each operation. It records failures rather than raising them, except that the report can later be raised withraise_for_failures(). The harness verifies that health returnsCapabilityHealthResult and execution returns CapabilityExecutionResult. If action is supplied, it also validates the result output againstaction.output_schema.

Each lifecycle call uses asyncio.wait_for and timeout_seconds. Initialization failure returns immediately with one initialize failure. Wrong health and result types, execution failures, and optional output-schema failures are recorded; shutdown is attempted in finally. The harness catches BaseException and formats failures as <ExceptionType>: <message>. It tests a narrow SDK contract; passing it does not establish hardware safety or Runtime deployment compatibility.

Example

python

import pytest

from vega_capability_sdk import CapabilityExecutionRequest
from vega_capability_sdk.testing import ProviderContractHarness


@pytest.mark.asyncio
async def test_provider_contract(provider):
    request = CapabilityExecutionRequest(
        execution_id="test-execution",
        capability_id="example.echo",
        provider_id="python",
        action_id="example.echo.run",
        action_version="1.0.0",
        inputs={"text": "hello"},
    )
    report = await ProviderContractHarness().run(provider, request)
    report.raise_for_failures()