name	write-conformance-tests
description	Author GenAI conformance-test scenarios for an OpenTelemetry instrumentation package — Scenario subclasses under tests/conformance/, the test_conformance.py runner, declared gaps, lib-specific assertions, and weaver live-check policies. Use when adding or updating conformance tests for any instrumentation, whether greenfield or ported.

Write GenAI conformance tests

Conformance tests validate that an instrumentation package emits telemetry matching the GenAI semantic conventions via Weaver live-check. They apply to any instrumentation package — greenfield or ported — and don't depend on how the package was built.

This skill covers authoring the tests/conformance/<scenario>.py modules and the tests/test_conformance.py runner. For the always-on rules that hold even without this skill loaded, see the Conformance tests section of AGENTS.md.

One scenario per operation

Put one scenario per emitted semconv operation under tests/conformance/. Write a scenario for every semconv operation the library emits, even one currently blocked by a util-genai or semconv gap. Skipping the scenario hides the gap; writing it records the gap (see Declared gaps). Never drop a scenario file because it would fail today.

Recommended scenarios

Cover the scenarios below that apply to the library. Skip a row only when the library genuinely can't perform that operation (e.g. an inference-only client has no embeddings scenario).

LLM client instrumentations:

Scenario	File	Covers
Inference	`inference.py`	A `chat` operation.
Tool calling	`tool_calling.py`	A `chat` turn where the model returns tool calls and a follow-up turn feeds tool results back. Asserts tool calls and tool results are present on input/output messages. weaver will validate the format. Do not expect `execute_tool` spans unless the client library itself instruments tool execution — most don't; tool execution is the caller's code.
Multimodal content	`multimodal.py`	A `chat` turn carrying the non-text parts the client accepts (inline image/audio bytes, media URLs, file refs, …), asserting each round-trips onto the messages. Cover only the part types the library emits — see Message-part coverage.
Reasoning	`reasoning.py`	A `chat` turn against a reasoning model where the response carries reasoning/thinking content, asserting a `reasoning` part lands on an output message (and `gen_ai.usage.output_tokens` / reasoning-token attributes if the library records them). Only when the client surfaces reasoning content — see Message-part coverage.
Embeddings	`embedding.py`	An `embeddings` operation.

Agent / orchestration instrumentations:

Scenario	File	Covers
Agent invocation with tooling	`invoke_agent.py`	An `invoke_agent` run that calls at least one tool — expects `invoke_agent` plus the nested `execute_tool` / `chat` spans the framework emits.
Multi-agent orchestration	`multi_agent.py`	One agent handing off to / invoking another — expects nested `invoke_agent` spans under the orchestrator.
Workflows	`invoke_workflow.py`	An `invoke_workflow` run wrapping the agent/tool spans it drives.

Message-part coverage

Weaver validates a part's shape, not which part types a scenario exercised — a text-only scenario leaves the package's image/audio/file/tool mapping unverified. So exercise every non-text part type the library can emit and assert it landed on a message. Cover only what the package instruments: walk its wrappers (the step-6 mapping for a port) for which opentelemetry.util.genai.types parts they produce.

Part `type`	util-genai type	Emitted when the library accepts…
`text`	`TextPart`	plain text (always)
`tool_call` / `tool_call_response`	`ToolCallRequestPart` / `ToolCallResponsePart`	function/tool calling — covered by `tool_calling.py`
`server_tool_call` / `server_tool_call_response`	`ServerToolCallPart` / `ServerToolCallResponsePart`	vendor server-side tools (web_search, code_interpreter, …)
`reasoning`	`ReasoningPart`	reasoning / thinking items
`blob`	`BlobPart`	inline image/audio/video bytes (`modality` distinguishes them)
`uri`	`UriPart`	an external media URL (`modality`)
`file`	`FilePart`	a file reference / id (`modality`)
`generic`	`GenericPart`	a provider item with no semconv mapping — flag, don't drop

Group by shared turn/cassette — typically one multimodal.py for the image/audio/file/url inputs the client accepts, tool_calling.py for tool parts, and reasoning.py for reasoning parts (a reasoning model emits those on output messages, not input). type alone gives blob / uri / file; to tell image from audio from video, read the part's modality with a _part_fields helper returning (type, modality) tuples (defined alongside _part_types in Lib-specific assertions):

def validate(self, report: LiveCheckReport) -> None:
    super().validate(report)
    chat_spans = [
        entry["span"]
        for entry in report["samples"]
        if "span" in entry
        and _attr(entry["span"], "gen_ai.operation.name") == "chat"
    ]
    input_parts = {
        (t, m)
        for span in chat_spans
        for t, m in _part_fields(_attr(span, "gen_ai.input.messages"))
    }
    # e.g. an inline image + an audio URL were sent
    assert ("blob", "image") in input_parts, (
        f"no image blob, saw {input_parts}"
    )
    assert ("uri", "audio") in input_parts, f"no audio uri, saw {input_parts}"

If a part type the library accepts can't round-trip yet (a util-genai/semconv gap), still write the scenario and record it as a declared gap — never silently omit the part.

Scenario modules

Each scenario module defines a subclass of Scenario from opentelemetry.test_util_genai.conformance. It sets the expected_spans / expected_metrics ClassVars and implements run(self, *, tracer_provider, meter_provider, logger_provider, vcr).

expected_spans is a dict[str, int] of exact per-operation span counts (e.g. {"chat": 2} for a tool-calling turn's two LLM calls); the base validate fails on any mismatch. Pin real counts here — don't re-assert them in a validate override. expected_metrics is a tuple of metric names that must each appear at least once. Drive instrumentation through the shared instrument context manager (not instr.instrument() / trace.set_tracer_provider). The runner injects an already-configured vcr, so a cassette-based scenario just calls vcr.use_cassette(...):

# tests/conformance/inference.py
from typing import Any

from opentelemetry.instrumentation.genai.<lib> import <Lib>Instrumentor
from opentelemetry.sdk._logs import LoggerProvider
from opentelemetry.sdk.metrics import MeterProvider
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.test_util_genai.conformance import Scenario
from opentelemetry.test_util_genai.instrumentor import instrument


class InferenceScenario(Scenario):
    expected_spans = {"chat": 1}
    expected_metrics = (
        "gen_ai.client.operation.duration",
        "gen_ai.client.token.usage",
    )

    def run(
        self,
        *,
        tracer_provider: TracerProvider,
        meter_provider: MeterProvider,
        logger_provider: LoggerProvider,
        vcr: Any,
    ) -> None:
        with instrument(
            <Lib>Instrumentor(),
            tracer_provider=tracer_provider,
            logger_provider=logger_provider,
            meter_provider=meter_provider,
            content_capture="SPAN_ONLY",
        ):
            with vcr.use_cassette("inference.yaml"):
                ...  # call the patched API

One operation per scenario. No env vars, no logging config.

VCR cassettes are not required — a transport mock works too. Mock HTTP however the package's unit tests already do, and use the same pattern across every scenario in that package (don't mix cassettes and transport mocks within one lib). If the package mocks the transport (e.g. httpx.MockTransport, respx) instead of replaying cassettes, build the client with that transport inside run() and ignore the injected vcr:

    def run(self, *, tracer_provider, meter_provider, logger_provider, vcr) -> None:
        with instrument(
            <Lib>Instrumentor(),
            tracer_provider=tracer_provider,
            logger_provider=logger_provider,
            meter_provider=meter_provider,
            content_capture="SPAN_ONLY",
        ):
            client = <Lib>(transport=httpx.MockTransport(_handler))  # canned response
            client.<method>(...)  # call the patched API

vcr stays in the signature either way (the runner always passes it).

Declared gaps

Declared gaps go in the expected_violations ClassVar (a tuple of ExpectedViolation), not xfail. run_conformance fails on undeclared weaver violations and on declared violations weaver no longer reports — so a known util-genai/semconv gap is recorded as an expected_violation that fails loudly the moment it's fixed.

When the gap is too big to capture as individual expected_violations — the whole operation can't run yet — skip the entire scenario instead, via pytest.mark.xfail / skip on the parametrize entry in test_conformance.py. Don't invent a one-off reason= string: ask the user to file a tracking bug and update the skip reason= with that issue (e.g. reason="blocked by open-telemetry/...#1234") so the skip is traceable and gets revisited when the bug is fixed. Never drop the scenario file itself — a skipped scenario still records that the operation exists; a deleted one hides it.

Lib-specific assertions

Lib-specific assertions go in a validate(self, report) override on the scenario (call super().validate(report) first) — there is no _local_assertions.py / LocalAssertions hook. Common lib-specific shapes:

Vendor server-tool payloads (code_interpreter, web_search, …).
Vendor-specific finish reasons outside semconv's enum (stop, length, content_filter, tool_call, error).
Provider-specific error.type — exception class names from the underlying SDK.

validate receives the weaver LiveCheckReport. Read span samples from report["samples"] (each entry["span"]["attributes"] is a list of {"name", "value"}) and seen metrics from report["statistics"]["seen_registry_metrics"]. Always call super().validate(report) first so the expected_spans / expected_metrics checks still run:

# tests/conformance/tool_calling.py
from __future__ import annotations

import json
from typing import Any

from opentelemetry.test.weaver_live_check import LiveCheckReport
from opentelemetry.test_util_genai.conformance import Scenario


class ToolCallingScenario(Scenario):
    expected_spans = {"chat": 2}  # initial turn + follow-up with tool results
    expected_metrics = ("gen_ai.client.operation.duration",)

    def run(
        self, *, tracer_provider, meter_provider, logger_provider, vcr
    ) -> None: ...  # drive a tool-calling turn — see "Scenario modules" above

    def validate(self, report: LiveCheckReport) -> None:
        super().validate(report)  # keep the expected_spans / _metrics checks

        # Lib-specific: weaver validates the *shape* of each message part, but
        # not that a tool call actually round-tripped. Assert the model's
        # tool_call landed on an output message and the tool result was fed
        # back on an input message (across the two chat turns).
        chat_spans = [
            entry["span"]
            for entry in report["samples"]
            if "span" in entry
            and _attr(entry["span"], "gen_ai.operation.name") == "chat"
        ]
        assert chat_spans, "no chat span emitted"

        output_part_types = {
            t
            for span in chat_spans
            for t in _part_types(_attr(span, "gen_ai.output.messages"))
        }
        input_part_types = {
            t
            for span in chat_spans
            for t in _part_types(_attr(span, "gen_ai.input.messages"))
        }
        assert "tool_call" in output_part_types, (
            f"expected a tool_call part on an output message, saw {output_part_types}"
        )
        assert "tool_call_response" in input_part_types, (
            f"expected a tool_call_response part on an input message, saw {input_part_types}"
        )


def _attr(span: dict[str, Any], name: str) -> Any:
    for attr in span["attributes"]:
        if attr["name"] == name:
            return attr["value"]
    return None


def _part_types(messages_json: str | None) -> list[str]:
    # gen_ai.{input,output}.messages is a JSON string of
    # [{"role": ..., "parts": [{"type": ..., ...}]}].
    messages = json.loads(messages_json) if messages_json else []
    return [part["type"] for message in messages for part in message["parts"]]


def _part_fields(messages_json: str | None) -> list[tuple[str, str | None]]:
    # Like _part_types, but keeps modality so image/audio/video are
    # distinguishable on blob/uri/file parts (None for parts without one).
    messages = json.loads(messages_json) if messages_json else []
    return [
        (part["type"], part.get("modality"))
        for message in messages
        for part in message["parts"]
    ]

The test_conformance.py runner

tests/test_conformance.py guards collection with pytest.importorskip("opentelemetry.test.weaver_live_check"), parametrizes the scenario instances, and calls run_conformance(scenario, vcr=vcr, weaver=weaver_live_check) — it builds its own providers from the weaver OTLP endpoint, so don't pass providers/exporters:

import pytest

pytest.importorskip("opentelemetry.test.weaver_live_check")

from opentelemetry.test.weaver_live_check import WeaverLiveCheck  # noqa: E402
from opentelemetry.test_util_genai.conformance import (  # noqa: E402
    Scenario,
    run_conformance,
)

from .conformance.embedding import EmbeddingScenario
from .conformance.inference import InferenceScenario


@pytest.mark.parametrize(
    "scenario",
    [InferenceScenario(), EmbeddingScenario()],
    ids=lambda s: type(s).__name__,
)
def test_conformance(
    scenario: Scenario, vcr, weaver_live_check: WeaverLiveCheck
) -> None:
    run_conformance(scenario, vcr=vcr, weaver=weaver_live_check)

Do not write a separate test_weaver_live.py; weaver is already wired through run_conformance. The weaver_live_check fixture skips the test (rather than passing without the gate) only when it can't start (unsupported platform / network) — never wrap it in a try/except, which would silently disable the gate.

Weaver policies

weaver_live_check enforces policies/genai_content_validation.rego (content-attribute JSON shape) and policies/genai_span_validation.rego (span name format, per-op expected attributes) — read these policy files when authoring scenarios; they're authoritative.

weaver_live_check downloads the pinned weaver binary on first use (cached under ~/.cache/otel-conformance/weaver/).

Recorded HTTP (cassettes or transport mock)

A scenario replays one HTTP interaction per operation. Use whichever mechanism the package's unit tests use, consistently across all of its scenarios:

VCR cassette — vcr.use_cassette("<scenario>.yaml"), one committed cassette per operation under tests/cassettes/.
Transport mock — build the SDK client with an httpx.MockTransport (or respx) returning a canned response; no cassette file needed.

Pick the one the lib already follows and don't mix the two within a package.

AI-generated cassettes. Lacking provider access, you may synthesize a cassette from the provider's API reference via AI. Start the cassette with a # TODO: this is generated by AI, re-record comment, mention it in the PR, and open a follow-up issue to re-record it against the real provider in CI.

Running

uv run tox -e py314-test-instrumentation-genai-<lib>-conformance

The *-conformance tox envs target tests/test_conformance.py directly; the regular *-{oldest,latest} envs --ignore it so they don't need the OTLP/gRPC exporter or weaver_live_check.

Navigation Menu

Repository navigation

SKILL.md

SKILL.md

Write GenAI conformance tests

One scenario per operation

Recommended scenarios

Message-part coverage

Scenario modules

Declared gaps

Lib-specific assertions

The test_conformance.py runner

Weaver policies

Recorded HTTP (cassettes or transport mock)

Running

FilesExpand file tree

SKILL.md

Latest commit

History

SKILL.md

File metadata and controls

Write GenAI conformance tests

One scenario per operation

Recommended scenarios

Message-part coverage

Scenario modules

Declared gaps

Lib-specific assertions

The test_conformance.py runner

Weaver policies

Recorded HTTP (cassettes or transport mock)

Running