Tests creation of spans/logs in an E2E runner invocation. Asserts the entire telemetry schema (spans + attributes + per-span logs) matches the hand-written expected shape for the given semconv + content-capture configuration.
(
case: FunctionalTestCase,
monkeypatch: pytest.MonkeyPatch,
)
| 64 | @pytest.mark.parametrize("case", ALL_CASES, ids=lambda c: c.test_id) |
| 65 | @pytest.mark.asyncio |
| 66 | async def test_telemetry_schema( |
| 67 | case: FunctionalTestCase, |
| 68 | monkeypatch: pytest.MonkeyPatch, |
| 69 | ) -> None: |
| 70 | """Tests creation of spans/logs in an E2E runner invocation. |
| 71 | |
| 72 | Asserts the entire telemetry schema (spans + attributes + per-span logs) |
| 73 | matches the hand-written expected shape for the given semconv + |
| 74 | content-capture configuration. |
| 75 | """ |
| 76 | case.apply_env(monkeypatch) |
| 77 | |
| 78 | span_exporter = InMemorySpanExporter() |
| 79 | log_exporter = InMemoryLogRecordExporter() |
| 80 | install_telemetry(monkeypatch, span_exporter, log_exporter) |
| 81 | |
| 82 | await run_agent_scenario(build_test_runner()) |
| 83 | |
| 84 | digest = SpanDigest.build( |
| 85 | span_exporter.get_finished_spans(), |
| 86 | log_exporter.get_finished_logs(), |
| 87 | ) |
| 88 | assert digest == case.expected_root |
| 89 | |
| 90 | |
| 91 | @pytest.mark.asyncio |
nothing calls this directly
no test coverage detected