mirror of
https://github.com/BerriAI/litellm.git
synced 2026-10-11 03:38:38 +00:00
0b0fdedd1e
* test(e2e): add enum values, auto-discovering label gates and secret hiding for e2e metadata * docs(e2e): name every markerless harness test file that carries no Subject * test(e2e): keep the step discovery comprehensions to one for clause
844 lines
34 KiB
Python
844 lines
34 KiB
Python
"""The e2e test metadata: `@meta(Subject(...))` properties and the step recorder's edge cases.
|
|
|
|
Harness logic, so it lives here rather than under tests/e2e, which holds only
|
|
tests that drive a live proxy. The harness modules are imported off
|
|
``PYTHONPATH=tests/e2e``, the way the Code Quality workflow's
|
|
test_e2e_metadata step runs this file. Call order, the failing test's last step,
|
|
the per-test reset and the JUnit attach are pinned end to end in
|
|
test_e2e_junit_report.py.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ast
|
|
import importlib
|
|
import inspect
|
|
import re
|
|
import string
|
|
import sys
|
|
import threading
|
|
import warnings
|
|
from collections import Counter
|
|
from collections.abc import Callable, Generator, Iterator, Mapping
|
|
from contextlib import contextmanager
|
|
from dataclasses import MISSING, dataclass, fields, is_dataclass, replace
|
|
from functools import cache, reduce
|
|
from itertools import chain
|
|
from pathlib import Path
|
|
from types import ModuleType, UnionType
|
|
from typing import Final, Union, cast, get_args, get_origin, get_type_hints
|
|
|
|
import pytest
|
|
from e2e_metadata import (
|
|
MASK,
|
|
MAX_STEPS,
|
|
STEP_FRAMES,
|
|
STEPS,
|
|
Capability,
|
|
Domain,
|
|
Mode,
|
|
Provider,
|
|
Route,
|
|
StepRecorder,
|
|
Subject,
|
|
environment_secrets,
|
|
meta,
|
|
step,
|
|
subject_properties,
|
|
)
|
|
from junit_properties import package_from_nodeid, result_properties, source_from_item
|
|
from pydantic import BaseModel, Field
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def empty_step_log() -> Generator[None]:
|
|
"""Each test starts from an empty log and leaves none behind, as conftest's
|
|
`pytest_runtest_setup` hook arranges for every live test."""
|
|
STEPS.reset()
|
|
yield
|
|
STEPS.reset()
|
|
|
|
|
|
def collected_item(request: pytest.FixtureRequest, name: str) -> pytest.Item:
|
|
return next(item for item in request.session.items if item.path == request.path and item.name == name)
|
|
|
|
|
|
def fixed_prefix(item: pytest.Item, covers: str) -> tuple[tuple[str, str], ...]:
|
|
"""Spelled out rather than taken from `result_properties`, so a change to either fails a test."""
|
|
return (
|
|
("package", package_from_nodeid(item.nodeid)),
|
|
("covers", covers),
|
|
("source", source_from_item(item)),
|
|
)
|
|
|
|
|
|
class TestSubjectProperties:
|
|
"""Markers go on via `request.applymarker` so the coverage registry's collect-only pass never sees them."""
|
|
|
|
def test_every_declared_field_becomes_a_property_in_field_order(self, request: pytest.FixtureRequest) -> None:
|
|
test = type(self).test_every_declared_field_becomes_a_property_in_field_order
|
|
request.applymarker(
|
|
meta(
|
|
Subject(
|
|
domain=Domain.SPEND_BUDGETS,
|
|
route=Route.CHAT_COMPLETIONS,
|
|
providers=(Provider.GEMINI, Provider.ANTHROPIC),
|
|
models=("gemini-2.5-flash", "claude-haiku-4-5"),
|
|
capabilities=(Capability.VISION, Capability.FUNCTION_CALLING, Capability.VISION),
|
|
mode=Mode.NONSTREAM,
|
|
)
|
|
)
|
|
)
|
|
assert subject_properties(collected_item(request, test.__name__)) == (
|
|
("domain", "spend-budgets"),
|
|
("route", "chat_completions"),
|
|
("provider", "anthropic"),
|
|
("provider", "gemini"),
|
|
("model", "claude-haiku-4-5"),
|
|
("model", "gemini-2.5-flash"),
|
|
("capability", "function_calling"),
|
|
("capability", "vision"),
|
|
("mode", "nonstream"),
|
|
)
|
|
|
|
def test_one_provider_with_three_models_pairs_nothing(self, request: pytest.FixtureRequest) -> None:
|
|
test = type(self).test_one_provider_with_three_models_pairs_nothing
|
|
request.applymarker(
|
|
meta(
|
|
Subject(
|
|
providers=(Provider.BEDROCK,),
|
|
models=("claude-sonnet-4-5", "claude-opus-4-7", "claude-haiku-4-5"),
|
|
)
|
|
)
|
|
)
|
|
assert subject_properties(collected_item(request, test.__name__)) == (
|
|
("provider", "bedrock"),
|
|
("model", "claude-haiku-4-5"),
|
|
("model", "claude-opus-4-7"),
|
|
("model", "claude-sonnet-4-5"),
|
|
)
|
|
|
|
def test_an_empty_plural_field_emits_nothing(self, request: pytest.FixtureRequest) -> None:
|
|
test = type(self).test_an_empty_plural_field_emits_nothing
|
|
request.applymarker(meta(Subject(domain=Domain.MANAGEMENT)))
|
|
assert subject_properties(collected_item(request, test.__name__)) == (("domain", "management"),)
|
|
|
|
def test_scalar_property_names_are_the_dataclass_field_names(self, request: pytest.FixtureRequest) -> None:
|
|
test = type(self).test_scalar_property_names_are_the_dataclass_field_names
|
|
request.applymarker(meta(Subject(domain=Domain.UNKNOWN, route=Route.HEALTH, mode=Mode.STREAM)))
|
|
declared = tuple(field.name for field in fields(Subject))
|
|
emitted = tuple(name for name, _ in subject_properties(collected_item(request, test.__name__)))
|
|
assert emitted == tuple(name for name in declared if name in {"domain", "route", "mode"})
|
|
|
|
def test_every_plural_field_is_deduped_and_sorted_at_declaration(self) -> None:
|
|
subject = Subject(
|
|
providers=(Provider.OPENAI, Provider.ANTHROPIC, Provider.OPENAI),
|
|
models=("gpt-5.5", "claude-haiku-4-5", "gpt-5.5"),
|
|
capabilities=(Capability.VISION, Capability.REASONING, Capability.VISION),
|
|
)
|
|
assert subject.providers == (Provider.ANTHROPIC, Provider.OPENAI)
|
|
assert subject.models == ("claude-haiku-4-5", "gpt-5.5")
|
|
assert subject.capabilities == (Capability.REASONING, Capability.VISION)
|
|
|
|
@pytest.mark.parametrize(
|
|
("field", "value"),
|
|
[
|
|
("models", "gpt-5.5"),
|
|
("models", ["gpt-5.5"]),
|
|
("providers", Provider.OPENAI),
|
|
("providers", [Provider.OPENAI]),
|
|
("capabilities", Capability.VISION),
|
|
("capabilities", frozenset({Capability.VISION})),
|
|
],
|
|
)
|
|
def test_a_plural_field_refuses_anything_but_a_tuple(self, field: str, value: object) -> None:
|
|
"""`replace` is the untyped way in, since the typed constructor would not let the test spell the mistake."""
|
|
with pytest.raises(TypeError, match=rf"Subject\.{field} must be a tuple"):
|
|
_ = replace(Subject(), **{field: value})
|
|
|
|
@pytest.mark.parametrize(
|
|
("field", "value", "member_type"),
|
|
[
|
|
("providers", ("openai",), "Provider"),
|
|
("capabilities", ("vision",), "Capability"),
|
|
("models", (5,), "str"),
|
|
],
|
|
)
|
|
def test_a_plural_field_refuses_a_member_of_the_wrong_type(
|
|
self, field: str, value: object, member_type: str
|
|
) -> None:
|
|
with pytest.raises(TypeError, match=rf"Subject\.{field} takes {member_type} members"):
|
|
_ = replace(Subject(), **{field: value})
|
|
|
|
def test_a_blank_model_is_dropped_rather_than_refused(self) -> None:
|
|
"""A blank env override must cost one missing property, not collection of the whole module."""
|
|
assert Subject(models=("", "gpt-5.5")).models == ("gpt-5.5",)
|
|
|
|
def test_the_typed_marker_only_ever_appends_to_the_fixed_prefix(self, request: pytest.FixtureRequest) -> None:
|
|
test = type(self).test_the_typed_marker_only_ever_appends_to_the_fixed_prefix
|
|
request.applymarker(pytest.mark.covers("quota_management.budget.key.blocks_over_limit"))
|
|
request.applymarker(meta(Subject(route=Route.SPEND_REPORTING)))
|
|
item = collected_item(request, test.__name__)
|
|
assert result_properties(item) == fixed_prefix(item, "quota_management.budget.key.blocks_over_limit") + (
|
|
("route", "spend_reporting"),
|
|
)
|
|
|
|
def test_a_test_with_only_the_old_string_covers_is_unchanged(self, request: pytest.FixtureRequest) -> None:
|
|
test = type(self).test_a_test_with_only_the_old_string_covers_is_unchanged
|
|
request.applymarker(pytest.mark.covers("llm.responses.openai.tool_use.nonstream.works"))
|
|
item = collected_item(request, test.__name__)
|
|
assert result_properties(item) == fixed_prefix(item, "llm.responses.openai.tool_use.nonstream.works")
|
|
|
|
def test_a_test_with_neither_marker_carries_only_the_prefix(self, request: pytest.FixtureRequest) -> None:
|
|
test = type(self).test_a_test_with_neither_marker_carries_only_the_prefix
|
|
item = collected_item(request, test.__name__)
|
|
assert subject_properties(item) == ()
|
|
assert result_properties(item) == fixed_prefix(item, "")
|
|
|
|
def test_a_marker_carrying_something_other_than_a_subject_emits_nothing(
|
|
self, request: pytest.FixtureRequest
|
|
) -> None:
|
|
test = type(self).test_a_marker_carrying_something_other_than_a_subject_emits_nothing
|
|
request.applymarker(pytest.mark.meta("spend-budgets"))
|
|
assert subject_properties(collected_item(request, test.__name__)) == ()
|
|
|
|
|
|
class TestProviderMirrorsLitellm:
|
|
"""`Provider` copies `LlmProviders` values so collecting tests/e2e never needs litellm; skips where it is absent."""
|
|
|
|
def test_every_provider_value_is_a_real_litellm_provider(self) -> None:
|
|
try:
|
|
from litellm.types.utils import LlmProviders
|
|
except ImportError: # pragma: no cover - the runner image's shape
|
|
pytest.skip("litellm is not importable here, which is the property under test")
|
|
known = {str(member.value) for member in LlmProviders}
|
|
unknown = sorted(member.value for member in Provider if member.value not in known)
|
|
assert not unknown, f"not LlmProviders values: {unknown}"
|
|
|
|
|
|
E2E_DIR: Final = Path(__file__).resolve().parents[1] / "e2e"
|
|
|
|
|
|
def _hand_typed_models(path: Path) -> Iterator[str]:
|
|
for node in ast.walk(ast.parse(path.read_text())):
|
|
match node:
|
|
case ast.Call(func=ast.Name(id="Subject"), keywords=keywords):
|
|
for keyword in keywords:
|
|
match keyword:
|
|
case ast.keyword(arg="models", value=ast.Tuple(elts=models)):
|
|
yield from (
|
|
f"{path.relative_to(E2E_DIR)}:{model.lineno} {model.value!r}"
|
|
for model in models
|
|
if isinstance(model, ast.Constant)
|
|
)
|
|
case _:
|
|
pass
|
|
case _:
|
|
pass
|
|
|
|
|
|
def test_a_declared_model_names_the_constant_the_test_drives() -> None:
|
|
offenders: Final = tuple(
|
|
offender for path in sorted(E2E_DIR.rglob("*.py")) for offender in _hand_typed_models(path)
|
|
)
|
|
assert offenders == ()
|
|
|
|
|
|
class TestStepRecording:
|
|
"""`@step`-decorated harness helpers append to the running test's story as
|
|
they execute.
|
|
|
|
Each test here starts from an empty log because `empty_step_log` resets the
|
|
recorder first, the same reset conftest's `pytest_runtest_setup` gives every
|
|
live test.
|
|
"""
|
|
|
|
def test_a_decorated_helper_still_returns_exactly_what_it_did(self) -> None:
|
|
"""`@step` records, it does not intercept: arguments, return value and
|
|
`__name__` all survive it, so decorating a live harness method cannot
|
|
change what the test observes."""
|
|
|
|
@step("POST /chat/completions")
|
|
def chat(key: str, *, model: str) -> str:
|
|
return f"{key}:{model}"
|
|
|
|
assert chat("sk-x", model="gpt-5.5") == "sk-x:gpt-5.5"
|
|
assert chat.__name__ == "chat"
|
|
|
|
def test_a_poll_loop_is_one_step_in_the_story_not_fifty(self) -> None:
|
|
@step("poll /spend/logs for the request id")
|
|
def poll() -> None:
|
|
return None
|
|
|
|
for _ in range(20):
|
|
poll()
|
|
assert STEPS.taken() == ("poll /spend/logs for the request id",)
|
|
|
|
def test_the_same_label_recorded_again_later_is_a_new_step(self) -> None:
|
|
"""Only CONSECUTIVE duplicates collapse; a helper called again after
|
|
something else happened is a genuine second beat of the story."""
|
|
STEPS.record("POST /chat/completions")
|
|
STEPS.record("poll /spend/logs")
|
|
STEPS.record("POST /chat/completions")
|
|
assert STEPS.taken() == ("POST /chat/completions", "poll /spend/logs", "POST /chat/completions")
|
|
|
|
def test_a_full_log_keeps_the_latest_steps_so_the_last_is_where_the_test_died(self) -> None:
|
|
"""A load test cannot bury the story in thousands of entries, and the cap
|
|
drops from the front: the step a test died on is the newest, so it is the
|
|
one that has to survive. The leading line says the story is partial."""
|
|
for index in range(MAX_STEPS + 10):
|
|
STEPS.record(f"call {index}")
|
|
assert STEPS.taken() == (
|
|
"(10 earlier steps not recorded)",
|
|
*(f"call {index}" for index in range(10, MAX_STEPS + 10)),
|
|
)
|
|
|
|
def test_reset_forgets_what_a_full_log_dropped(self) -> None:
|
|
for index in range(MAX_STEPS + 1):
|
|
STEPS.record(f"call {index}")
|
|
STEPS.reset()
|
|
STEPS.record("register deployment")
|
|
assert STEPS.taken() == ("register deployment",)
|
|
|
|
def test_whitespace_is_normalized_and_an_empty_label_records_nothing(self) -> None:
|
|
STEPS.record(" POST /chat/completions\n ")
|
|
STEPS.record(" ")
|
|
assert STEPS.taken() == ("POST /chat/completions",)
|
|
|
|
def test_a_decorated_helper_warns_at_its_caller_with_step_frames(self) -> None:
|
|
"""`stacklevel` counts frames, and the wrapper is one of them: a cleanup
|
|
helper that warns about its caller would otherwise report every warning at
|
|
e2e_metadata.py. Pins `STEP_FRAMES` to the frames the wrapper really adds."""
|
|
|
|
@step("delete team")
|
|
def delete_team() -> None:
|
|
warnings.warn("delete_team('t') failed", stacklevel=2 + STEP_FRAMES)
|
|
|
|
with warnings.catch_warnings(record=True) as caught:
|
|
warnings.simplefilter("always")
|
|
delete_team()
|
|
assert [Path(warning.filename).name for warning in caught] == [Path(__file__).name]
|
|
|
|
|
|
class _KeyBody(BaseModel):
|
|
models: list[str] = []
|
|
rpm_limit: int | None = None
|
|
tpm_limit: int | None = None
|
|
team_id: str | None = None
|
|
api_key: str | None = Field(default=None, repr=False)
|
|
|
|
|
|
class _Params(BaseModel):
|
|
model: str
|
|
api_key: str | None = Field(default=None, repr=False)
|
|
|
|
|
|
class _DeploymentBody(BaseModel):
|
|
model_name: str
|
|
params: _Params
|
|
|
|
|
|
def _alternatives(annotation: object) -> tuple[object, ...]:
|
|
if not (isinstance(annotation, UnionType) or get_origin(annotation) is Union):
|
|
return (annotation,)
|
|
return tuple(arg for arg in cast("tuple[object, ...]", get_args(annotation)) if arg is not type(None))
|
|
|
|
|
|
def _evaluated(annotation: object, module: str) -> object:
|
|
holder: Final = type("Hint", (), {"__annotations__": {"value": annotation}, "__module__": module})
|
|
hints: Final[Mapping[str, object]] = get_type_hints(holder)
|
|
return hints["value"]
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class StepHelper:
|
|
path: Path
|
|
qualname: str
|
|
label: str
|
|
owner: object
|
|
function: Callable[..., object]
|
|
|
|
@property
|
|
def where(self) -> str:
|
|
return f"{self.path.relative_to(E2E_DIR)}::{self.qualname}"
|
|
|
|
def hint(self, placeholder: str) -> object:
|
|
root: Final = placeholder.split(".")[0]
|
|
if root == "self":
|
|
return self.owner
|
|
annotation: Final = cast("object", inspect.signature(self.function).parameters[root].annotation)
|
|
return _evaluated(annotation, self.function.__module__)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class FieldRead:
|
|
always_set: bool
|
|
annotation: object
|
|
|
|
|
|
def _model_field(owner: type[BaseModel], attribute: str) -> FieldRead | None:
|
|
info: Final = owner.model_fields.get(attribute)
|
|
if info is None:
|
|
return None
|
|
return FieldRead(info.is_required() or (info.default_factory is None and info.default is not None), info.annotation)
|
|
|
|
|
|
def _dataclass_field(owner: type, attribute: str) -> FieldRead | None:
|
|
found: Final = next((field for field in fields(owner) if field.name == attribute), None)
|
|
if found is None:
|
|
return None
|
|
always_set: Final = found.default_factory is MISSING and found.default is not None
|
|
return FieldRead(always_set, _evaluated(found.type, owner.__module__))
|
|
|
|
|
|
def _attribute(owner: object, attribute: str) -> FieldRead | None:
|
|
if isinstance(owner, type) and issubclass(owner, BaseModel):
|
|
return _model_field(owner, attribute)
|
|
if isinstance(owner, type) and is_dataclass(owner):
|
|
return _dataclass_field(owner, attribute)
|
|
return None
|
|
|
|
|
|
STEP_DECORATOR: Final = re.compile(r"^[ \t]*@step\(", re.MULTILINE)
|
|
|
|
|
|
def _is_step(decorator: ast.expr) -> bool:
|
|
match decorator:
|
|
case ast.Call(func=ast.Name(id="step")):
|
|
return True
|
|
case _:
|
|
return False
|
|
|
|
|
|
def _decorated_defs(body: list[ast.stmt], prefix: str = "") -> Iterator[str]:
|
|
for node in body:
|
|
match node:
|
|
case ast.ClassDef(name=name, body=inner):
|
|
yield from _decorated_defs(inner, f"{prefix}{name}.")
|
|
case (
|
|
ast.FunctionDef(name=name, decorator_list=decorators)
|
|
| ast.AsyncFunctionDef(name=name, decorator_list=decorators)
|
|
):
|
|
yield from (f"{prefix}{name}" for decorator in decorators if _is_step(decorator))
|
|
case _:
|
|
pass
|
|
|
|
|
|
def _import_root(directory: Path) -> Path:
|
|
return _import_root(directory.parent) if (directory / "__init__.py").exists() else directory
|
|
|
|
|
|
@contextmanager
|
|
def _importable_from(directory: Path) -> Generator[None]:
|
|
sys.path.insert(0, str(directory))
|
|
try:
|
|
yield
|
|
finally:
|
|
sys.path.remove(str(directory))
|
|
|
|
|
|
def _imported(path: Path) -> ModuleType:
|
|
root: Final = _import_root(path.parent)
|
|
with _importable_from(root):
|
|
module: Final = importlib.import_module(".".join(path.relative_to(root).with_suffix("").parts))
|
|
assert module.__file__ is not None and Path(module.__file__).resolve() == path, (
|
|
f"{path} imports as {module.__name__}, which is {module.__file__}"
|
|
)
|
|
return module
|
|
|
|
|
|
def _helper(path: Path, module: ModuleType, qualname: str) -> StepHelper:
|
|
*scope, name = qualname.split(".")
|
|
owner: Final = reduce(lambda found, part: cast("object", getattr(found, part)), scope, cast("object", module))
|
|
wrapper: Final = cast("Callable[..., object]", getattr(owner, name))
|
|
label: Final = cast("object", inspect.getclosurevars(wrapper).nonlocals.get("label"))
|
|
assert isinstance(label, str), f"{path}::{qualname} is not wrapped by @step"
|
|
return StepHelper(path, qualname, label, owner, cast("Callable[..., object]", inspect.unwrap(wrapper)))
|
|
|
|
|
|
def _helpers_in(path: Path) -> tuple[StepHelper, ...]:
|
|
qualnames: Final = tuple(_decorated_defs(ast.parse(path.read_text()).body))
|
|
if not qualnames:
|
|
return ()
|
|
module: Final = _imported(path)
|
|
return tuple(_helper(path, module, qualname) for qualname in qualnames)
|
|
|
|
|
|
def _harness_files() -> tuple[Path, ...]:
|
|
return tuple(sorted(path.resolve() for path in E2E_DIR.rglob("*.py") if "node_modules" not in path.parts))
|
|
|
|
|
|
@cache
|
|
def step_helpers() -> tuple[StepHelper, ...]:
|
|
return tuple(chain.from_iterable(_helpers_in(path) for path in _harness_files()))
|
|
|
|
|
|
def _placeholders() -> Iterator[tuple[StepHelper, str]]:
|
|
for helper in step_helpers():
|
|
for _, field, _, _ in string.Formatter().parse(helper.label):
|
|
if field is not None:
|
|
yield helper, field
|
|
|
|
|
|
def _dotted_placeholders() -> Iterator[tuple[StepHelper, str]]:
|
|
return ((helper, field) for helper, field in _placeholders() if "." in field)
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class PlaceholderRead:
|
|
fields: tuple[FieldRead, ...]
|
|
printed: tuple[object, ...]
|
|
|
|
|
|
def _read(helper: StepHelper, field: str) -> PlaceholderRead | None:
|
|
"""The fields a placeholder reads, outermost first, across every member of a union,
|
|
and the types it ends up printing, or None if one of the fields doesn't exist."""
|
|
read: PlaceholderRead = PlaceholderRead((), _alternatives(helper.hint(field))) # rebind-ok: one hop per attribute
|
|
for attribute in field.split(".")[1:]:
|
|
found = tuple(_attribute(owner, attribute) for owner in read.printed)
|
|
hop = tuple(entry for entry in found if entry is not None)
|
|
if len(hop) != len(found):
|
|
return None
|
|
printed = tuple(chain.from_iterable(_alternatives(entry.annotation) for entry in hop))
|
|
read = PlaceholderRead((*read.fields, *hop), printed)
|
|
return read
|
|
|
|
|
|
def _fields_read(helper: StepHelper, field: str) -> tuple[FieldRead, ...] | None:
|
|
read: Final = _read(helper, field)
|
|
return None if read is None else read.fields
|
|
|
|
|
|
SECRET_NAME: Final = re.compile(
|
|
r"secret|password|api_key|access_key|private_key|credential_values|^token$|(access|auth|bearer|refresh|session)_token$"
|
|
r"|^key$|credentials$|headers$|_host$|_endpoint$|^api_base$"
|
|
)
|
|
|
|
|
|
def _models_in(annotation: object, seen: frozenset[type] = frozenset()) -> frozenset[type[BaseModel]]:
|
|
"""Every request model a value of this type can print, however deeply nested."""
|
|
if isinstance(annotation, type) and issubclass(annotation, BaseModel):
|
|
if annotation in seen:
|
|
return frozenset()
|
|
nested: Final = (
|
|
_models_in(field.annotation, seen | {annotation}) for field in annotation.model_fields.values()
|
|
)
|
|
return frozenset({annotation}).union(*nested)
|
|
args: Final = cast("tuple[object, ...]", get_args(annotation))
|
|
return frozenset[type[BaseModel]]().union(*(_models_in(arg, seen) for arg in args))
|
|
|
|
|
|
def _printed(helper: StepHelper, field: str) -> tuple[object, ...]:
|
|
read: Final = _read(helper, field)
|
|
return () if read is None else read.printed
|
|
|
|
|
|
def _printed_models() -> frozenset[type[BaseModel]]:
|
|
printed: Final = chain.from_iterable(_printed(helper, field) for helper, field in _placeholders())
|
|
return frozenset[type[BaseModel]]().union(*(_models_in(annotation) for annotation in printed))
|
|
|
|
|
|
class TestLabelTemplates:
|
|
"""A label's `{placeholders}` are filled from the call's own arguments, so the
|
|
story says what the test asked for in words, and nothing the label doesn't name
|
|
ever reaches the report."""
|
|
|
|
def test_placeholders_take_the_call_arguments_and_defaults(self) -> None:
|
|
@step('Send a request to {model} with the prompt "{content}" capped at {max_tokens} tokens')
|
|
def chat(key: str, model: str, content: str, *, max_tokens: int = 16) -> None:
|
|
return None
|
|
|
|
chat("sk-live", "claude-haiku-4-5", content="hi")
|
|
assert STEPS.taken() == ('Send a request to claude-haiku-4-5 with the prompt "hi" capped at 16 tokens',)
|
|
|
|
def test_a_request_model_reads_as_only_the_fields_the_test_set(self) -> None:
|
|
@step("Generate a virtual key with {body}")
|
|
def generate_key(body: _KeyBody) -> None:
|
|
return None
|
|
|
|
generate_key(_KeyBody(models=["a", "b"], rpm_limit=3, tpm_limit=None, api_key="sk-live"))
|
|
generate_key(_KeyBody())
|
|
assert STEPS.taken() == (
|
|
"Generate a virtual key with models: a, b and rpm limit: 3",
|
|
"Generate a virtual key with default settings",
|
|
)
|
|
|
|
def test_calls_differing_only_in_arguments_are_separate_steps(self) -> None:
|
|
@step('Send "{content}"')
|
|
def chat(content: str) -> None:
|
|
return None
|
|
|
|
for content in ("one", "one", "two"):
|
|
chat(content)
|
|
assert STEPS.taken() == ('Send "one"', 'Send "two"')
|
|
|
|
def test_a_placeholder_the_helper_does_not_take_fails_at_import(self) -> None:
|
|
def chat(model: str) -> None:
|
|
return None
|
|
|
|
with pytest.raises(TypeError, match="modle"):
|
|
_ = step("Send a request to {modle}")(chat)
|
|
|
|
def test_a_dotted_placeholder_reads_one_field_of_a_request_model(self) -> None:
|
|
@step("Add a deployment named {body.model_name} that calls {body.params.model}")
|
|
def register_model(body: _DeploymentBody) -> None:
|
|
return None
|
|
|
|
register_model(_DeploymentBody(model_name="gpt", params=_Params(model="openai/gpt-5.5")))
|
|
assert STEPS.taken() == ("Add a deployment named gpt that calls openai/gpt-5.5",)
|
|
|
|
def test_a_placeholder_that_indexes_or_calls_is_refused(self) -> None:
|
|
def chat(body: _DeploymentBody) -> None:
|
|
return None
|
|
|
|
with pytest.raises(TypeError, match=r"body\.messages\[0\]"):
|
|
_ = step("Send {body.messages[0]}")(chat)
|
|
|
|
def test_every_step_in_the_harness_is_checked(self) -> None:
|
|
written: Final = Counter({path: len(STEP_DECORATOR.findall(path.read_text())) for path in _harness_files()})
|
|
discovered: Final = Counter(helper.path for helper in step_helpers())
|
|
assert written[E2E_DIR / "proxy_client.py"] > 0
|
|
assert discovered == +written
|
|
|
|
def test_every_dotted_placeholder_in_the_harness_names_a_real_field(self) -> None:
|
|
"""A dotted placeholder is read on every live call, so one naming a field the
|
|
request model doesn't have would fail the test calling it, not the label."""
|
|
placeholders: Final = tuple(_dotted_placeholders())
|
|
assert placeholders
|
|
missing: Final = tuple(
|
|
f"{helper.where}: {field}" for helper, field in placeholders if _fields_read(helper, field) is None
|
|
)
|
|
assert missing == ()
|
|
|
|
def test_every_dotted_placeholder_in_the_harness_reads_a_field_that_is_always_set(self) -> None:
|
|
"""A field that defaults to None is usually left unset, and an unset field
|
|
prints as None, so the step would read "Save a provider credential for None".
|
|
A required field or one with a real default, like a discriminator, always reads."""
|
|
unset: Final = tuple(
|
|
f"{helper.where}: {field}"
|
|
for helper, field in _dotted_placeholders()
|
|
if not all(read.always_set for read in _fields_read(helper, field) or ())
|
|
)
|
|
assert unset == ()
|
|
|
|
def test_every_secret_field_a_label_can_print_is_hidden(self) -> None:
|
|
"""A `{body}` label prints nested models too, so a callback's credentials
|
|
inside key metadata would land in the public report unless marked `repr=False`."""
|
|
models: Final = _printed_models()
|
|
assert models
|
|
exposed: Final = sorted(
|
|
f"{model.__name__}.{name}"
|
|
for model in models
|
|
for name, field in model.model_fields.items()
|
|
if field.repr and SECRET_NAME.search(name)
|
|
)
|
|
assert exposed == []
|
|
|
|
def test_escaped_braces_stay_literal(self) -> None:
|
|
@step("GET /v1/batches/{{id}}")
|
|
def retrieve_batch(batch_id: str) -> None:
|
|
return None
|
|
|
|
retrieve_batch("batch_123")
|
|
assert STEPS.taken() == ("GET /v1/batches/{id}",)
|
|
|
|
|
|
class TestSecretMasking:
|
|
"""Steps are published with the results, so a credential the run holds is
|
|
masked wherever it shows up in a label: a nested model field nobody marked
|
|
`repr=False`, a dict value, or a prompt."""
|
|
|
|
def test_a_secret_anywhere_in_a_label_is_masked(self) -> None:
|
|
recorder: Final = StepRecorder(secrets=lambda: ("sk-live-abcdef123", "wandb-9f8e7d6c"))
|
|
recorder.record("Generate a virtual key with callback vars: wandb api key: wandb-9f8e7d6c")
|
|
recorder.record('Send "use sk-live-abcdef123 please" to claude-haiku-4-5')
|
|
assert recorder.taken() == (
|
|
f"Generate a virtual key with callback vars: wandb api key: {MASK}",
|
|
f'Send "use {MASK} please" to claude-haiku-4-5',
|
|
)
|
|
|
|
def test_a_secret_is_masked_before_the_label_is_cut(self) -> None:
|
|
secret: Final = "s3cr3t-" + "x" * 40
|
|
recorder: Final = StepRecorder(secrets=lambda: (secret,))
|
|
recorder.record("a" * 170 + " " + secret)
|
|
assert recorder.taken() == ("a" * 170 + f" {MASK}",)
|
|
|
|
def test_a_longer_secret_containing_a_shorter_one_is_masked_whole(self) -> None:
|
|
recorder: Final = StepRecorder(secrets=lambda: ("abcdefgh", "abcdefgh-ijklmnop"))
|
|
recorder.record("key abcdefgh-ijklmnop")
|
|
assert recorder.taken() == (f"key {MASK}",)
|
|
|
|
def test_only_secret_named_variables_long_enough_to_be_credentials_count(self) -> None:
|
|
environ: Final = {
|
|
"OPENAI_API_KEY": "sk-proj-0123456789",
|
|
"AWS_SECRET_ACCESS_KEY": "wJalrXUtnFEMI/K7MDENG",
|
|
"LITELLM_MASTER_KEY": "sk-test",
|
|
"GOOGLE_APPLICATION_CREDENTIALS": "/secrets/vertex.json",
|
|
"KEYCLOAK_URL": "http://localhost:8080",
|
|
"E2E_MODEL": "claude-haiku-4-5",
|
|
}
|
|
assert environment_secrets(environ) == frozenset(
|
|
{"sk-proj-0123456789", "wJalrXUtnFEMI/K7MDENG", "/secrets/vertex.json"}
|
|
)
|
|
|
|
def test_the_shared_log_masks_the_live_environment(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setenv("WANDB_API_KEY", "wandb-live-5a4b3c2d")
|
|
|
|
@step("Generate a virtual key with {body}")
|
|
def generate_key(body: _KeyBody) -> None:
|
|
return None
|
|
|
|
generate_key(_KeyBody(team_id="wandb-live-5a4b3c2d"))
|
|
assert STEPS.taken() == (f"Generate a virtual key with team id: {MASK}",)
|
|
|
|
|
|
class TestNestedSteps:
|
|
"""Harness layers call each other, so a step's helper routinely calls other
|
|
decorated helpers. Only the outermost records."""
|
|
|
|
def test_a_step_called_inside_a_step_is_not_recorded(self) -> None:
|
|
"""`ProxyClient.create_model` wraps `register_model`: one action, one
|
|
beat of the story, at the level the test called in at."""
|
|
|
|
@step("POST /key/generate")
|
|
def generate_key() -> str:
|
|
return "sk-x"
|
|
|
|
@step("generate virtual key")
|
|
def key() -> str:
|
|
return generate_key()
|
|
|
|
assert key() == "sk-x"
|
|
assert STEPS.taken() == ("generate virtual key",)
|
|
|
|
def test_the_inner_step_records_again_once_the_outer_one_returns(self) -> None:
|
|
@step("POST /key/generate")
|
|
def generate_key() -> str:
|
|
return "sk-x"
|
|
|
|
@step("generate virtual key")
|
|
def key() -> str:
|
|
return generate_key()
|
|
|
|
_ = key()
|
|
_ = generate_key()
|
|
assert STEPS.taken() == ("generate virtual key", "POST /key/generate")
|
|
|
|
def test_an_inner_step_that_raises_leaves_the_outer_label_last_and_unwinds(self) -> None:
|
|
"""The helper the test called is where it died, and the nesting flag is
|
|
released on the way out, so the next top-level call still records."""
|
|
|
|
@step("POST /team/new")
|
|
def post_team() -> None:
|
|
raise RuntimeError("/team/new answered 500")
|
|
|
|
@step("create team with a budget")
|
|
def create_team() -> None:
|
|
post_team()
|
|
|
|
@step("POST /chat/completions")
|
|
def chat() -> None:
|
|
return None
|
|
|
|
with pytest.raises(RuntimeError, match="answered 500"):
|
|
create_team()
|
|
chat()
|
|
assert STEPS.taken() == ("create team with a budget", "POST /chat/completions")
|
|
|
|
def test_a_worker_thread_a_step_fans_out_to_records_its_own_steps(self) -> None:
|
|
"""Nesting is per thread: a load helper that fans chats out to workers is
|
|
not inside a step on those workers, so their calls are still recorded."""
|
|
|
|
@step("POST /chat/completions")
|
|
def chat() -> None:
|
|
return None
|
|
|
|
@step("fire concurrent chats")
|
|
def fan_out() -> None:
|
|
worker = threading.Thread(target=chat)
|
|
worker.start()
|
|
worker.join()
|
|
|
|
fan_out()
|
|
assert STEPS.taken() == ("fire concurrent chats", "POST /chat/completions")
|
|
|
|
|
|
class TestContextManagerSteps:
|
|
"""A `@contextmanager` helper's setup and cleanup run at `__enter__` and
|
|
`__exit__`, after the decorated call has returned. Both still count as part
|
|
of its step; the `with` body is the test's own code and records as usual."""
|
|
|
|
def test_setup_and_cleanup_stay_inside_the_step_and_the_body_records(self) -> None:
|
|
@step("run a SQL statement")
|
|
def execute() -> None:
|
|
return None
|
|
|
|
@step("create a read-only database role")
|
|
@contextmanager
|
|
def restricted_user() -> Generator[str]:
|
|
execute()
|
|
try:
|
|
yield "reader"
|
|
finally:
|
|
execute()
|
|
|
|
@step("POST /chat/completions")
|
|
def chat() -> None:
|
|
return None
|
|
|
|
with restricted_user() as user:
|
|
assert user == "reader"
|
|
chat()
|
|
assert STEPS.taken() == ("create a read-only database role", "POST /chat/completions")
|
|
|
|
def test_a_test_that_dies_in_the_with_body_keeps_its_last_step_last(self) -> None:
|
|
"""The guarantee the field makes: the cleanup that runs on the way out of
|
|
the `with` must not append a step behind the one the test died on."""
|
|
|
|
@step("drop the role")
|
|
def drop_role() -> None:
|
|
return None
|
|
|
|
@step("create a read-only database role")
|
|
@contextmanager
|
|
def restricted_user() -> Generator[None]:
|
|
try:
|
|
yield
|
|
finally:
|
|
drop_role()
|
|
|
|
@step("POST /chat/completions")
|
|
def chat() -> None:
|
|
raise RuntimeError("502 from upstream")
|
|
|
|
with pytest.raises(RuntimeError, match="502 from upstream"), restricted_user():
|
|
chat()
|
|
assert STEPS.taken() == ("create a read-only database role", "POST /chat/completions")
|
|
|
|
def test_the_wrapped_context_keeps_its_exception_handling(self) -> None:
|
|
"""`__exit__` is forwarded, return value included, so a context that
|
|
suppresses an exception still does."""
|
|
|
|
@step("hold an advisory lock")
|
|
@contextmanager
|
|
def swallowing() -> Generator[None]:
|
|
try:
|
|
yield
|
|
except KeyError:
|
|
pass
|
|
|
|
with swallowing():
|
|
raise KeyError("suppressed by the context")
|
|
assert STEPS.taken() == ("hold an advisory lock",)
|
|
|
|
def test_a_bare_generator_is_refused_where_the_decorator_runs(self) -> None:
|
|
"""Its body runs only as the caller iterates, interleaved with the caller's
|
|
own steps, so no single point in the story is where it happened. Refused at
|
|
decoration, which for a harness module is import, so it lands as a
|
|
collection error rather than a story that quietly reads out of order."""
|
|
|
|
def rows() -> Generator[int]:
|
|
yield 1
|
|
|
|
with pytest.raises(TypeError, match="cannot wrap the generator function"):
|
|
_ = step("poll /spend/logs")(rows)
|