opentelemetry-trace-assertions

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

_exporter = InMemorySpanExporter()
_provider = TracerProvider()
_provider.add_span_processor(SimpleSpanProcessor(_exporter))
trace.set_tracer_provider(_provider)        # once per process

@pytest.fixture
def spans():
    _exporter.clear()                        # each test sees only its own spans
    def by_name(name):
        found = [s for s in _exporter.get_finished_spans() if s.name == name]
        assert found, f"no finished span named {name!r}"
        return found[-1]
    yield by_name
    _exporter.clear()

from opentelemetry.trace import SpanKind, StatusCode

def test_order_create_span(spans):
    create_order(items=[item])

    order = spans("order.create")
    assert order.attributes["order.item_count"] == 1
    assert order.status.status_code != StatusCode.ERROR
    assert order.kind == SpanKind.INTERNAL

def test_rates_provider_down_is_an_error(spans, provider_down):
    price = convert(100, "EUR")              # falls back to the cached rate
    assert price == 8600                     # the conversion still succeeds

    fetch = spans("rates.fetch")
    assert fetch.kind == SpanKind.CLIENT     # an outgoing call, not in-process
    assert fetch.status.status_code == StatusCode.ERROR
    assert fetch.attributes["error.type"] == "ConnectionRefusedError"
    assert spans("pricing.convert").status.status_code != StatusCode.ERROR

def test_db_span_is_child_of_order(spans):
    create_order(items=[item])
    order, db = spans("order.create"), spans("db.query")

    assert db.context.trace_id == order.context.trace_id
    assert db.parent is not None, "db.query started a new trace"
    assert db.parent.span_id == order.context.span_id

def test_charge_call_attributes(spans):
    charge_card(order)
    http = spans("POST")                     # "{method}" or "{method} {url.template}"
    assert http.kind == SpanKind.CLIENT
    assert http.attributes["http.request.method"] == "POST"
    assert http.attributes["url.full"].startswith("https://gateway.example.com/charges")
    assert http.attributes["server.address"] == "gateway.example.com"
    assert http.attributes["http.response.status_code"] == 201

import { InMemorySpanExporter, SimpleSpanProcessor } from '@opentelemetry/sdk-trace-base';
import { NodeTracerProvider } from '@opentelemetry/sdk-trace-node';
import { SpanKind, SpanStatusCode } from '@opentelemetry/api';

const exporter = new InMemorySpanExporter();
new NodeTracerProvider({ spanProcessors: [new SimpleSpanProcessor(exporter)] }).register();

const byName = (name: string) => {
  const span = exporter.getFinishedSpans().findLast((s) => s.name === name);
  if (!span) throw new Error(`no finished span named ${name}`);
  return span;
};

beforeEach(() => exporter.reset());

test('rates.fetch records the outage', async () => {
  await convert(100, 'EUR');
  const fetch = byName('rates.fetch');
  expect(fetch.kind).toBe(SpanKind.CLIENT);
  expect(fetch.status.code).toBe(SpanStatusCode.ERROR);
  expect(fetch.attributes['error.type']).toBe('ECONNREFUSED');
});

@RegisterExtension
static final OpenTelemetryExtension otel = OpenTelemetryExtension.create();

@Test
void orderCreateEmitsSpan() {
    new OrderService(otel.getOpenTelemetry()).create(order);

    SpanData span = otel.getSpans().stream()
        .filter(s -> s.getName().equals("order.create"))
        .findFirst().orElseThrow();
    assertThat(span.getAttributes().get(AttributeKey.longKey("order.item_count"))).isEqualTo(1L);
}

- run: pip install -r requirements-test.txt   # opentelemetry-sdk pinned there
- run: pytest tests -v