Observability

Testing with OpenTelemetry: A Practical Guide to Trace-Based Testing

HelpMeTest

30 Jul 2026 — 6 min read

Observability tooling has matured to the point where the same signals you use to debug production incidents can drive your test assertions. OpenTelemetry (OTel) gives you a vendor-neutral SDK to emit traces, metrics, and logs — and with a small amount of setup you can capture those signals inside your test suite and assert against them directly. This post walks through exactly how to do that.

Why Trace-Based Testing?

Unit tests verify logic in isolation. Integration tests verify that components connect. But neither tells you whether a user request actually flows through your system in the way you designed — whether the database was hit, whether a cache was consulted, whether a third-party call happened exactly once. Traces tell you that.

Trace-based testing flips the model: run the operation, then inspect the trace it produced. Did the expected spans appear? Were the attributes correct? Was latency within bounds? These are assertions you cannot make from return values alone.

Setting Up OTel in Test Environments

The key is using an in-memory span exporter rather than shipping traces to a collector during tests. Both the Java and Python SDKs ship one out of the box.

Python

# conftest.py
import pytest
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter
from opentelemetry.sdk.trace.export import SimpleSpanProcessor

@pytest.fixture(scope="session")
def span_exporter():
    exporter = InMemorySpanExporter()
    provider = TracerProvider()
    provider.add_span_processor(SimpleSpanProcessor(exporter))
    trace.set_tracer_provider(provider)
    return exporter

@pytest.fixture(autouse=True)
def clear_spans(span_exporter):
    span_exporter.clear()
    yield
    # spans available for assertions after yield

Now every test starts with a clean exporter and spans accumulate as your code runs.

Java (JUnit 5)

// OtelTestExtension.java
public class OtelTestExtension implements BeforeEachCallback, AfterEachCallback {
    private static final InMemorySpanExporter exporter = InMemorySpanExporter.create();
    private static final SdkTracerProvider provider = SdkTracerProvider.builder()
        .addSpanProcessor(SimpleSpanProcessor.create(exporter))
        .build();

    static {
        GlobalOpenTelemetry.set(
            OpenTelemetrySdk.builder().setTracerProvider(provider).build()
        );
    }

    @Override
    public void beforeEach(ExtensionContext ctx) {
        exporter.reset();
    }

    public static List<SpanData> getSpans() {
        return exporter.getFinishedSpanItems();
    }
}

@ExtendWith(OtelTestExtension.class)
class OrderServiceTest {
    // spans available via OtelTestExtension.getSpans()
}

Instrumenting Application Code

Your application code should already be instrumented — if it isn't, this is the nudge to start. Use the OTel auto-instrumentation agent for the easy wins (HTTP clients, database drivers, message queues all get instrumented automatically), then add manual spans for business logic.

# order_service.py
from opentelemetry import trace

tracer = trace.get_tracer("order-service")

def process_order(order_id: str, user_id: str):
    with tracer.start_as_current_span("process_order") as span:
        span.set_attribute("order.id", order_id)
        span.set_attribute("user.id", user_id)

        inventory = check_inventory(order_id)
        span.set_attribute("inventory.available", inventory.available)

        if not inventory.available:
            span.set_attribute("order.status", "rejected")
            raise InsufficientInventoryError(order_id)

        charge_result = charge_payment(user_id, inventory.price)
        span.set_attribute("payment.transaction_id", charge_result.transaction_id)
        span.set_attribute("order.status", "confirmed")

        return OrderResult(order_id=order_id, transaction_id=charge_result.transaction_id)

Writing Span Assertions

With the in-memory exporter in place, you can now query finished spans by name, attributes, or parent relationship.

# test_order_service.py
def test_successful_order_creates_expected_spans(span_exporter, mock_inventory, mock_payment):
    mock_inventory.return_value = InventoryResult(available=True, price=49.99)
    mock_payment.return_value = PaymentResult(transaction_id="txn_abc123")

    result = process_order("order-123", "user-456")

    spans = span_exporter.get_finished_spans()
    span_names = [s.name for s in spans]

    # Root span was created
    assert "process_order" in span_names

    # Verify root span attributes
    root = next(s for s in spans if s.name == "process_order")
    assert root.attributes["order.id"] == "order-123"
    assert root.attributes["user.id"] == "user-456"
    assert root.attributes["order.status"] == "confirmed"
    assert root.attributes["payment.transaction_id"] == "txn_abc123"

    # Span completed without error
    assert root.status.status_code == trace.StatusCode.OK

Asserting on Child Spans

def test_inventory_check_is_child_of_process_order(span_exporter, mock_inventory, mock_payment):
    mock_inventory.return_value = InventoryResult(available=True, price=10.0)
    mock_payment.return_value = PaymentResult(transaction_id="txn_xyz")

    process_order("order-999", "user-001")

    spans = span_exporter.get_finished_spans()
    root = next(s for s in spans if s.name == "process_order")
    inventory_span = next((s for s in spans if s.name == "check_inventory"), None)

    assert inventory_span is not None
    # Child span's parent should be the root span
    assert inventory_span.parent.span_id == root.context.span_id

Error Path Assertions

def test_insufficient_inventory_records_error_on_span(span_exporter, mock_inventory):
    mock_inventory.return_value = InventoryResult(available=False, price=0)

    with pytest.raises(InsufficientInventoryError):
        process_order("order-000", "user-789")

    spans = span_exporter.get_finished_spans()
    root = next(s for s in spans if s.name == "process_order")

    assert root.attributes["order.status"] == "rejected"
    assert root.status.status_code == trace.StatusCode.ERROR
    # OTel records exception events on spans
    events = [e for e in root.events if e.name == "exception"]
    assert len(events) == 1
    assert "InsufficientInventoryError" in events[0].attributes["exception.type"]

Trace Context Propagation in Tests

When testing HTTP handlers or message consumers, you want to verify that incoming trace context is correctly propagated — not just that spans are created, but that they join the right trace.

# test_http_handler.py
from opentelemetry.propagate import inject, extract
from opentelemetry.trace.propagation.tracecontext import TraceContextTextMapPropagator

def test_handler_propagates_incoming_trace_context(client, span_exporter):
    # Simulate an upstream service sending trace context
    propagator = TraceContextTextMapPropagator()
    carrier = {}
    
    with trace.get_tracer("test").start_as_current_span("upstream-span") as upstream:
        propagator.inject(carrier)
        upstream_trace_id = upstream.get_span_context().trace_id

    response = client.post(
        "/api/orders",
        json={"item": "widget", "qty": 1},
        headers={"traceparent": carrier["traceparent"]}
    )

    assert response.status_code == 200

    spans = span_exporter.get_finished_spans()
    handler_span = next(s for s in spans if s.name == "POST /api/orders")

    # Handler span must belong to the upstream trace
    assert handler_span.context.trace_id == upstream_trace_id

OTel Collector in CI

For integration tests that exercise real services (databases, queues), you may want a lightweight collector rather than an in-memory exporter. The OTel Collector can be run as a sidecar in Docker Compose:

# docker-compose.test.yml
services:
  app:
    build: .
    environment:
      OTEL_EXPORTER_OTLP_ENDPOINT: http://otel-collector:4317
      OTEL_SERVICE_NAME: my-app-test
    depends_on:
      - otel-collector

  otel-collector:
    image: otel/opentelemetry-collector-contrib:0.96.0
    command: ["--config=/etc/otel-collector-config.yaml"]
    volumes:
      - ./otel-collector-test.yaml:/etc/otel-collector-config.yaml
    ports:
      - "4317:4317"   # OTLP gRPC
      - "55679:55679" # zpages for debugging

# otel-collector-test.yaml
receivers:
  otlp:
    protocols:
      grpc:
        endpoint: 0.0.0.0:4317

exporters:
  file:
    path: /tmp/spans.json
  logging:
    verbosity: detailed

service:
  pipelines:
    traces:
      receivers: [otlp]
      exporters: [file, logging]

Your test suite can then parse /tmp/spans.json for assertions after the test run, or use the zpages endpoint (http://localhost:55679/debug/tracez) for manual inspection during development.

Sampling Considerations

Production systems often sample traces (e.g., 1% of requests). In tests you always want 100% sampling. Set this explicitly:

# Never rely on environment defaults in tests
from opentelemetry.sdk.trace.sampling import ALWAYS_ON

provider = TracerProvider(sampler=ALWAYS_ON)

In Java:

SdkTracerProvider.builder()
    .setSampler(Sampler.alwaysOn())
    .addSpanProcessor(SimpleSpanProcessor.create(exporter))
    .build();

If your application reads the OTEL_TRACES_SAMPLER environment variable, override it in your test configuration:

OTEL_TRACES_SAMPLER=always_on pytest tests/

Practical Patterns

Pattern 1: Span count assertions for N+1 detection

def test_batch_order_does_not_cause_n_plus_one_db_queries(span_exporter, db):
    orders = [create_order(i) for i in range(10)]
    
    process_batch(orders)
    
    db_spans = [s for s in span_exporter.get_finished_spans() 
                if s.name.startswith("db.")]
    
    # Should be 1 batch query, not 10 individual queries
    assert len(db_spans) == 1, f"Expected 1 DB span, got {len(db_spans)}"

Pattern 2: Latency budgets in tests

def test_order_processing_completes_within_budget(span_exporter, mock_deps):
    process_order("order-lat", "user-lat")
    
    root = next(s for s in span_exporter.get_finished_spans() 
                if s.name == "process_order")
    
    duration_ms = (root.end_time - root.start_time) / 1_000_000
    assert duration_ms < 200, f"Order processing took {duration_ms}ms, budget is 200ms"

Pattern 3: Verifying cache hit/miss signals

def test_second_request_hits_cache(span_exporter, cache, db):
    # First request — should hit DB
    get_product("prod-1")
    spans_first = span_exporter.get_finished_spans()
    span_exporter.clear()
    
    # Second request — should hit cache only
    get_product("prod-1")
    spans_second = span_exporter.get_finished_spans()
    
    db_spans_second = [s for s in spans_second if "db." in s.name]
    cache_spans_second = [s for s in spans_second if "cache.get" in s.name]
    
    assert len(db_spans_second) == 0, "DB should not be queried on cache hit"
    assert len(cache_spans_second) == 1
    assert cache_spans_second[0].attributes.get("cache.hit") is True

Integrating with Your CI Pipeline

Add a span assertion step to your GitHub Actions workflow:

# .github/workflows/test.yml
- name: Run tests with OTel
  env:
    OTEL_TRACES_SAMPLER: always_on
    OTEL_SERVICE_NAME: ${{ github.repository }}-test
  run: pytest tests/ -v --tb=short

- name: Upload span artifacts on failure
  if: failure()
  uses: actions/upload-artifact@v3
  with:
    name: test-spans
    path: /tmp/spans.json

When a test fails because a span is missing or an attribute is wrong, the uploaded spans.json gives you the full trace for debugging — far more context than a stack trace alone.

Conclusion

Trace-based testing with OpenTelemetry adds a dimension to your test suite that neither unit tests nor integration tests can provide: verification that the actual execution path matches your design. The in-memory exporter makes this cheap to set up, and span assertions read naturally alongside your existing test assertions. Start by adding the exporter to your test fixtures, instrument the one or two critical paths in your application, and write span assertions for the behaviors you care most about. The investment pays off the first time a refactor silently changes how many database queries a handler makes.