mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-16 23:41:43 +00:00
44 lines
1.6 KiB
Python
44 lines
1.6 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Final
|
|
|
|
import pytest
|
|
|
|
from .profiler import FunctionTraceEvent
|
|
from .steps import pipeline_projection, trace_depths
|
|
|
|
|
|
def event(event_id: int, function: str, parent_id: int | None = None) -> FunctionTraceEvent:
|
|
return FunctionTraceEvent(event_id, parent_id, function)
|
|
|
|
|
|
def test_projection_keeps_every_call_and_parent() -> None:
|
|
events: Final = (
|
|
event(0, "module.py:1 entry"),
|
|
event(1, "module.py:2 internal_helper", 0),
|
|
event(2, "module.py:3 nested", 1),
|
|
event(3, "module.py:2 internal_helper", 0),
|
|
)
|
|
|
|
steps: Final = pipeline_projection(events)
|
|
|
|
assert tuple((step.id, step.parent_id, step.span, step.raw) for step in steps) == tuple(
|
|
(item.id, item.parent_id, item.function, item.raw) for item in events
|
|
)
|
|
|
|
|
|
def test_projection_preserves_repeated_occurrences() -> None:
|
|
steps: Final = pipeline_projection((event(0, "route"), event(1, "http", 0), event(2, "http", 0)))
|
|
assert [step.span for step in steps] == ["route", "http", "http"]
|
|
|
|
|
|
def test_projection_preserves_multiple_roots() -> None:
|
|
steps: Final = pipeline_projection((event(0, "route"), event(1, "request")))
|
|
assert trace_depths(steps) == {0: 0, 1: 0}
|
|
|
|
|
|
def test_projection_rejects_duplicate_and_unknown_parent_ids() -> None:
|
|
with pytest.raises(ValueError, match="duplicate trace event id"):
|
|
pipeline_projection((event(0, "route"), event(0, "request")))
|
|
with pytest.raises(ValueError, match="unknown or later parent"):
|
|
pipeline_projection((event(1, "request", 0),))
|