empty assertions

This commit is contained in:
Alex Hall 2025-03-21 18:26:04 +02:00
parent a00b61f355
commit 6b509e33f6
4 changed files with 30 additions and 29 deletions

View file

@ -9,7 +9,7 @@ from agents import Agent, RunConfig, Runner, trace
from .fake_model import FakeModel
from .test_responses import get_text_message
from .testing_processor import fetch_normalized_spans, fetch_traces
from .testing_processor import assert_no_traces, fetch_normalized_spans, fetch_traces
@pytest.mark.asyncio
@ -164,7 +164,7 @@ async def test_parent_disabled_trace_disabled_agent_trace():
await Runner.run(agent, input="first_test")
assert fetch_normalized_spans() == snapshot([])
assert_no_traces()
@pytest.mark.asyncio
@ -178,7 +178,7 @@ async def test_manual_disabling_works():
await Runner.run(agent, input="first_test", run_config=RunConfig(tracing_disabled=True))
assert fetch_normalized_spans() == snapshot([])
assert_no_traces()
@pytest.mark.asyncio
@ -370,8 +370,7 @@ async def test_parent_disabled_trace_disables_streaming_agent_trace():
async for _ in x.stream_events():
pass
traces = fetch_traces()
assert len(traces) == 0, f"Expected 0 traces, got {len(traces)}"
assert_no_traces()
@pytest.mark.asyncio
@ -392,5 +391,4 @@ async def test_manual_streaming_disabling_works():
async for _ in x.stream_events():
pass
traces = fetch_traces()
assert len(traces) == 0, f"Expected 0 traces, got {len(traces)}"
assert_no_traces()

View file

@ -7,7 +7,7 @@ from agents import ModelSettings, ModelTracing, OpenAIResponsesModel, trace
from agents.tracing.span_data import ResponseSpanData
from tests import fake_model
from .testing_processor import fetch_normalized_spans, fetch_ordered_spans
from .testing_processor import fetch_normalized_spans, fetch_ordered_spans, assert_no_spans
class DummyTracing:
@ -89,9 +89,8 @@ async def test_non_data_tracing_doesnt_set_response_id(monkeypatch):
[{"workflow_name": "test", "children": [{"type": "response"}]}]
)
spans = fetch_ordered_spans()
assert len(spans) == 1
assert spans[0].span_data.response is None
[span] = fetch_ordered_spans()
assert span.span_data.response is None
@pytest.mark.allow_call_model_methods
@ -116,9 +115,7 @@ async def test_disable_tracing_does_not_create_span(monkeypatch):
assert fetch_normalized_spans() == snapshot([{"workflow_name": "test"}])
spans = fetch_ordered_spans()
assert len(spans) == 0
assert_no_spans()
@pytest.mark.allow_call_model_methods
@pytest.mark.asyncio
@ -190,10 +187,9 @@ async def test_stream_non_data_tracing_doesnt_set_response_id(monkeypatch):
[{"workflow_name": "test", "children": [{"type": "response"}]}]
)
spans = fetch_ordered_spans()
assert len(spans) == 1
assert isinstance(spans[0].span_data, ResponseSpanData)
assert spans[0].span_data.response is None
[span] = fetch_ordered_spans()
assert isinstance(span.span_data, ResponseSpanData)
assert span.span_data.response is None
@pytest.mark.allow_call_model_methods
@ -226,5 +222,4 @@ async def test_stream_disabled_tracing_doesnt_create_span(monkeypatch):
assert fetch_normalized_spans() == snapshot([{"workflow_name": "test"}])
spans = fetch_ordered_spans()
assert len(spans) == 0
assert_no_spans()

View file

@ -24,6 +24,7 @@ from .testing_processor import (
fetch_normalized_spans,
fetch_ordered_spans,
fetch_traces,
assert_no_traces,
)
### HELPERS
@ -281,10 +282,7 @@ def disabled_tracing():
def test_disabled_tracing():
disabled_tracing()
spans, traces = fetch_ordered_spans(), fetch_traces()
assert len(spans) == 0
assert len(traces) == 0
assert_no_traces()
def enabled_trace_disabled_span():
@ -374,9 +372,7 @@ async def test_noop_span_doesnt_record():
with custom_span(name="span_1") as span:
span.set_error(SpanError(message="test", data={}))
spans, traces = fetch_ordered_spans(), fetch_traces()
assert len(spans) == 0
assert len(traces) == 0
assert_no_traces()
assert t.export() is None
assert span.export() is None

View file

@ -80,6 +80,19 @@ def fetch_events() -> list[TestSpanProcessorEvent]:
return SPAN_PROCESSOR_TESTING._events
def assert_no_spans():
spans = fetch_ordered_spans()
if spans:
raise AssertionError(f"Expected 0 spans, got {len(spans)}")
def assert_no_traces():
traces = fetch_traces()
if traces:
raise AssertionError(f"Expected 0 traces, got {len(traces)}")
assert_no_spans()
def fetch_normalized_spans(keep_span_id: bool = False):
nodes: dict[tuple[str, str | None], dict[str, Any]] = {}
traces = []
@ -92,8 +105,7 @@ def fetch_normalized_spans(keep_span_id: bool = False):
nodes[(trace_obj.trace_id, None)] = trace
traces.append(trace)
if not traces:
assert not fetch_ordered_spans()
assert traces, "Use assert_no_traces() to check for empty traces"
for span_obj in fetch_ordered_spans():
span = span_obj.export()