test(proxy): add harness for proxy_server.py behavior-pinning (#28827)

* test(proxy): add harness for proxy_server.py behavior-pinning

Creates tests/test_litellm/proxy/proxy_server/ with:
- conftest.py: 11 shared fixtures (app, client, mock_prisma, auth_as,
  mock_router with parametrized response builders, normalize, etc.)
- _coverage_check.py: per-PR coverage gate (line + branch) against a
  baseline, self-selects target by inspecting which placeholder files
  have been filled
- _pin_check.py: AST-based gate that verifies every pin-list item has
  >=1 happy + >=1 error test with a real assertion (no status-only)
- test_harness_smoke.py: 19 smoke tests covering every fixture +
  both scripts end-to-end
- 26 placeholder test files (one docstring each) reserved for
  follow-up PRs per the directory ownership in the Notion plan
- .coverage_baseline pinned at 0% so future PRs measure deltas
  against new-tests-only and aren't entangled with the broader
  scattered test suite

Adds a dedicated proxy-server job to test-unit-proxy-endpoints.yml
so this directory's runtime + coverage are tracked independently.

Plan: https://www.notion.so/36c43b8acdab81ee845fd5365128a2fc

* ci(proxy-endpoints): allow workflow_dispatch

Lets the workflow be triggered manually on a branch via
`gh workflow run`, which is needed for the verify-first
flow on workflow changes before opening a PR.

* test(proxy): address review feedback on proxy_server harness

- conftest.py: anchor sys.path insert to __file__ (Path(__file__).resolve().parents[4])
  instead of CWD-relative os.path.abspath("../../../../") which resolved
  to the wrong directory when pytest is launched from the repo root.
- _coverage_check.py: actually read .coverage_baseline and use it as
  the floor (line_min = max(target, baseline)). Closes the gap between
  the PR description's "delta semantics" and what the script was doing.
  With baseline=0.0 today this is a no-op; future PRs that update the
  baseline cause regressions (test deletions etc.) to trip the gate
  even if the static PR target is still met.
- _pin_check.py: drop unreachable startswith("_") guard
  (test_*.py glob never yields underscore-prefixed names) and read
  each test file once instead of twice.

.github/workflows/test-unit-proxy-endpoints.yml

@@ -7,6 +7,7 @@ on:
       - litellm_internal_staging
       - litellm_oss_branch
       - "litellm_**"
+  workflow_dispatch:
 
 permissions:
   contents: read
@@ -42,3 +43,16 @@ jobs:
       workers: 2
       reruns: 2
       artifact-name: proxy-endpoints
+
+  # Behavior-pinning tests for litellm/proxy/proxy_server.py. Owns its
+  # own job (not a path on the proxy-endpoints job above) so its budget
+  # is independent and its coverage artifact is uploaded separately.
+  # See: https://www.notion.so/36c43b8acdab81ee845fd5365128a2fc
+  proxy-server:
+    uses: ./.github/workflows/_test-unit-base.yml
+    with:
+      test-path: tests/test_litellm/proxy/proxy_server
+      workers: 4
+      reruns: 2
+      timeout-minutes: 60
+      artifact-name: proxy-server

tests/test_litellm/proxy/proxy_server/.coverage_baseline

@@ -0,0 +1 @@
+line:0.0 branch:0.0

tests/test_litellm/proxy/proxy_server/_coverage_check.py

@@ -0,0 +1,201 @@
+#!/usr/bin/env python3
+"""Coverage gate for the proxy_server.py behavior-pinning project.
+
+Reads a coverage XML report (produced by ``pytest --cov-branch
+--cov-report=xml:<path>``) and asserts that line + branch coverage on
+``litellm/proxy/proxy_server.py`` meets the per-PR target.
+
+Target selection:
+    --pr-target {1|2|3}   explicit target
+    (none)                self-selected by inspecting which placeholder
+                          test files have been filled (PR1 fills before
+                          PR2, PR2 before PR3). With nothing filled, the
+                          target is "PR0" (baseline, no minimum).
+
+Exits 0 on PASS, non-zero on FAIL.
+"""
+
+from __future__ import annotations
+
+import argparse
+import ast
+import sys
+import xml.etree.ElementTree as ET
+from pathlib import Path
+from typing import Dict, List, Tuple
+
+HERE = Path(__file__).resolve().parent
+SOURCE_FILE = "litellm/proxy/proxy_server.py"
+
+# PR target gates: (line%, branch%)
+TARGETS: Dict[str, Tuple[float, float]] = {
+    "PR0": (0.0, 0.0),
+    "PR1": (25.0, 18.0),
+    "PR2": (50.0, 38.0),
+    "PR3": (70.0, 55.0),
+}
+
+# Which placeholder files each PR is expected to fill (see Notion plan).
+PR1_FILES: List[str] = [
+    "test_lifecycle.py",
+    "test_proxy_config.py",
+    "test_spend_counters.py",
+    "test_background_health.py",
+    "test_openapi_customization.py",
+    "test_exception_handlers.py",
+    "test_streaming_helpers.py",
+]
+PR2_FILES: List[str] = [
+    "test_routes_models.py",
+    "test_routes_chat_completions.py",
+    "test_routes_completions.py",
+    "test_routes_embeddings.py",
+    "test_routes_moderations.py",
+    "test_routes_audio.py",
+    "test_routes_assistants.py",
+    "test_routes_threads.py",
+    "test_routes_utils.py",
+    "test_routes_model_info.py",
+    "test_routes_model_metrics.py",
+    "test_routes_queue.py",
+]
+PR3_FILES: List[str] = [
+    "test_routes_login_sso.py",
+    "test_routes_onboarding.py",
+    "test_routes_invitation.py",
+    "test_routes_config.py",
+    "test_routes_model_cost_map.py",
+    "test_routes_anthropic_beta.py",
+    "test_routes_misc.py",
+]
+
+
+def file_has_tests(path: Path) -> bool:
+    """A test file is considered filled if it defines at least one ``test_*``."""
+    if not path.is_file():
+        return False
+    try:
+        tree = ast.parse(path.read_text())
+    except SyntaxError:
+        return False
+    for node in ast.walk(tree):
+        if isinstance(
+            node, (ast.FunctionDef, ast.AsyncFunctionDef)
+        ) and node.name.startswith("test_"):
+            return True
+    return False
+
+
+def detect_pr_target(dir_path: Path) -> str:
+    """Pick the strictest PR whose files are fully filled in this directory."""
+    pr3_filled = all(file_has_tests(dir_path / f) for f in PR3_FILES)
+    pr2_filled = all(file_has_tests(dir_path / f) for f in PR2_FILES)
+    pr1_filled = all(file_has_tests(dir_path / f) for f in PR1_FILES)
+    if pr3_filled and pr2_filled and pr1_filled:
+        return "PR3"
+    if pr2_filled and pr1_filled:
+        return "PR2"
+    if pr1_filled:
+        return "PR1"
+    return "PR0"
+
+
+def parse_coverage_xml(xml_path: Path) -> Tuple[float, float]:
+    """Extract (line%, branch%) for proxy_server.py from a coverage XML report.
+
+    Returns (0.0, 0.0) if the file isn't found in the report.
+    """
+    if not xml_path.is_file():
+        raise FileNotFoundError(f"Coverage XML not found at {xml_path}")
+    tree = ET.parse(xml_path)
+    root = tree.getroot()
+    for class_elem in root.iter("class"):
+        filename = class_elem.get("filename", "")
+        # Coverage tools emit either a repo-relative path or just the basename
+        # depending on configuration. Match by suffix.
+        if filename.endswith("proxy/proxy_server.py") or filename.endswith(
+            "proxy_server.py"
+        ):
+            line_rate = float(class_elem.get("line-rate", "0"))
+            branch_rate = float(class_elem.get("branch-rate", "0"))
+            return line_rate * 100.0, branch_rate * 100.0
+    return 0.0, 0.0
+
+
+def parse_baseline(baseline_path: Path) -> Tuple[float, float]:
+    """Parse ``line:<float> branch:<float>`` baseline; missing file -> (0, 0)."""
+    if not baseline_path.is_file():
+        return 0.0, 0.0
+    line_pct = 0.0
+    branch_pct = 0.0
+    for token in baseline_path.read_text().split():
+        if ":" not in token:
+            continue
+        key, _, value = token.partition(":")
+        try:
+            num = float(value)
+        except ValueError:
+            continue
+        if key == "line":
+            line_pct = num
+        elif key == "branch":
+            branch_pct = num
+    return line_pct, branch_pct
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--pr-target",
+        choices=["1", "2", "3"],
+        default=None,
+        help="Explicit PR target (1, 2, or 3). If omitted, self-selected.",
+    )
+    parser.add_argument(
+        "--coverage-xml",
+        default=str(HERE.parent.parent.parent.parent / ".cov_new.xml"),
+        help="Path to coverage XML (default: <repo>/.cov_new.xml)",
+    )
+    args = parser.parse_args()
+
+    if args.pr_target:
+        target = f"PR{args.pr_target}"
+    else:
+        target = detect_pr_target(HERE)
+    target_line, target_branch = TARGETS[target]
+
+    # The effective floor is the max of the PR target and the committed
+    # baseline. The baseline is updated as each PR lands so a future
+    # regression (e.g. a test deletion) trips this gate even if the
+    # static PR target is already met.
+    baseline_line, baseline_branch = parse_baseline(HERE / ".coverage_baseline")
+    line_min = max(target_line, baseline_line)
+    branch_min = max(target_branch, baseline_branch)
+
+    xml_path = Path(args.coverage_xml)
+    try:
+        line_pct, branch_pct = parse_coverage_xml(xml_path)
+    except FileNotFoundError as exc:
+        print(f"FAIL: {exc}", file=sys.stderr)
+        return 2
+
+    line_ok = line_pct >= line_min
+    branch_ok = branch_pct >= branch_min
+    status = "PASS" if (line_ok and branch_ok) else "FAIL"
+
+    print(
+        f"target={target} baseline=(line:{baseline_line:.2f} branch:{baseline_branch:.2f})"
+    )
+    print(
+        f"line:   {line_pct:6.2f}% / {line_min:6.2f}% " f"{'OK' if line_ok else 'MISS'}"
+    )
+    print(
+        f"branch: {branch_pct:6.2f}% / {branch_min:6.2f}% "
+        f"{'OK' if branch_ok else 'MISS'}"
+    )
+    print(status)
+    return 0 if status == "PASS" else 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tests/test_litellm/proxy/proxy_server/_pin_check.py

@@ -0,0 +1,249 @@
+#!/usr/bin/env python3
+"""Pin-list gate for the proxy_server.py behavior-pinning project.
+
+For each identifier in a pin list, asserts that the test directory contains:
+  1. At least one happy-path test that references the identifier and uses
+     a real assertion (normalize(response.json()) == {...}, .model_validate,
+     or a dict-equality with >= 3 keys).
+  2. At least one error-path test (name hints at error OR asserts a 4xx/5xx
+     status OR uses pytest.raises).
+  3. No test that is "status-only" (its sole assert is on response.status_code).
+
+``test_harness_smoke.py`` is ignored (harness self-tests don't count toward
+behavior pinning).
+
+Exits 0 on PASS, non-zero on FAIL.
+"""
+
+from __future__ import annotations
+
+import argparse
+import ast
+import re
+import sys
+from dataclasses import dataclass, field
+from pathlib import Path
+from typing import Dict, List, Optional, Set, Tuple
+
+HERE = Path(__file__).resolve().parent
+
+PIN_LINE_RE = re.compile(r"^- `([^`]+)`\s*$")
+ERROR_NAME_HINTS = (
+    "error",
+    "fail",
+    "invalid",
+    "unauthorized",
+    "forbidden",
+    "missing",
+    "denied",
+    "rejected",
+    "bad",
+    "raises",
+    "exception",
+    "404",
+    "401",
+    "403",
+    "422",
+    "500",
+)
+ERROR_STATUS_CODES = frozenset({400, 401, 402, 403, 404, 405, 409, 422, 500, 502, 503})
+
+
+@dataclass
+class TestFunction:
+    name: str
+    file: Path
+    source: str
+    asserts: List[ast.Assert] = field(default_factory=list)
+    raises_calls: int = 0
+    status_code_asserts: List[int] = field(default_factory=list)
+    has_strong_assertion: bool = (
+        False  # normalize() or .model_validate() or large dict-eq
+    )
+
+
+def parse_pin_list(path: Path) -> List[str]:
+    items: List[str] = []
+    for line in path.read_text().splitlines():
+        m = PIN_LINE_RE.match(line)
+        if m:
+            items.append(m.group(1).strip())
+    return items
+
+
+def _has_strong_assertion(node: ast.AST) -> bool:
+    """True if an assert subtree contains normalize(), .model_validate(), or dict-eq with >=3 keys."""
+    for sub in ast.walk(node):
+        if isinstance(sub, ast.Call):
+            func = sub.func
+            if isinstance(func, ast.Name) and func.id == "normalize":
+                return True
+            if isinstance(func, ast.Attribute) and func.attr == "model_validate":
+                return True
+        if (
+            isinstance(sub, ast.Compare)
+            and len(sub.ops) == 1
+            and isinstance(sub.ops[0], ast.Eq)
+        ):
+            # response.json() == {<dict literal with >= 3 keys>}
+            rhs = sub.comparators[0]
+            if isinstance(rhs, ast.Dict) and len(rhs.keys) >= 3:
+                return True
+    return False
+
+
+def _extract_status_code(node: ast.Assert) -> Optional[int]:
+    """If this assert is exactly ``X.status_code == <int>``, return the int."""
+    test = node.test
+    if not isinstance(test, ast.Compare):
+        return None
+    if len(test.ops) != 1 or not isinstance(test.ops[0], ast.Eq):
+        return None
+    left = test.left
+    if not (isinstance(left, ast.Attribute) and left.attr == "status_code"):
+        return None
+    right = test.comparators[0]
+    if isinstance(right, ast.Constant) and isinstance(right.value, int):
+        return right.value
+    return None
+
+
+def collect_test_functions(test_dir: Path) -> List[TestFunction]:
+    funcs: List[TestFunction] = []
+    for path in sorted(test_dir.glob("test_*.py")):
+        # Skip the harness's own smoke tests — they don't count toward
+        # behavior pinning.
+        if path.name == "test_harness_smoke.py":
+            continue
+        source = path.read_text()
+        try:
+            tree = ast.parse(source)
+        except SyntaxError:
+            continue
+        for node in ast.walk(tree):
+            if not isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)):
+                continue
+            if not node.name.startswith("test_"):
+                continue
+            tf = TestFunction(name=node.name, file=path, source=source)
+            for sub in ast.walk(node):
+                if isinstance(sub, ast.Assert):
+                    tf.asserts.append(sub)
+                    sc = _extract_status_code(sub)
+                    if sc is not None:
+                        tf.status_code_asserts.append(sc)
+                    if _has_strong_assertion(sub):
+                        tf.has_strong_assertion = True
+                if isinstance(sub, ast.With):
+                    for item in sub.items:
+                        ctx = item.context_expr
+                        if isinstance(ctx, ast.Call) and isinstance(
+                            ctx.func, ast.Attribute
+                        ):
+                            if ctx.func.attr == "raises":
+                                tf.raises_calls += 1
+            funcs.append(tf)
+    return funcs
+
+
+def _is_status_only(tf: TestFunction) -> bool:
+    """A test that has >=1 status_code assert and ALL its asserts are status_code."""
+    return len(tf.asserts) >= 1 and len(tf.status_code_asserts) == len(tf.asserts)
+
+
+def _looks_like_error_test(tf: TestFunction) -> bool:
+    name_lower = tf.name.lower()
+    if any(hint in name_lower for hint in ERROR_NAME_HINTS):
+        return True
+    if tf.raises_calls > 0:
+        return True
+    if any(sc in ERROR_STATUS_CODES for sc in tf.status_code_asserts):
+        return True
+    return False
+
+
+def _references_pin(tf: TestFunction, pin: str) -> bool:
+    """Cheap string-contains check against the test function's source.
+
+    This is intentionally permissive — if the pin identifier (e.g.
+    ``update_cache`` or ``POST /chat/completions``) appears anywhere in
+    the test file we count it. Aliased route paths or parametrize
+    cases trigger the same reference.
+    """
+    return pin in tf.source
+
+
+def check(pin_list: List[str], funcs: List[TestFunction]) -> Tuple[bool, List[str]]:
+    failures: List[str] = []
+
+    status_only = [tf for tf in funcs if _is_status_only(tf)]
+    for tf in status_only:
+        failures.append(
+            f"status-only test (only asserts response.status_code): "
+            f"{tf.file.name}::{tf.name}"
+        )
+
+    by_pin: Dict[str, List[TestFunction]] = {pin: [] for pin in pin_list}
+    for tf in funcs:
+        for pin in pin_list:
+            if _references_pin(tf, pin):
+                by_pin[pin].append(tf)
+
+    for pin, matches in by_pin.items():
+        if not matches:
+            failures.append(f"no tests reference pin: {pin}")
+            continue
+        has_happy = any(
+            tf.has_strong_assertion and not _looks_like_error_test(tf) for tf in matches
+        )
+        has_error = any(_looks_like_error_test(tf) for tf in matches)
+        if not has_happy:
+            failures.append(
+                f"no happy-path test with strong assertion (normalize/model_validate/dict-eq>=3) "
+                f"for pin: {pin}"
+            )
+        if not has_error:
+            failures.append(f"no error-path test for pin: {pin}")
+
+    return (not failures), failures
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument(
+        "--list",
+        required=True,
+        help="Path to pin list file (markdown bullets in `- ` + backtick + symbol + backtick format)",
+    )
+    parser.add_argument(
+        "--test-dir",
+        default=str(HERE),
+        help="Test directory to scan (default: this directory)",
+    )
+    args = parser.parse_args()
+
+    pin_path = Path(args.list)
+    if not pin_path.is_file():
+        print(f"FAIL: pin list not found at {pin_path}", file=sys.stderr)
+        return 2
+
+    pin_list = parse_pin_list(pin_path)
+    if not pin_list:
+        print(f"FAIL: pin list at {pin_path} contained zero items", file=sys.stderr)
+        return 2
+
+    test_dir = Path(args.test_dir)
+    funcs = collect_test_functions(test_dir)
+
+    ok, failures = check(pin_list, funcs)
+    print(f"pins:  {len(pin_list)}")
+    print(f"tests: {len(funcs)}")
+    if failures:
+        for f in failures:
+            print(f"  - {f}")
+    print("PASS" if ok else "FAIL")
+    return 0 if ok else 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())

tests/test_litellm/proxy/proxy_server/conftest.py

@@ -0,0 +1,513 @@
+"""Shared fixtures for tests/test_litellm/proxy/proxy_server/.
+
+All fixtures and helpers used by PR1/PR2/PR3 test files live here. Do NOT
+add fixtures inside individual test files. If a fixture is missing, add it
+here and update the Notion plan.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import os
+import sys
+from pathlib import Path
+from typing import Any, AsyncIterator, Callable, Dict, Iterator, List, Optional
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+
+# Repo root, anchored to this file (not CWD) so the path is correct no
+# matter where pytest is invoked from. With the project installed via
+# uv this is defensive — `litellm` already resolves through site-packages
+# — but it lets the harness work in editable-source layouts too.
+sys.path.insert(0, str(Path(__file__).resolve().parents[4]))
+
+
+# ---------------------------------------------------------------------------
+# normalize() — used by every dict-equality assertion to scrub volatile fields
+# ---------------------------------------------------------------------------
+
+VOLATILE_KEYS = frozenset(
+    {
+        "created_at",
+        "updated_at",
+        "key",
+        "token",
+        "id",
+        "request_id",
+        "expires",
+        "expires_at",
+        "litellm_call_id",
+        "key_alias",
+        "created",
+    }
+)
+
+
+def normalize(data: Any, volatile: frozenset[str] = VOLATILE_KEYS) -> Any:
+    """Replace volatile field values with "<VOLATILE>" so dict equality works.
+
+    Recursive over dicts and lists. Pass an explicit ``volatile`` set to
+    extend or override the default.
+    """
+    if isinstance(data, dict):
+        return {
+            k: ("<VOLATILE>" if k in volatile else normalize(v, volatile))
+            for k, v in data.items()
+        }
+    if isinstance(data, list):
+        return [normalize(v, volatile) for v in data]
+    return data
+
+
+# ---------------------------------------------------------------------------
+# app + client — session-scoped so app import + TestClient setup amortize
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture(scope="session")
+def app():
+    """Return the proxy_server FastAPI app with lifespan effectively disabled.
+
+    TestClient used WITHOUT the ``with`` context manager skips the lifespan,
+    so the startup event (DB connect, Router init, OTEL setup) never fires.
+    Module import still runs once; module-level globals are harmless.
+    """
+    os.environ.setdefault("LITELLM_LOG", "ERROR")
+    from litellm.proxy.proxy_server import app as _app
+
+    return _app
+
+
+@pytest.fixture(scope="session")
+def client(app):
+    """TestClient wrapping the session app.
+
+    NOT entered as a context manager — lifespan does not fire. Tests that
+    require a real lifespan should use a function-scoped TestClient with
+    a ``with`` block locally and accept the per-test cost.
+    """
+    from fastapi.testclient import TestClient
+
+    return TestClient(app, raise_server_exceptions=False)
+
+
+# ---------------------------------------------------------------------------
+# mock_prisma — function-scoped MagicMock with the common table methods stubbed
+# ---------------------------------------------------------------------------
+
+# Tables most-touched by proxy_server.py routes. Add to this list if a
+# test discovers a missing table.
+_PRISMA_TABLES: List[str] = [
+    "litellm_verificationtoken",
+    "litellm_teamtable",
+    "litellm_usertable",
+    "litellm_endusertable",
+    "litellm_organizationtable",
+    "litellm_organizationmembership",
+    "litellm_proxymodeltable",
+    "litellm_modeltable",
+    "litellm_budgettable",
+    "litellm_spendlogs",
+    "litellm_invitationlink",
+    "litellm_credentialstable",
+    "litellm_mcpservertable",
+    "litellm_objectpermissiontable",
+    "litellm_configtable",
+    "litellm_audit_log",
+    "litellm_dailyuserspend",
+    "litellm_dailyteamspend",
+    "litellm_dailytagspend",
+    "litellm_managed_object_table",
+    "litellm_managed_vector_stores_table",
+    "litellm_promptstable",
+    "litellm_guardrailstable",
+    "litellm_managed_files",
+    "litellm_session_token_table",
+    "litellm_passthrough_endpoint_table",
+    "litellm_cron_job",
+    "litellm_passthrough_logs",
+    "litellm_health_check_table",
+    "litellm_mcpusercredentials",
+]
+
+
+def _make_table_mock() -> MagicMock:
+    table = MagicMock()
+    table.find_unique = AsyncMock(return_value=None)
+    table.find_many = AsyncMock(return_value=[])
+    table.find_first = AsyncMock(return_value=None)
+    table.create = AsyncMock()
+    table.create_many = AsyncMock()
+    table.update = AsyncMock()
+    table.update_many = AsyncMock()
+    table.upsert = AsyncMock()
+    table.delete = AsyncMock()
+    table.delete_many = AsyncMock()
+    table.count = AsyncMock(return_value=0)
+    table.group_by = AsyncMock(return_value=[])
+    table.aggregate = AsyncMock(return_value={})
+    return table
+
+
+@pytest.fixture
+def mock_prisma() -> MagicMock:
+    """MagicMock prisma_client with .db.<table> methods stubbed.
+
+    Default returns: find_unique/find_first -> None, find_many/group_by -> [],
+    count -> 0. Override in a test with::
+
+        mock_prisma.db.litellm_teamtable.find_unique.return_value = ...
+    """
+    client_mock = MagicMock()
+    client_mock.db = MagicMock()
+    client_mock.connect = AsyncMock()
+    client_mock.disconnect = AsyncMock()
+    client_mock.health_check = AsyncMock(return_value=True)
+    for table_name in _PRISMA_TABLES:
+        setattr(client_mock.db, table_name, _make_table_mock())
+    return client_mock
+
+
+# ---------------------------------------------------------------------------
+# auth_as — context manager that overrides user_api_key_auth dependency
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture
+def auth_as(app) -> Callable[..., contextlib.AbstractContextManager]:
+    """Context manager that overrides ``user_api_key_auth`` for a role.
+
+    Usage::
+
+        def test_admin_only(client, auth_as):
+            from litellm.proxy._types import LitellmUserRoles
+            with auth_as(LitellmUserRoles.PROXY_ADMIN):
+                response = client.get("/some/admin/route")
+                assert response.status_code == 200
+
+    Outside the ``with`` block the override is removed so other tests see
+    the real dependency.
+    """
+    from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
+
+    @contextlib.contextmanager
+    def _auth_as(
+        role: Any = None,
+        user_id: str = "test-user-id",
+        team_id: Optional[str] = None,
+        api_key: str = "sk-test-key",
+        **kwargs: Any,
+    ) -> Iterator[Any]:
+        from litellm.proxy._types import LitellmUserRoles, UserAPIKeyAuth
+
+        if role is None:
+            role = LitellmUserRoles.PROXY_ADMIN
+
+        fake_auth = UserAPIKeyAuth(
+            api_key=api_key,
+            user_id=user_id,
+            team_id=team_id,
+            user_role=role,
+            **kwargs,
+        )
+
+        async def _override() -> UserAPIKeyAuth:
+            return fake_auth
+
+        previous = app.dependency_overrides.get(user_api_key_auth)
+        app.dependency_overrides[user_api_key_auth] = _override
+        try:
+            yield fake_auth
+        finally:
+            if previous is None:
+                app.dependency_overrides.pop(user_api_key_auth, None)
+            else:
+                app.dependency_overrides[user_api_key_auth] = previous
+
+    return _auth_as
+
+
+# ---------------------------------------------------------------------------
+# Response builders — used by mock_router for parametrized responses
+# ---------------------------------------------------------------------------
+
+
+def make_acompletion_response(
+    model: str = "gpt-4",
+    messages: Optional[List[Dict[str, Any]]] = None,
+    stream: bool = False,
+    tools: Optional[List[Dict[str, Any]]] = None,
+    content: str = "Hello from mock",
+    **kwargs: Any,
+) -> Any:
+    """Build a deterministic chat-completion response.
+
+    Returns:
+        - An async generator when ``stream=True``
+        - A tool-call shape when ``tools`` is non-empty
+        - A plain text response otherwise
+    """
+    from litellm.types.utils import (
+        ChatCompletionMessageToolCall,
+        Choices,
+        Function,
+        Message,
+        ModelResponse,
+        Usage,
+    )
+
+    if stream:
+        return _stream_chunks(model=model, content=content)
+
+    if tools:
+        tool_name = tools[0].get("function", {}).get("name", "fake_tool")
+        message = Message(
+            role="assistant",
+            content=None,
+            tool_calls=[
+                ChatCompletionMessageToolCall(
+                    id="call_test",
+                    type="function",
+                    function=Function(name=tool_name, arguments="{}"),
+                )
+            ],
+        )
+    else:
+        message = Message(role="assistant", content=content)
+
+    return ModelResponse(
+        id="chatcmpl-test",
+        choices=[Choices(finish_reason="stop", index=0, message=message)],
+        created=0,
+        model=model,
+        object="chat.completion",
+        usage=Usage(prompt_tokens=1, completion_tokens=1, total_tokens=2),
+    )
+
+
+async def _stream_chunks(
+    model: str = "gpt-4", content: str = "Hi"
+) -> AsyncIterator[Any]:
+    from litellm.types.utils import (
+        Delta,
+        ModelResponseStream,
+        StreamingChoices,
+    )
+
+    for piece in [content, ""]:
+        yield ModelResponseStream(
+            id="chatcmpl-test",
+            choices=[
+                StreamingChoices(
+                    finish_reason=None if piece else "stop",
+                    index=0,
+                    delta=Delta(content=piece or None, role="assistant"),
+                )
+            ],
+            created=0,
+            model=model,
+            object="chat.completion.chunk",
+        )
+
+
+def make_embedding_response(
+    model: str = "text-embedding-ada-002",
+    input: Any = None,
+    dimensions: int = 8,
+    **kwargs: Any,
+) -> Any:
+    from litellm.types.utils import EmbeddingResponse
+
+    if isinstance(input, list):
+        n = len(input)
+    elif input is None:
+        n = 1
+    else:
+        n = 1
+    return EmbeddingResponse(
+        model=model,
+        data=[
+            {"embedding": [0.0] * dimensions, "index": i, "object": "embedding"}
+            for i in range(n)
+        ],
+        object="list",
+        usage={"prompt_tokens": n, "total_tokens": n},
+    )
+
+
+def make_image_response(model: str = "dall-e-3", **kwargs: Any) -> Any:
+    from litellm.types.utils import ImageResponse
+
+    return ImageResponse(
+        created=0,
+        data=[{"url": "https://example.invalid/image.png"}],
+    )
+
+
+def make_speech_response(**kwargs: Any) -> bytes:
+    """Return a fake audio blob. The route serializes bytes to a streaming response."""
+    return b"\x00" * 128
+
+
+def make_transcription_response(**kwargs: Any) -> Any:
+    from litellm.types.utils import TranscriptionResponse
+
+    return TranscriptionResponse(text="hello world")
+
+
+def make_moderation_response(**kwargs: Any) -> Dict[str, Any]:
+    return {
+        "id": "modr-test",
+        "model": "text-moderation-latest",
+        "results": [
+            {
+                "flagged": False,
+                "categories": {},
+                "category_scores": {},
+            }
+        ],
+    }
+
+
+# ---------------------------------------------------------------------------
+# mock_router — fake Router with all the *async* call surfaces stubbed
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture
+def mock_router() -> MagicMock:
+    """A MagicMock standing in for ``llm_router`` with parametrized responses."""
+
+    async def _acompletion(model: str = "gpt-4", messages=None, **kwargs):
+        return make_acompletion_response(model=model, messages=messages, **kwargs)
+
+    async def _aembedding(model: str = "text-embedding-ada-002", input=None, **kwargs):
+        return make_embedding_response(model=model, input=input, **kwargs)
+
+    async def _aimage_generation(**kwargs):
+        return make_image_response(**kwargs)
+
+    async def _aspeech(**kwargs):
+        return make_speech_response(**kwargs)
+
+    async def _atranscription(**kwargs):
+        return make_transcription_response(**kwargs)
+
+    async def _amoderation(**kwargs):
+        return make_moderation_response(**kwargs)
+
+    router = MagicMock()
+    router.acompletion = AsyncMock(side_effect=_acompletion)
+    router.aembedding = AsyncMock(side_effect=_aembedding)
+    router.aimage_generation = AsyncMock(side_effect=_aimage_generation)
+    router.aspeech = AsyncMock(side_effect=_aspeech)
+    router.atranscription = AsyncMock(side_effect=_atranscription)
+    router.amoderation = AsyncMock(side_effect=_amoderation)
+    router.model_list = [
+        {"model_name": "gpt-4", "litellm_params": {"model": "gpt-4"}},
+        {
+            "model_name": "claude-sonnet",
+            "litellm_params": {"model": "anthropic/claude-3-5-sonnet-latest"},
+        },
+        {
+            "model_name": "bedrock-claude",
+            "litellm_params": {"model": "bedrock/anthropic.claude-3-5-sonnet"},
+        },
+    ]
+    router.model_names = ["gpt-4", "claude-sonnet", "bedrock-claude"]
+    router.get_model_list = MagicMock(return_value=router.model_list)
+    return router
+
+
+# ---------------------------------------------------------------------------
+# mock_callbacks_disabled — autouse: zero out global callbacks per test
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture(autouse=True)
+def mock_callbacks_disabled(monkeypatch) -> None:
+    """Wipe ``litellm.callbacks`` and friends so tests don't leak side effects."""
+    import litellm
+
+    for attr in (
+        "callbacks",
+        "success_callback",
+        "failure_callback",
+        "_async_success_callback",
+        "_async_failure_callback",
+        "input_callback",
+        "service_callback",
+    ):
+        if hasattr(litellm, attr):
+            monkeypatch.setattr(litellm, attr, [], raising=False)
+
+
+# ---------------------------------------------------------------------------
+# Builders for DB-like objects (used by routes that load from DB)
+# ---------------------------------------------------------------------------
+
+
+def make_user(
+    user_id: str = "user-test",
+    role: Any = None,
+    teams: Optional[List[str]] = None,
+    max_budget: Optional[float] = None,
+    spend: float = 0.0,
+    **kwargs: Any,
+) -> Any:
+    from litellm.proxy._types import LiteLLM_UserTable, LitellmUserRoles
+
+    if role is None:
+        role = LitellmUserRoles.INTERNAL_USER
+
+    return LiteLLM_UserTable(
+        user_id=user_id,
+        user_role=role,
+        teams=teams or [],
+        max_budget=max_budget,
+        spend=spend,
+        **kwargs,
+    )
+
+
+def make_team(
+    team_id: str = "team-test",
+    team_alias: str = "Test Team",
+    max_budget: Optional[float] = None,
+    spend: float = 0.0,
+    members_with_roles: Optional[List[Dict[str, Any]]] = None,
+    **kwargs: Any,
+) -> Any:
+    from litellm.proxy._types import LiteLLM_TeamTable
+
+    return LiteLLM_TeamTable(
+        team_id=team_id,
+        team_alias=team_alias,
+        max_budget=max_budget,
+        spend=spend,
+        members_with_roles=members_with_roles or [],
+        **kwargs,
+    )
+
+
+def make_key(
+    token: str = "hashed-test-key",
+    key_alias: Optional[str] = None,
+    team_id: Optional[str] = None,
+    user_id: str = "user-test",
+    spend: float = 0.0,
+    max_budget: Optional[float] = None,
+    **kwargs: Any,
+) -> Any:
+    from litellm.proxy._types import LiteLLM_VerificationToken
+
+    return LiteLLM_VerificationToken(
+        token=token,
+        key_alias=key_alias,
+        team_id=team_id,
+        user_id=user_id,
+        spend=spend,
+        max_budget=max_budget,
+        **kwargs,
+    )

tests/test_litellm/proxy/proxy_server/test_background_health.py

@@ -0,0 +1 @@
+"""Placeholder. Filled by a follow-up PR per the Notion plan."""

tests/test_litellm/proxy/proxy_server/test_exception_handlers.py

@@ -0,0 +1 @@
+"""Placeholder. Filled by a follow-up PR per the Notion plan."""

tests/test_litellm/proxy/proxy_server/test_harness_smoke.py

@@ -0,0 +1,283 @@
+"""Smoke tests for the proxy_server/ test harness.
+
+Validates that fixtures + scripts work end-to-end before PR1/PR2/PR3 depend
+on them. ``_pin_check.py`` skips this file explicitly so it doesn't count
+toward behavior pinning.
+"""
+
+from __future__ import annotations
+
+import importlib.util
+import sys
+import textwrap
+from pathlib import Path
+
+import pytest
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+
+from .conftest import (  # type: ignore[import-not-found]
+    make_acompletion_response,
+    make_embedding_response,
+    normalize,
+)
+
+HERE = Path(__file__).resolve().parent
+
+
+# ---------------------------------------------------------------------------
+# Fixture smoke tests
+# ---------------------------------------------------------------------------
+
+
+def test_app_fixture_returns_fastapi_app(app):
+    assert isinstance(app, FastAPI)
+    assert app.router is not None
+
+
+def test_client_fixture_returns_testclient(client):
+    assert isinstance(client, TestClient)
+    assert hasattr(client, "post")
+    assert hasattr(client, "get")
+
+
+def test_mock_prisma_has_team_table(mock_prisma):
+    assert hasattr(mock_prisma.db, "litellm_teamtable")
+    assert callable(mock_prisma.db.litellm_teamtable.find_unique)
+    assert callable(mock_prisma.db.litellm_teamtable.find_many)
+
+
+def test_mock_prisma_has_key_table(mock_prisma):
+    assert hasattr(mock_prisma.db, "litellm_verificationtoken")
+    assert callable(mock_prisma.db.litellm_verificationtoken.find_unique)
+
+
+def test_auth_as_admin_overrides_dependency(app, auth_as):
+    from litellm.proxy._types import LitellmUserRoles
+    from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
+
+    with auth_as(LitellmUserRoles.PROXY_ADMIN):
+        assert user_api_key_auth in app.dependency_overrides
+
+
+def test_auth_as_internal_user_overrides_dependency(app, auth_as):
+    from litellm.proxy._types import LitellmUserRoles
+    from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
+
+    with auth_as(LitellmUserRoles.INTERNAL_USER) as fake_auth:
+        assert user_api_key_auth in app.dependency_overrides
+        assert fake_auth.user_role == LitellmUserRoles.INTERNAL_USER
+
+
+def test_auth_as_cleans_up_on_exit(app, auth_as):
+    from litellm.proxy._types import LitellmUserRoles
+    from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
+
+    assert user_api_key_auth not in app.dependency_overrides
+    with auth_as(LitellmUserRoles.PROXY_ADMIN):
+        pass
+    assert user_api_key_auth not in app.dependency_overrides
+
+
+def test_mock_router_acompletion_callable(mock_router):
+    from unittest.mock import AsyncMock
+
+    assert isinstance(mock_router.acompletion, AsyncMock)
+    assert isinstance(mock_router.aembedding, AsyncMock)
+    assert isinstance(mock_router.aimage_generation, AsyncMock)
+
+
+@pytest.mark.asyncio
+async def test_make_acompletion_response_stream():
+    gen = make_acompletion_response(model="gpt-4", stream=True)
+    chunks = [chunk async for chunk in gen]
+    assert len(chunks) >= 1
+    # Last chunk should have finish_reason set
+    assert chunks[-1].choices[0].finish_reason == "stop"
+
+
+def test_make_acompletion_response_tools():
+    resp = make_acompletion_response(
+        model="gpt-4",
+        tools=[{"type": "function", "function": {"name": "fake_tool"}}],
+    )
+    assert resp.choices[0].message.tool_calls is not None
+    assert resp.choices[0].message.tool_calls[0].function.name == "fake_tool"
+
+
+def test_make_embedding_response_shape():
+    resp = make_embedding_response(input=["a", "b", "c"], dimensions=4)
+    data = resp.data
+    assert len(data) == 3
+    assert len(data[0]["embedding"]) == 4
+
+
+def test_normalize_replaces_volatile_keys():
+    out = normalize({"key": "abc", "spend": 0, "nested": {"id": "x", "value": 5}})
+    assert out == {
+        "key": "<VOLATILE>",
+        "spend": 0,
+        "nested": {"id": "<VOLATILE>", "value": 5},
+    }
+
+
+def test_normalize_handles_lists():
+    out = normalize([{"key": "a"}, {"key": "b"}])
+    assert out == [{"key": "<VOLATILE>"}, {"key": "<VOLATILE>"}]
+
+
+# ---------------------------------------------------------------------------
+# Script smoke tests — _coverage_check.py
+# ---------------------------------------------------------------------------
+
+
+def _load_script(name: str):
+    spec = importlib.util.spec_from_file_location(name, HERE / f"{name}.py")
+    assert spec is not None and spec.loader is not None
+    mod = importlib.util.module_from_spec(spec)
+    # Register in sys.modules so dataclasses can resolve cls.__module__.
+    sys.modules[name] = mod
+    spec.loader.exec_module(mod)
+    return mod
+
+
+def _write_cov_xml(tmp_path: Path, line_rate: float, branch_rate: float) -> Path:
+    xml = textwrap.dedent(f"""\
+        <?xml version="1.0" ?>
+        <coverage version="7.0">
+          <packages>
+            <package name="litellm.proxy">
+              <classes>
+                <class filename="litellm/proxy/proxy_server.py"
+                       line-rate="{line_rate}" branch-rate="{branch_rate}"/>
+              </classes>
+            </package>
+          </packages>
+        </coverage>
+        """)
+    path = tmp_path / "cov.xml"
+    path.write_text(xml)
+    return path
+
+
+def test_coverage_check_pass_on_synthetic_xml(tmp_path):
+    cov_check = _load_script("_coverage_check")
+    xml = _write_cov_xml(tmp_path, line_rate=0.75, branch_rate=0.60)
+    line_pct, branch_pct = cov_check.parse_coverage_xml(xml)
+    assert line_pct == pytest.approx(75.0)
+    assert branch_pct == pytest.approx(60.0)
+
+
+def test_coverage_check_fail_on_low_coverage(tmp_path, monkeypatch, capsys):
+    cov_check = _load_script("_coverage_check")
+    xml = _write_cov_xml(tmp_path, line_rate=0.10, branch_rate=0.05)
+    monkeypatch.setattr(
+        sys,
+        "argv",
+        ["_coverage_check.py", "--pr-target", "3", "--coverage-xml", str(xml)],
+    )
+    rc = cov_check.main()
+    assert rc == 1
+    out = capsys.readouterr().out
+    assert "FAIL" in out
+
+
+def test_coverage_check_pass_on_high_coverage(tmp_path, monkeypatch, capsys):
+    cov_check = _load_script("_coverage_check")
+    xml = _write_cov_xml(tmp_path, line_rate=0.75, branch_rate=0.60)
+    monkeypatch.setattr(
+        sys,
+        "argv",
+        ["_coverage_check.py", "--pr-target", "3", "--coverage-xml", str(xml)],
+    )
+    rc = cov_check.main()
+    assert rc == 0
+    out = capsys.readouterr().out
+    assert "PASS" in out
+
+
+# ---------------------------------------------------------------------------
+# Script smoke tests — _pin_check.py
+# ---------------------------------------------------------------------------
+
+
+def _write_pin_list(tmp_path: Path, items: list) -> Path:
+    path = tmp_path / "pins.txt"
+    path.write_text("\n".join(f"- `{item}`" for item in items) + "\n")
+    return path
+
+
+def _write_test_file(tmp_path: Path, name: str, body: str) -> Path:
+    path = tmp_path / name
+    path.write_text(textwrap.dedent(body))
+    return path
+
+
+def test_pin_check_pass_on_complete_pins(tmp_path):
+    pin_check = _load_script("_pin_check")
+    _write_pin_list(tmp_path, ["update_cache"])
+    _write_test_file(
+        tmp_path,
+        "test_thing.py",
+        """\
+        def test_update_cache_happy():
+            data = update_cache(value=1)
+            assert data == {"key1": 1, "key2": 2, "key3": 3}
+
+        def test_update_cache_error():
+            import pytest
+            with pytest.raises(ValueError):
+                update_cache(value=None)
+        """,
+    )
+    pin_list = pin_check.parse_pin_list(tmp_path / "pins.txt")
+    funcs = pin_check.collect_test_functions(tmp_path)
+    ok, failures = pin_check.check(pin_list, funcs)
+    assert ok, failures
+
+
+def test_pin_check_fail_on_missing_pin(tmp_path):
+    pin_check = _load_script("_pin_check")
+    _write_pin_list(tmp_path, ["update_cache", "never_referenced_symbol"])
+    _write_test_file(
+        tmp_path,
+        "test_thing.py",
+        """\
+        def test_update_cache_happy():
+            data = update_cache(value=1)
+            assert data == {"key1": 1, "key2": 2, "key3": 3}
+
+        def test_update_cache_error():
+            import pytest
+            with pytest.raises(ValueError):
+                update_cache(value=None)
+        """,
+    )
+    pin_list = pin_check.parse_pin_list(tmp_path / "pins.txt")
+    funcs = pin_check.collect_test_functions(tmp_path)
+    ok, failures = pin_check.check(pin_list, funcs)
+    assert not ok
+    assert any("never_referenced_symbol" in f for f in failures)
+
+
+def test_pin_check_fail_on_status_only_test(tmp_path):
+    pin_check = _load_script("_pin_check")
+    _write_pin_list(tmp_path, ["some_route"])
+    _write_test_file(
+        tmp_path,
+        "test_thing.py",
+        """\
+        def test_some_route_happy():
+            response = client.get("/some_route")
+            assert response.status_code == 200
+
+        def test_some_route_error():
+            response = client.get("/some_route")
+            assert response.status_code == 404
+        """,
+    )
+    pin_list = pin_check.parse_pin_list(tmp_path / "pins.txt")
+    funcs = pin_check.collect_test_functions(tmp_path)
+    ok, failures = pin_check.check(pin_list, funcs)
+    assert not ok
+    assert any("status-only" in f for f in failures)

tests/test_litellm/proxy/proxy_server/test_lifecycle.py

@@ -0,0 +1 @@
+"""Placeholder. Filled by a follow-up PR per the Notion plan."""

tests/test_litellm/proxy/proxy_server/test_openapi_customization.py

@@ -0,0 +1 @@
+"""Placeholder. Filled by a follow-up PR per the Notion plan."""

tests/test_litellm/proxy/proxy_server/test_proxy_config.py

@@ -0,0 +1 @@
+"""Placeholder. Filled by a follow-up PR per the Notion plan."""

tests/test_litellm/proxy/proxy_server/test_routes_anthropic_beta.py

@@ -0,0 +1 @@
+"""Placeholder. Filled by a follow-up PR per the Notion plan."""

tests/test_litellm/proxy/proxy_server/test_routes_assistants.py

@@ -0,0 +1 @@
+"""Placeholder. Filled by a follow-up PR per the Notion plan."""

tests/test_litellm/proxy/proxy_server/test_routes_audio.py

@@ -0,0 +1 @@
+"""Placeholder. Filled by a follow-up PR per the Notion plan."""

tests/test_litellm/proxy/proxy_server/test_routes_chat_completions.py

@@ -0,0 +1 @@
+"""Placeholder. Filled by a follow-up PR per the Notion plan."""

tests/test_litellm/proxy/proxy_server/test_routes_completions.py

@@ -0,0 +1 @@
+"""Placeholder. Filled by a follow-up PR per the Notion plan."""

tests/test_litellm/proxy/proxy_server/test_routes_config.py

@@ -0,0 +1 @@
+"""Placeholder. Filled by a follow-up PR per the Notion plan."""

tests/test_litellm/proxy/proxy_server/test_routes_embeddings.py

@@ -0,0 +1 @@
+"""Placeholder. Filled by a follow-up PR per the Notion plan."""

tests/test_litellm/proxy/proxy_server/test_routes_invitation.py

@@ -0,0 +1 @@
+"""Placeholder. Filled by a follow-up PR per the Notion plan."""

tests/test_litellm/proxy/proxy_server/test_routes_login_sso.py

@@ -0,0 +1 @@
+"""Placeholder. Filled by a follow-up PR per the Notion plan."""

tests/test_litellm/proxy/proxy_server/test_routes_misc.py

@@ -0,0 +1 @@
+"""Placeholder. Filled by a follow-up PR per the Notion plan."""

tests/test_litellm/proxy/proxy_server/test_routes_model_cost_map.py

@@ -0,0 +1 @@
+"""Placeholder. Filled by a follow-up PR per the Notion plan."""

tests/test_litellm/proxy/proxy_server/test_routes_model_info.py

@@ -0,0 +1 @@
+"""Placeholder. Filled by a follow-up PR per the Notion plan."""

tests/test_litellm/proxy/proxy_server/test_routes_model_metrics.py

@@ -0,0 +1 @@
+"""Placeholder. Filled by a follow-up PR per the Notion plan."""

tests/test_litellm/proxy/proxy_server/test_routes_models.py

@@ -0,0 +1 @@
+"""Placeholder. Filled by a follow-up PR per the Notion plan."""

tests/test_litellm/proxy/proxy_server/test_routes_moderations.py

@@ -0,0 +1 @@
+"""Placeholder. Filled by a follow-up PR per the Notion plan."""

tests/test_litellm/proxy/proxy_server/test_routes_onboarding.py

@@ -0,0 +1 @@
+"""Placeholder. Filled by a follow-up PR per the Notion plan."""

tests/test_litellm/proxy/proxy_server/test_routes_queue.py

@@ -0,0 +1 @@
+"""Placeholder. Filled by a follow-up PR per the Notion plan."""

tests/test_litellm/proxy/proxy_server/test_routes_threads.py

@@ -0,0 +1 @@
+"""Placeholder. Filled by a follow-up PR per the Notion plan."""

tests/test_litellm/proxy/proxy_server/test_routes_utils.py

@@ -0,0 +1 @@
+"""Placeholder. Filled by a follow-up PR per the Notion plan."""

tests/test_litellm/proxy/proxy_server/test_spend_counters.py

@@ -0,0 +1 @@
+"""Placeholder. Filled by a follow-up PR per the Notion plan."""

tests/test_litellm/proxy/proxy_server/test_streaming_helpers.py

@@ -0,0 +1 @@
+"""Placeholder. Filled by a follow-up PR per the Notion plan."""