ci: match the lifecycle create verb to the resource's own routes

The gate accepted any of new, generate, add or register as a resource's
create cell, so its failure message told the reader to add
mgmt.key.new.persists when the only create route is /key/generate. It now
reads the verb from the resource's own route paths and falls back to all
four only when no route names one.

Also imports assert_never from typing_extensions so the script runs on the
3.10 floor the project declares, and drops the docstring's claim that the
baseline can only shrink, which nothing enforces.
This commit is contained in:
yuneng-jiang 2026-09-06 07:12:00 +00:00
parent ed4aa718e8
commit a5cadfa14c
No known key found for this signature in database
2 changed files with 101 additions and 21 deletions

View file

@ -1,5 +1,5 @@
"""Every management resource with a write route needs full lifecycle e2e coverage,
or an entry in management_lifecycle_baseline.json, which can only shrink.
or an entry in management_lifecycle_baseline.json.
A write route is any POST, PUT, PATCH or DELETE registration on a router in these
modules, whether it is written as a decorator or as an `add_api_route` call. A
@ -7,8 +7,9 @@ resource is the first path segment of its write routes (`/key/generate` is `key`
after any `/v1` or `/v2` version segment and behind its router's own prefix
(`/v1/mcp` + `/server/register` is `mcp_server`). Its lifecycle is covered when
`tests/e2e/coverage_registry/mgmt.yaml` holds, and a collected non-skipped e2e test
declares, `mgmt.<resource>.<create>.persists` for the create verb the resource uses
(`new`, `generate`, `add` or `register`), `mgmt.<resource>.update.preserves_unrelated_fields`,
declares, `mgmt.<resource>.<create>.persists` for the create verb the resource's own
routes use (`new`, `generate`, `add` or `register`, so `/key/generate` wants
`mgmt.key.generate.persists`), `mgmt.<resource>.update.preserves_unrelated_fields`,
`mgmt.<resource>.update.clear_persists` and `mgmt.<resource>.delete.persists`.
The gate fails on a resource with write routes that is neither covered nor baselined,
@ -16,6 +17,10 @@ and on a baseline entry that is now covered or no longer has write routes, so th
baseline never carries stale headroom. `--write-baseline` regenerates the file from the
current tree for local use and exits non-zero when it changed.
The gate itself only ever removes entries. Nothing here stops `--write-baseline` from
adding one for a brand new uncovered resource, so that growth is caught by a human
reading the one-line JSON diff in the pull request, not by this script.
uv run --no-sync python tests/code_coverage_tests/check_management_lifecycle_coverage.py
"""
@ -29,9 +34,10 @@ from collections.abc import Iterable, Mapping, Sequence
from dataclasses import dataclass
from pathlib import Path
from types import MappingProxyType
from typing import Final, Literal, TypeAlias, assert_never
from typing import Final, Literal, TypeAlias
from pydantic import BaseModel, TypeAdapter
from typing_extensions import assert_never
REPO_ROOT: Final = Path(__file__).resolve().parents[2]
MANAGEMENT_ENDPOINTS_DIR: Final = REPO_ROOT / "litellm" / "proxy" / "management_endpoints"
@ -275,15 +281,22 @@ def write_routes_by_resource(package_dir: Path) -> Mapping[str, tuple[WriteRoute
)
def _lifecycle_candidates(resource: str) -> tuple[tuple[str, ...], ...]:
def _create_verbs(routes: tuple[WriteRoute, ...]) -> tuple[str, ...]:
declared: Final = frozenset(
segment for route in routes for segment in _segments(route.path) if segment in CREATE_ROUTES
)
return tuple(verb for verb in CREATE_ROUTES if verb in declared) or CREATE_ROUTES
def _lifecycle_candidates(resource: str, routes: tuple[WriteRoute, ...]) -> tuple[tuple[str, ...], ...]:
return tuple(
(f"mgmt.{resource}.{create}.persists", *(f"mgmt.{resource}.{tail}" for tail in LIFECYCLE_TAILS))
for create in CREATE_ROUTES
for create in _create_verbs(routes)
)
def _missing_cells(resource: str, proven: frozenset[str]) -> tuple[str, ...]:
closest: Final = min(_lifecycle_candidates(resource), key=lambda cells: len(frozenset(cells) - proven))
def _missing_cells(resource: str, routes: tuple[WriteRoute, ...], proven: frozenset[str]) -> tuple[str, ...]:
closest: Final = min(_lifecycle_candidates(resource, routes), key=lambda cells: len(frozenset(cells) - proven))
return tuple(cell for cell in closest if cell not in proven)
@ -294,9 +307,11 @@ def evaluate(
baseline: frozenset[str],
) -> Verdict:
proven: Final = registry_ids & covered_markers
covered: Final = frozenset(resource for resource in routes if not _missing_cells(resource, proven))
covered: Final = frozenset(
resource for resource, declared in routes.items() if not _missing_cells(resource, declared, proven)
)
uncovered: Final = tuple(
UncoveredResource(resource, routes[resource], _missing_cells(resource, proven))
UncoveredResource(resource, routes[resource], _missing_cells(resource, routes[resource], proven))
for resource in sorted(routes)
if resource not in covered and resource not in baseline
)
@ -367,7 +382,9 @@ def write_baseline(
covered_markers: frozenset[str],
) -> int:
proven: Final = registry_ids & covered_markers
content: Final = render_baseline(resource for resource in routes if _missing_cells(resource, proven))
content: Final = render_baseline(
resource for resource, declared in routes.items() if _missing_cells(resource, declared, proven)
)
previous: Final = path.read_text(encoding="utf-8") if path.is_file() else None
path.write_text(content, encoding="utf-8")
if content == previous:

View file

@ -11,9 +11,11 @@ from check_management_lifecycle_coverage import (
evaluate,
read_baseline,
render,
resource_name,
write_baseline,
write_routes_by_resource,
)
from pydantic import ValidationError
KEY_ROUTES: Final = (
WriteRoute(method="POST", path="/key/generate", location="key_management_endpoints.py:10", resource="key"),
@ -234,31 +236,92 @@ def test_marker_outside_the_registry_does_not_count() -> None:
assert verdict.uncovered[0].missing_cells == ("mgmt.key.update.clear_persists",)
@pytest.mark.parametrize("create", ("new", "generate", "add", "register"))
def test_each_create_verb_completes_a_lifecycle(create: str) -> None:
lifecycle: Final = frozenset(
def _lifecycle_for(resource: str, create: str) -> frozenset[str]:
return frozenset(
{
f"mgmt.key.{create}.persists",
"mgmt.key.update.preserves_unrelated_fields",
"mgmt.key.update.clear_persists",
"mgmt.key.delete.persists",
f"mgmt.{resource}.{create}.persists",
f"mgmt.{resource}.update.preserves_unrelated_fields",
f"mgmt.{resource}.update.clear_persists",
f"mgmt.{resource}.delete.persists",
}
)
@pytest.mark.parametrize("create", ("new", "generate", "add", "register"))
def test_the_create_verb_a_resource_route_uses_completes_its_lifecycle(create: str) -> None:
routes: Final = MappingProxyType(
{
"thing": (
WriteRoute(method="POST", path=f"/thing/{create}", location="thing.py:1", resource="thing"),
WriteRoute(method="POST", path="/thing/delete", location="thing.py:2", resource="thing"),
)
}
)
lifecycle: Final = _lifecycle_for("thing", create)
verdict: Final = evaluate(routes, registry_ids=lifecycle, covered_markers=lifecycle, baseline=frozenset())
assert verdict == GatePassed(covered=frozenset({"thing"}), baselined=frozenset())
@pytest.mark.parametrize("create", ("new", "add", "register"))
def test_a_create_verb_the_resource_does_not_use_does_not_complete_its_lifecycle(create: str) -> None:
lifecycle: Final = _lifecycle_for("key", create)
verdict: Final = evaluate(ROUTES, registry_ids=lifecycle, covered_markers=lifecycle, baseline=frozenset())
assert verdict == GatePassed(covered=frozenset({"key"}), baselined=frozenset())
assert isinstance(verdict, GateFailed)
assert verdict.uncovered[0].missing_cells == ("mgmt.key.generate.persists",)
def test_a_resource_with_no_create_verb_in_any_route_accepts_any_of_them() -> None:
routes: Final = MappingProxyType(
{
"access_group": (
WriteRoute(method="POST", path="/v1/access_group", location="ag.py:1", resource="access_group"),
WriteRoute(method="DELETE", path="/v1/access_group/{id}", location="ag.py:2", resource="access_group"),
)
}
)
lifecycle: Final = _lifecycle_for("access_group", "new")
verdict: Final = evaluate(routes, registry_ids=lifecycle, covered_markers=lifecycle, baseline=frozenset())
assert verdict == GatePassed(covered=frozenset({"access_group"}), baselined=frozenset())
def test_a_create_cell_alone_is_not_a_lifecycle() -> None:
verdict: Final = evaluate(
ROUTES,
registry_ids=KEY_LIFECYCLE | frozenset({"mgmt.key.new.persists"}),
covered_markers=frozenset({"mgmt.key.new.persists", "mgmt.key.generate.persists"}),
registry_ids=KEY_LIFECYCLE,
covered_markers=frozenset({"mgmt.key.generate.persists"}),
baseline=frozenset(),
)
assert isinstance(verdict, GateFailed)
assert verdict.uncovered[0].missing_cells == (
"mgmt.key.update.preserves_unrelated_fields",
"mgmt.key.update.clear_persists",
"mgmt.key.delete.persists",
)
@pytest.mark.parametrize(
"content",
('{"key": true}', "[1, 2]", "not json", "[]extra"),
ids=("object", "non_strings", "not_json", "trailing_garbage"),
)
def test_a_baseline_that_is_not_a_list_of_strings_is_rejected(tmp_path: Path, content: str) -> None:
baseline_path: Final = tmp_path / "baseline.json"
baseline_path.write_text(content)
with pytest.raises(ValidationError):
read_baseline(baseline_path)
def test_a_route_with_no_nameable_segment_is_an_error() -> None:
with pytest.raises(ValueError, match=r"cannot name a resource"):
resource_name("", "/")
def test_write_baseline_lists_uncovered_resources_sorted_and_reports_change(tmp_path: Path) -> None: