From 3fb6f8740b109d638c55c19436cc9440595642f4 Mon Sep 17 00:00:00 2001 From: "devin-ai-integration[bot]" <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Thu, 24 Sep 2026 00:25:03 -0700 Subject: [PATCH] test(integration): add read-replica routing harness to the CircleCI integration suite (#42692) * test(integration): add read-replica routing harness * refactor(integration): hoist the maintenance url imports * fix(integration): keep per-test databases and the witness sequence readable under replica roles * fix(integration): opt bespoke database and pool tests out of the injected read replica * test(integration): commit recorded replica routing expectations * fix(integration): judge routing by role containment so shrinking role sets do not fail * fix(integration): run the pool-limit shutdown choreography on the superuser database url * ci(integration): add the mcp group to the replica matrix * fix(integration): judge routing by exact role sets with a named either-role allowlist * test(integration): drop containment-era routing expectations for re-recording * chore(integration): drop docstrings from the replica harness scripts * docs(integration): describe exact routing matching and the either-role list * test(integration): record exact replica routing expectations * test(integration): allow the SELECT 1 health probe on either role * test(integration): replace committed routing expectations with an on-demand base-vs-head parity run * test(integration): fix parity env scope, readme wording, and seed-deterministic serialization test * test(integration): make the sorted-role serialization test deterministic in-process * test(integration): swap all product code in parity runs and pin role gains * ci(integration): force tracked-file removal before parity checkout --------- Co-authored-by: yuneng --- .circleci/config.yml | 124 ++++- .circleci/scripts/prepare_replica_roles.py | 52 +++ .circleci/scripts/run_integration.sh | 31 +- tests/integration/README.md | 2 + tests/integration/_support/process.py | 18 +- tests/integration/_support/routing.py | 333 +++++++++++++ tests/integration/conftest.py | 3 + .../database/test_transaction_atomicity.py | 1 + ..._user_updates_wedged_coordination_redis.py | 1 + .../test_vector_store_config_ownership.py | 9 +- tests/integration/routing/either_role.json | 3 + .../routing/test_redis_recovery.py | 2 +- .../spend/test_daily_rollup_retry.py | 1 + .../integration/spend/test_shutdown_flush.py | 2 + tests/unit/integration_support/__init__.py | 0 .../unit/integration_support/test_routing.py | 438 ++++++++++++++++++ 16 files changed, 1008 insertions(+), 12 deletions(-) create mode 100644 .circleci/scripts/prepare_replica_roles.py create mode 100644 tests/integration/_support/routing.py create mode 100644 tests/integration/routing/either_role.json create mode 100644 tests/unit/integration_support/__init__.py create mode 100644 tests/unit/integration_support/test_routing.py diff --git a/.circleci/config.yml b/.circleci/config.yml index eb76244c1ab..370424dca86 100644 --- a/.circleci/config.yml +++ b/.circleci/config.yml @@ -12,6 +12,9 @@ parameters: migration_source_sha: type: string default: "" + routing_parity_base: + type: string + default: "" orbs: codecov: codecov/codecov@4.0.1 node: circleci/node@5.1.0 # Add this line to declare the node orb @@ -176,6 +179,9 @@ commands: image: type: string default: postgres:14@sha256:6a70deda415ec296f977890e11aba04a0db9f632a362e3fce45e845e3db74f26 + server_args: + type: string + default: "" steps: - run: name: Start PostgreSQL @@ -186,7 +192,7 @@ commands: -e POSTGRES_PASSWORD=postgres \ -e POSTGRES_DB=<< parameters.db_name >> \ -p 5432:5432 \ - << parameters.image >> + << parameters.image >> << parameters.server_args >> - wait_for_service: url: tcp://localhost:5432 timeout: "60" @@ -3108,6 +3114,10 @@ jobs: parameters: suite: type: string + mode: + type: enum + enum: [standard, replica] + default: standard machine: image: ubuntu-2204:2024.04.1 resource_class: large @@ -3142,18 +3152,19 @@ jobs: command: cd ui/litellm-dashboard && NEXT_TELEMETRY_DISABLED=1 npm run build - start_postgres: image: postgres:16@sha256:e17e86066e5ef83e0952a9347f5c792b7ece00972e2aa787a6986f471b3dd3d5 + server_args: "-c shared_preload_libraries=pg_stat_statements -c pg_stat_statements.track=all -c pg_stat_statements.max=20000" - start_redis - run: name: Run owned integration contracts - command: bash .circleci/scripts/run_integration.sh << parameters.suite >> + command: bash .circleci/scripts/run_integration.sh << parameters.suite >> << parameters.mode >> no_output_timeout: 15m - run: name: Stop owned database and Redis when: always command: | - mkdir -p test-results/integration-<< parameters.suite >> - docker logs postgres-db > test-results/integration-<< parameters.suite >>/postgres.log 2>&1 || true - docker logs redis-cache > test-results/integration-<< parameters.suite >>/redis.log 2>&1 || true + mkdir -p test-results/services-<< parameters.suite >>-<< parameters.mode >> + docker logs postgres-db > test-results/services-<< parameters.suite >>-<< parameters.mode >>/postgres.log 2>&1 || true + docker logs redis-cache > test-results/services-<< parameters.suite >>-<< parameters.mode >>/redis.log 2>&1 || true docker rm -f postgres-db redis-cache test -z "$(docker ps -aq --filter name=postgres-db --filter name=redis-cache)" - store_test_results: @@ -3161,6 +3172,76 @@ jobs: - store_artifacts: path: test-results + routing_parity: + parameters: + suite: + type: string + machine: + image: ubuntu-2204:2024.04.1 + resource_class: large + working_directory: ~/project + steps: + - setup_litellm_test_deps + - run: + name: Check out base product code + environment: + ROUTING_PARITY_BASE: << pipeline.parameters.routing_parity_base >> + command: | + [[ "$ROUTING_PARITY_BASE" =~ ^[0-9a-f]{40}$ ]] || exit 1 + git fetch --depth 1 origin "$ROUTING_PARITY_BASE" + git rm -r -f --quiet litellm enterprise litellm-proxy-extras + git checkout "$ROUTING_PARITY_BASE" -- litellm enterprise litellm-proxy-extras + git reset --quiet + test -f litellm/rust_bridge/_native.abi3.so + - start_postgres: + image: postgres:16@sha256:e17e86066e5ef83e0952a9347f5c792b7ece00972e2aa787a6986f471b3dd3d5 + server_args: "-c shared_preload_libraries=pg_stat_statements -c pg_stat_statements.track=all -c pg_stat_statements.max=20000" + - start_redis + - run: + name: Run base side + command: bash .circleci/scripts/run_integration.sh << parameters.suite >> parity base + no_output_timeout: 15m + - run: + name: Stop base database and Redis + when: always + command: | + mkdir -p test-results/services-<< parameters.suite >>-parity-base + docker logs postgres-db > test-results/services-<< parameters.suite >>-parity-base/postgres.log 2>&1 || true + docker logs redis-cache > test-results/services-<< parameters.suite >>-parity-base/redis.log 2>&1 || true + docker rm -f postgres-db redis-cache + test -z "$(docker ps -aq --filter name=postgres-db --filter name=redis-cache)" + - run: + name: Check out head product code + command: | + git rm -r -f --quiet litellm enterprise litellm-proxy-extras + git checkout "$CIRCLE_SHA1" -- litellm enterprise litellm-proxy-extras + git reset --quiet + test -f litellm/rust_bridge/_native.abi3.so + - start_postgres: + image: postgres:16@sha256:e17e86066e5ef83e0952a9347f5c792b7ece00972e2aa787a6986f471b3dd3d5 + server_args: "-c shared_preload_libraries=pg_stat_statements -c pg_stat_statements.track=all -c pg_stat_statements.max=20000" + - start_redis + - run: + name: Run head side + command: bash .circleci/scripts/run_integration.sh << parameters.suite >> parity head + no_output_timeout: 15m + - run: + name: Stop head database and Redis + when: always + command: | + mkdir -p test-results/services-<< parameters.suite >>-parity-head + docker logs postgres-db > test-results/services-<< parameters.suite >>-parity-head/postgres.log 2>&1 || true + docker logs redis-cache > test-results/services-<< parameters.suite >>-parity-head/redis.log 2>&1 || true + docker rm -f postgres-db redis-cache + test -z "$(docker ps -aq --filter name=postgres-db --filter name=redis-cache)" + - run: + name: Compare routing parity + command: PYTHONPATH="$PWD/tests" .venv/bin/python -m integration._support.routing check test-results/parity-<< parameters.suite >>/base test-results/parity-<< parameters.suite >>/head + - store_test_results: + path: test-results + - store_artifacts: + path: test-results + unit: machine: image: ubuntu-2204:2024.04.1 @@ -3224,8 +3305,22 @@ workflows: branches: only: main jobs: *migration_jobs + routing_parity: + when: + not: + equal: ["", << pipeline.parameters.routing_parity_base >>] + jobs: + - routing_parity: + name: routing-parity-<< matrix.suite >> + matrix: + parameters: + suite: [management, accounting, database, providers, extensions, cost, mcp] integration: - unless: << pipeline.parameters.run_migration_tests >> + unless: + or: + - << pipeline.parameters.run_migration_tests >> + - not: + equal: ["", << pipeline.parameters.routing_parity_base >>] jobs: - integration_contracts: name: integration-<< matrix.suite >> @@ -3237,8 +3332,23 @@ workflows: only: - main - /litellm_.*/ + - integration_contracts: + name: integration-<< matrix.suite >>-replica + matrix: + parameters: + suite: [management, database] + mode: [replica] + filters: + branches: + only: + - main + - /litellm_.*/ build_and_test: - unless: << pipeline.parameters.run_migration_tests >> + unless: + or: + - << pipeline.parameters.run_migration_tests >> + - not: + equal: ["", << pipeline.parameters.routing_parity_base >>] jobs: - using_litellm_on_windows: filters: &main_branches diff --git a/.circleci/scripts/prepare_replica_roles.py b/.circleci/scripts/prepare_replica_roles.py new file mode 100644 index 00000000000..fb4b7fcae97 --- /dev/null +++ b/.circleci/scripts/prepare_replica_roles.py @@ -0,0 +1,52 @@ +from __future__ import annotations + +import os +from typing import Final +from urllib.parse import urlsplit, urlunsplit + +import psycopg + +DATABASE_URL: Final = os.environ["DATABASE_URL"] + + +def postgres_url() -> str: + parsed: Final = urlsplit(DATABASE_URL) + return urlunsplit(parsed._replace(path="/postgres")) + + +def main() -> None: + with psycopg.connect(postgres_url(), autocommit=True) as admin: + admin.execute("CREATE EXTENSION IF NOT EXISTS pg_stat_statements") + admin.execute("CREATE ROLE litellm_writer LOGIN PASSWORD 'litellm-writer' NOSUPERUSER") + admin.execute("CREATE ROLE litellm_reader LOGIN PASSWORD 'litellm-reader' NOSUPERUSER NOINHERIT") + admin.execute("ALTER ROLE litellm_reader SET default_transaction_read_only = on") + admin.execute("ALTER DATABASE circle_test OWNER TO litellm_writer") + admin.execute("GRANT CONNECT ON DATABASE circle_test TO litellm_reader") + with psycopg.connect(DATABASE_URL, autocommit=True) as admin: + admin.execute("GRANT USAGE ON SCHEMA public TO litellm_reader") + admin.execute( + "ALTER DEFAULT PRIVILEGES FOR ROLE litellm_writer IN SCHEMA public GRANT SELECT ON TABLES TO litellm_reader" + ) + admin.execute("GRANT SELECT ON ALL TABLES IN SCHEMA public TO litellm_reader") + + parsed: Final = urlsplit(DATABASE_URL) + reader_url: Final = urlunsplit( + parsed._replace(netloc=f"litellm_reader:litellm-reader@{parsed.hostname}:{parsed.port}") + ) + writer_url: Final = urlunsplit( + parsed._replace(netloc=f"litellm_writer:litellm-writer@{parsed.hostname}:{parsed.port}") + ) + with psycopg.connect(reader_url, autocommit=True) as reader: + assert reader.execute("SHOW transaction_read_only").fetchone() == ("on",) + try: + reader.execute("CREATE TABLE integration_readonly_probe (id int)") + except psycopg.errors.ReadOnlySqlTransaction: + pass + else: + raise AssertionError("litellm_reader executed a write statement") + with psycopg.connect(writer_url, autocommit=True) as writer: + assert writer.execute("SELECT current_user").fetchone() == ("litellm_writer",) + + +if __name__ == "__main__": + main() diff --git a/.circleci/scripts/run_integration.sh b/.circleci/scripts/run_integration.sh index d16ac9cd124..b617a79946c 100644 --- a/.circleci/scripts/run_integration.sh +++ b/.circleci/scripts/run_integration.sh @@ -7,7 +7,15 @@ if [ "${GITHUB_ACTIONS:-}" = true ]; then fi suite="${1:?integration suite required}" -results="test-results/integration-${suite}" +mode="${2:-standard}" +side="${3:-}" +if [ "$mode" = replica ]; then + results="test-results/integration-${suite}-replica" +elif [ "$mode" = parity ]; then + results="test-results/parity-${suite}/${side:?parity side required}" +else + results="test-results/integration-${suite}" +fi mkdir -p "$results" integration_identity="$(.venv/bin/python -c 'import uuid; print(uuid.uuid4().hex)')" upstream_pid="" @@ -80,6 +88,18 @@ export INTEGRATION_ORDER_SEED="$INTEGRATION_SEED" uv run --no-sync prisma generate --schema litellm/proxy/schema.prisma > "$results/prisma-generate.log" 2>&1 +export INTEGRATION_PROXY_DATABASE_URL="" +export INTEGRATION_PROXY_READ_REPLICA_URL="" +export INTEGRATION_ROUTING="" +if [ "$mode" = replica ] || [ "$mode" = parity ]; then + .venv/bin/python .circleci/scripts/prepare_replica_roles.py > "$results/prepare-replica-roles.log" 2>&1 + export INTEGRATION_PROXY_DATABASE_URL="postgresql://litellm_writer:litellm-writer@127.0.0.1:5432/circle_test" + export INTEGRATION_PROXY_READ_REPLICA_URL="postgresql://litellm_reader:litellm-reader@127.0.0.1:5432/circle_test" +fi +if [ "$mode" = parity ]; then + export INTEGRATION_ROUTING=capture +fi + sudo iptables -N integration_only guard_created=true sudo iptables -A integration_only -o lo -j ACCEPT @@ -137,8 +157,12 @@ start_proxy() { else cost_map_env=("LITELLM_LOCAL_MODEL_COST_MAP=True") fi + local -a database_env=("DATABASE_URL=${INTEGRATION_PROXY_DATABASE_URL:-$DATABASE_URL}") + if [ -n "$INTEGRATION_PROXY_READ_REPLICA_URL" ]; then + database_env+=("DATABASE_URL_READ_REPLICA=$INTEGRATION_PROXY_READ_REPLICA_URL") + fi setsid env -i PATH="$PATH" HOME="$HOME" PYTHONPATH="$PYTHONPATH" INTEGRATION_RUN_ID="$integration_identity" \ - DATABASE_URL="$DATABASE_URL" REDIS_HOST="$REDIS_HOST" REDIS_PORT="$REDIS_PORT" \ + "${database_env[@]}" REDIS_HOST="$REDIS_HOST" REDIS_PORT="$REDIS_PORT" \ INTEGRATION_UPSTREAM_URL="$INTEGRATION_UPSTREAM_URL" \ LITELLM_MASTER_KEY="$LITELLM_MASTER_KEY" LITELLM_SALT_KEY="$LITELLM_SALT_KEY" LITELLM_UI_PATH="$LITELLM_UI_PATH" PROXY_BASE_URL="http://127.0.0.1:$port" \ LITELLM_MODE=PRODUCTION STORE_MODEL_IN_DB=True "${cost_map_env[@]}" \ @@ -195,6 +219,9 @@ env -i PATH="$PATH" HOME="$HOME" PYTHONPATH="$PYTHONPATH" \ INTEGRATION_SEED="$INTEGRATION_SEED" \ INTEGRATION_ORDER_SEED="$INTEGRATION_ORDER_SEED" \ LITELLM_LOCAL_MODEL_COST_MAP=True AWS_EC2_METADATA_DISABLED=true DO_NOT_TRACK=1 \ + INTEGRATION_PROXY_DATABASE_URL="$INTEGRATION_PROXY_DATABASE_URL" \ + INTEGRATION_PROXY_READ_REPLICA_URL="$INTEGRATION_PROXY_READ_REPLICA_URL" \ + INTEGRATION_ROUTING="$INTEGRATION_ROUTING" \ .venv/bin/python tests/integration/run.py "$suite" --results "$results" if [ "${INTEGRATION_COVERAGE:-0}" = 1 ]; then diff --git a/tests/integration/README.md b/tests/integration/README.md index f7c1305ad2e..ac9b01786b9 100644 --- a/tests/integration/README.md +++ b/tests/integration/README.md @@ -35,3 +35,5 @@ The extensions shard uses the built-in generic callback and guardrail transports The mcp shard runs the MCP gateway against SDK peers owned by each test (`_support/mcp.py`): streamable HTTP, SSE and stdio peers, an OpenAPI-spec app, and an OAuth 2.1 authorization-server double. Every peer records the requests it receives so a test can assert what reached the peer, not only what the proxy answered. The shard runs with `INTEGRATION_WORKERS` set and with `INTEGRATION_COVERAGE=1`, which starts the proxy under `coverage run --parallel-mode` limited to the MCP modules and stores `coverage.txt` plus an HTML report with the job artifacts. A test that fails because the product is wrong is skipped with `pytest.skip("BUG: ")` so the skip list in `execution.json` is the open MCP bug list Browser contracts live in `tests/e2e/ui/tests/integrationCritical` and run only through `tests/e2e/ui/integration.config.ts`. The expected browser results are listed in `expected.json` in that directory and checked by `.circleci/scripts/verify_integration_browser.py`. The CircleCI browser shard builds the checked-out dashboard, starts the owned proxy with that build, and verifies one exact browser result without retries or skips. The default Playwright selection excludes this directory. The focused project flow asserts the submitted create and clear values, fresh SQL state and actual blocked/restored serving while preserving model restrictions + +Two always-on `-replica` CircleCI jobs (management, database) run their groups in replica mode, where every proxy connects through a real `litellm_writer` role and a real read-only `litellm_reader` role against the same PostgreSQL. Nothing is captured there: the job passes when the tests pass, and a write routed to the read-only reader fails the test that issued it. A deeper check runs on demand as the `routing_parity` workflow, triggered through the CircleCI API v2 pipeline endpoint on the PR branch with `{"parameters": {"routing_parity_base": "<40-hex merge-base sha>"}}`. The workflow fans out over the seven groups, and each `routing-parity-` job runs its own group twice against the same test harness, once with `litellm/`, `enterprise/`, and `litellm-proxy-extras/` checked out from the base revision and once from the head, with a pytest plugin snapshotting `pg_stat_statements` into `routing-observed.json` per side. The `check` step then compares the two observations and writes `routing-diff.txt`: a statement seen on both sides fails when its role set changed, globally or for the same test (per-test capture is skipped under xdist), unless it is listed in `tests/integration/routing/either_role.json`, where each entry names the statement and a one-line reason it legitimately runs on whichever role asks for it, printed under `== either role ==`. Queries seen on only one side are listed, never failed, `pg_stat_statements` evictions and a role that never ran a statement are failures diff --git a/tests/integration/_support/process.py b/tests/integration/_support/process.py index 5c44beaa570..fcbaf7c8d8c 100644 --- a/tests/integration/_support/process.py +++ b/tests/integration/_support/process.py @@ -9,6 +9,7 @@ from collections.abc import Iterator, Mapping from contextlib import contextmanager from dataclasses import dataclass from pathlib import Path +from types import MappingProxyType from typing import Final import httpx @@ -16,6 +17,17 @@ import psutil from integration._support.client import Gateway +def proxy_database_environment() -> Mapping[str, str]: + writer: Final = os.environ.get("INTEGRATION_PROXY_DATABASE_URL", "") + reader: Final = os.environ.get("INTEGRATION_PROXY_READ_REPLICA_URL", "") + return MappingProxyType( + { + **({"DATABASE_URL": writer} if writer else {}), + **({"DATABASE_URL_READ_REPLICA": reader} if reader else {}), + } + ) + + def in_group(process: psutil.Process, group: int) -> bool: try: return os.getpgid(process.pid) == group @@ -83,7 +95,11 @@ def owned_proxy_process( port: Final = reserve.getsockname()[1] root: Final = Path(os.environ.get("INTEGRATION_PROXY_ROOT") or Path(__file__).resolve().parents[3]) environment: Final = { - **{name: value for name, value in os.environ.items() if name not in remove_environment}, + **{ + name: value + for name, value in {**os.environ, **proxy_database_environment()}.items() + if name not in remove_environment + }, "LITELLM_MASTER_KEY": gateway.key, "LITELLM_SALT_KEY": os.environ.get("LITELLM_SALT_KEY", "sk-integration-salt"), "STORE_MODEL_IN_DB": "True", diff --git a/tests/integration/_support/routing.py b/tests/integration/_support/routing.py new file mode 100644 index 00000000000..14f4741367a --- /dev/null +++ b/tests/integration/_support/routing.py @@ -0,0 +1,333 @@ +from __future__ import annotations + +import argparse +import itertools +import json +import os +import re +import sys +from collections.abc import Iterator, Mapping +from dataclasses import dataclass +from pathlib import Path +from types import MappingProxyType +from typing import Final +from urllib.parse import urlsplit, urlunsplit + +import psycopg +import pytest +from pydantic import TypeAdapter + +WRITER_ROLE: Final = "litellm_writer" +READER_ROLE: Final = "litellm_reader" +ROLES: Final = (READER_ROLE, WRITER_ROLE) +DATABASE_NAME: Final = "circle_test" +OBSERVED_FILE: Final = "routing-observed.json" +DIFF_FILE: Final = "routing-diff.txt" +EITHER_ROLE_FILE: Final = Path(__file__).resolve().parents[1] / "routing" / "either_role.json" + +RoleSet = frozenset[str] +RoutingMap = Mapping[str, frozenset[str]] +Snapshot = Mapping[tuple[str, str], int] + +_PLACEHOLDERS: Final = re.compile(r"\$\d+(?:\s*,\s*\$\d+)*") +_QUERIES: Final = TypeAdapter(dict[str, tuple[str, ...]]) +_OBSERVED: Final = TypeAdapter(dict[str, object]) + + +def normalize(query: str) -> str: + return _PLACEHOLDERS.sub("$n", " ".join(query.split())) + + +@dataclass(frozen=True, slots=True) +class Observation: + queries: RoutingMap + tests: Mapping[str, RoutingMap] + calls: Mapping[str, int] + dealloc: int + + +@dataclass(frozen=True, slots=True) +class Mismatch: + test: str | None + query: str + base: tuple[str, ...] + head: tuple[str, ...] + + +@dataclass(frozen=True, slots=True) +class Report: + mismatches: tuple[Mismatch, ...] + only_base: tuple[str, ...] + only_head: tuple[str, ...] + calls: Mapping[str, Mapping[str, int]] + dealloc: Mapping[str, int] + either_role: tuple[str, ...] = () + + def failures(self) -> tuple[str, ...]: + mismatch_failures: Final = tuple( + f"{mismatch.test if mismatch.test is not None else 'global'}: {mismatch.query}: " + f"base [{', '.join(mismatch.base)}] head [{', '.join(mismatch.head)}]" + for mismatch in self.mismatches + ) + side_failures: Final = tuple( + failure + for side in ("base", "head") + for failure in ( + *( + (f"{side}: pg_stat_statements evicted {self.dealloc[side]} entries (dealloc > 0)",) + if self.dealloc[side] > 0 + else () + ), + *(f"{side}: no {role} calls observed" for role in ROLES if self.calls[side].get(role, 0) == 0), + ) + ) + return (*mismatch_failures, *side_failures) + + +def _sorted_map(value: RoutingMap) -> RoutingMap: + return MappingProxyType(dict(sorted(value.items()))) + + +def compare(base: Observation, head: Observation, either_role: frozenset[str] = frozenset()) -> Report: + mismatches: Final = ( + *( + Mismatch( + None, + query, + tuple(sorted(base_roles)), + tuple(sorted(head.queries[query])), + ) + for query, base_roles in base.queries.items() + if query in head.queries and head.queries[query] != base_roles and query not in either_role + ), + *( + Mismatch( + test, + query, + tuple(sorted(base_roles)), + tuple(sorted(head.tests[test][query])), + ) + for test, queries in base.tests.items() + if test in head.tests + for query, base_roles in queries.items() + if query in head.tests[test] and head.tests[test][query] != base_roles and query not in either_role + ), + ) + varying: Final = frozenset( + query + for query in either_role + if (query in base.queries and query in head.queries and head.queries[query] != base.queries[query]) + or any( + query in base.tests[test] + and query in head.tests[test] + and head.tests[test][query] != base.tests[test][query] + for test in frozenset(base.tests) & frozenset(head.tests) + ) + ) + return Report( + mismatches, + tuple(sorted(query for query in base.queries if query not in head.queries)), + tuple(sorted(query for query in head.queries if query not in base.queries)), + MappingProxyType({"base": base.calls, "head": head.calls}), + MappingProxyType({"base": base.dealloc, "head": head.dealloc}), + tuple(sorted(varying)), + ) + + +def render(report: Report) -> str: + failures: Final = report.failures() + lines: Final = ( + "== failures ==", + *(failures or ("none",)), + "", + "== either role ==", + *(report.either_role or ("none",)), + "", + "== queries only in base ==", + *(report.only_base or ("none",)), + "", + "== queries only in head ==", + *(report.only_head or ("none",)), + "", + "== calls ==", + *( + line + for side in ("base", "head") + for line in ( + *(f"{side} {role}: {report.calls[side].get(role, 0)}" for role in ROLES), + f"{side} dealloc: {report.dealloc[side]}", + ) + ), + ) + return "\n".join(lines) + "\n" + + +def _roles(document: Mapping[str, tuple[str, ...]]) -> RoutingMap: + return _sorted_map({query: frozenset(roles) for query, roles in document.items()}) + + +def _tests(document: Mapping[str, Mapping[str, tuple[str, ...]]]) -> Mapping[str, RoutingMap]: + return MappingProxyType({node: _roles(queries) for node, queries in document.items()}) + + +def load_observation(path: Path) -> Observation: + document: Final = _OBSERVED.validate_python(json.loads(path.read_text())) + queries: Final = _QUERIES.validate_python(document.get("queries", {})) + tests: Final = TypeAdapter(dict[str, dict[str, tuple[str, ...]]]).validate_python(document.get("tests", {})) + calls: Final = TypeAdapter(dict[str, int]).validate_python(document.get("calls", {})) + dealloc: Final = TypeAdapter(int).validate_python(document.get("dealloc", 0)) + return Observation(_roles(queries), _tests(tests), MappingProxyType(calls), dealloc) + + +def load_either_role(path: Path) -> frozenset[str]: + if not path.exists(): + return frozenset() + document: Final = TypeAdapter(dict[str, str]).validate_python(json.loads(path.read_text())) + return frozenset(document) + + +def _serializable(queries: RoutingMap, tests: Mapping[str, RoutingMap]) -> dict[str, object]: + return { + "queries": {query: sorted(roles) for query, roles in queries.items()}, + "tests": {node: {query: sorted(roles) for query, roles in mapping.items()} for node, mapping in tests.items()}, + } + + +def dump_observation(observation: Observation) -> str: + document: Final = _serializable(observation.queries, observation.tests) + return ( + json.dumps( + {**document, "calls": dict(observation.calls), "dealloc": observation.dealloc}, + indent=2, + sort_keys=True, + ) + + "\n" + ) + + +def _maintenance_url() -> str: + parsed: Final = urlsplit(os.environ["DATABASE_URL"]) + return urlunsplit(parsed._replace(path="/postgres")) + + +def snapshot(connection: psycopg.Connection[object]) -> Mapping[tuple[str, str], int]: + rows: Final = connection.execute( + """ + SELECT r.rolname, s.query, s.calls + FROM pg_stat_statements s + JOIN pg_roles r ON r.oid = s.userid + WHERE s.dbid = (SELECT oid FROM pg_database WHERE datname = %s) + AND r.rolname = ANY(%s) + """, + (DATABASE_NAME, list(ROLES)), + ).fetchall() + return MappingProxyType( + { + key: sum(calls for _, _, calls in grouped) + for key, grouped in itertools.groupby( + sorted((str(role), normalize(str(query)), int(calls)) for role, query, calls in rows), + key=lambda row: (row[0], row[1]), + ) + } + ) + + +def delta(before: Snapshot, after: Snapshot) -> RoutingMap: + pairs: Final = {key: after.get(key, 0) - before.get(key, 0) for key in frozenset(before) | frozenset(after)} + queries: Final = frozenset(query for (_, query), change in pairs.items() if change > 0) + return MappingProxyType( + {query: frozenset(role for role in ROLES if pairs.get((role, query), 0) > 0) for query in sorted(queries)} + ) + + +def role_calls(before: Snapshot, after: Snapshot) -> Mapping[str, int]: + return MappingProxyType( + { + role: sum( + max(after.get((role, query), 0) - before.get((role, query), 0), 0) + for query in frozenset(q for _, q in before) | frozenset(q for _, q in after) + ) + for role in ROLES + } + ) + + +def dealloc(connection: psycopg.Connection[object]) -> int: + return int(connection.execute("SELECT dealloc FROM pg_stat_statements_info").fetchone()[0]) + + +class RoutingPlugin: + def __init__(self, config: pytest.Config) -> None: + self.config = config + self._session_start: Snapshot | None = None + self._tests: tuple[tuple[str, RoutingMap], ...] = () + + def _snapshot(self) -> Snapshot: + with psycopg.connect(_maintenance_url(), autocommit=True) as connection: + return snapshot(connection) + + def pytest_sessionstart(self, session: pytest.Session) -> None: + if hasattr(self.config, "workerinput"): + return + self._session_start = self._snapshot() + + @pytest.hookimpl(hookwrapper=True) + def pytest_runtest_protocol(self, item: pytest.Item, nextitem: pytest.Item | None) -> Iterator[None]: + if self.config.getoption("numprocesses", default=None) or hasattr(self.config, "workerinput"): + yield + return + before: Final = self._snapshot() + yield + after: Final = self._snapshot() + self._tests = (*self._tests, (item.nodeid, delta(before, after))) + + def pytest_sessionfinish(self, session: pytest.Session, exitstatus: int) -> None: + if hasattr(self.config, "workerinput"): + return + end: Final = self._snapshot() + with psycopg.connect(_maintenance_url(), autocommit=True) as connection: + evictions: Final = dealloc(connection) + start: Final = self._session_start or {} + tests: Final = MappingProxyType({node: mapping for node, mapping in self._tests}) + destination: Final = Path(os.environ["INTEGRATION_RESULTS_DIR"]) + destination.mkdir(parents=True, exist_ok=True) + (destination / OBSERVED_FILE).write_text( + dump_observation( + Observation( + delta(start, end), + tests, + role_calls(start, end), + evictions, + ) + ) + ) + + +def main(argv: tuple[str, ...] | list[str]) -> int: + parser: Final = argparse.ArgumentParser() + parser.add_argument("command", choices=("check",)) + parser.add_argument("base_dir", type=Path) + parser.add_argument("head_dir", type=Path) + parser.add_argument("--either-role", type=Path, default=EITHER_ROLE_FILE) + parser.add_argument("--diff", type=Path, default=None) + options: Final = parser.parse_args(argv) + base_path: Final = options.base_dir / OBSERVED_FILE + head_path: Final = options.head_dir / OBSERVED_FILE + for path in (base_path, head_path): + if not path.exists(): + sys.stderr.write(f"observed routing file missing: {path}\n") + if not base_path.exists() or not head_path.exists(): + return 1 + report: Final = compare( + load_observation(base_path), + load_observation(head_path), + load_either_role(options.either_role), + ) + diff: Final = render(report) + (options.diff or options.head_dir.parent / DIFF_FILE).write_text(diff) + sys.stdout.write(diff) + return 1 if report.failures() else 0 + + +if __name__ == "__main__": + raise SystemExit(main(sys.argv[1:])) diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py index 4986f5ddcc0..368b3ebee75 100644 --- a/tests/integration/conftest.py +++ b/tests/integration/conftest.py @@ -15,6 +15,7 @@ from redis import Redis from tests.integration._support.client import Gateway, eventually, gateway_from_environment from tests.integration._support.generation import LIFECYCLE_SETTINGS from tests.integration._support.manifest import OWNED_DIRECTORIES +from tests.integration._support.routing import RoutingPlugin COLLECTED: Final = pytest.StashKey[tuple[str, ...]]() REPORTS: Final = pytest.StashKey[list[pytest.TestReport]]() @@ -29,6 +30,8 @@ def pytest_configure(config: pytest.Config) -> None: config.addinivalue_line("markers", "covers(*ids): legacy contract IDs kept for existing tests, not enforced") config.stash[REPORTS] = [] config.pluginmanager.register(IntegrationReportPlugin(config)) + if os.environ.get("INTEGRATION_ROUTING"): + config.pluginmanager.register(RoutingPlugin(config)) class IntegrationReportPlugin: diff --git a/tests/integration/database/test_transaction_atomicity.py b/tests/integration/database/test_transaction_atomicity.py index c150354d9a6..030f0f3e445 100644 --- a/tests/integration/database/test_transaction_atomicity.py +++ b/tests/integration/database/test_transaction_atomicity.py @@ -43,6 +43,7 @@ def test_access_group_second_key_constraint_failure_rolls_back_all_writes(gatewa ) with psycopg.connect(os.environ["DATABASE_URL"], autocommit=True) as connection, ExitStack() as cleanup: connection.execute(sql.SQL("CREATE SEQUENCE {}").format(sql.Identifier(witness))) + connection.execute(sql.SQL("GRANT USAGE ON SEQUENCE {} TO PUBLIC").format(sql.Identifier(witness))) cleanup.callback(connection.execute, sql.SQL("DROP SEQUENCE {}").format(sql.Identifier(witness))) connection.execute( sql.SQL( diff --git a/tests/integration/management/test_user_updates_wedged_coordination_redis.py b/tests/integration/management/test_user_updates_wedged_coordination_redis.py index d8c84a70778..0715059744e 100644 --- a/tests/integration/management/test_user_updates_wedged_coordination_redis.py +++ b/tests/integration/management/test_user_updates_wedged_coordination_redis.py @@ -121,6 +121,7 @@ def test_user_budget_updates_return_promptly_while_coordination_redis_is_wedged( "REDIS_PORT": str(coordination.port), }, config=Path("tests/integration/coordination_redis_proxy_config.yaml"), + remove_environment=("DATABASE_URL_READ_REPLICA",), workers=2, ) as candidate, Redis(host=coordination.host, port=coordination.port, socket_timeout=1) as subscriber_client, diff --git a/tests/integration/management/test_vector_store_config_ownership.py b/tests/integration/management/test_vector_store_config_ownership.py index e1e7ac42472..e4ea4e324ff 100644 --- a/tests/integration/management/test_vector_store_config_ownership.py +++ b/tests/integration/management/test_vector_store_config_ownership.py @@ -318,7 +318,14 @@ def test_redis_outage_keeps_config_store_served_and_recovers( "REDIS_PORT": str(cache.port), "REDIS_CIRCUIT_BREAKER_RECOVERY_TIMEOUT": "1", } - with owned_proxy(gateway, tmp_path, overrides, config=PROXY_CONFIG, workers=2) as candidate: + with owned_proxy( + gateway, + tmp_path, + overrides, + config=PROXY_CONFIG, + workers=2, + remove_environment=("DATABASE_URL_READ_REPLICA",), + ) as candidate: db_store_id: Final = f"vs_db_{uuid.uuid4().hex}" for phase in ("before", "during", "after"): if phase == "during": diff --git a/tests/integration/routing/either_role.json b/tests/integration/routing/either_role.json new file mode 100644 index 00000000000..68824b50e11 --- /dev/null +++ b/tests/integration/routing/either_role.json @@ -0,0 +1,3 @@ +{ + "SELECT $n": "SELECT 1 health probe: the database watchdog and probe target query whichever pool reader_unavailable selects and the reconnect smoke test always uses the writer, all timer driven, so it lands under whichever test is in flight" +} diff --git a/tests/integration/routing/test_redis_recovery.py b/tests/integration/routing/test_redis_recovery.py index 81d27a190b0..9b41d44926f 100644 --- a/tests/integration/routing/test_redis_recovery.py +++ b/tests/integration/routing/test_redis_recovery.py @@ -26,7 +26,7 @@ def test_owned_redis_outage_recovers_requests_and_real_response_cache(gateway: G try: with owned_redis(tmp_path) as cache, monkeypatch.context() as environment: environment.setenv("DATABASE_URL", database_url) - with owned_proxy(gateway, tmp_path, {"DATABASE_URL": database_url, "REDIS_HOST": cache.host, "REDIS_PORT": str(cache.port), "REDIS_CIRCUIT_BREAKER_RECOVERY_TIMEOUT": "1"}) as candidate, candidate.scenario() as scenario, httpx.Client(base_url=gateway.upstream_url, timeout=5, trust_env=False) as upstream: + with owned_proxy(gateway, tmp_path, {"DATABASE_URL": database_url, "REDIS_HOST": cache.host, "REDIS_PORT": str(cache.port), "REDIS_CIRCUIT_BREAKER_RECOVERY_TIMEOUT": "1"}, remove_environment=("DATABASE_URL_READ_REPLICA",)) as candidate, candidate.scenario() as scenario, httpx.Client(base_url=gateway.upstream_url, timeout=5, trust_env=False) as upstream: model: Final = scenario.model() key: Final = scenario.key(models=[model]) for generation in ("before", "after"): diff --git a/tests/integration/spend/test_daily_rollup_retry.py b/tests/integration/spend/test_daily_rollup_retry.py index cf1b989639a..5b48d841f75 100644 --- a/tests/integration/spend/test_daily_rollup_retry.py +++ b/tests/integration/spend/test_daily_rollup_retry.py @@ -26,6 +26,7 @@ def _install_daily_user_rollup_fault(user_id: str) -> str: _execute( ( sql.SQL("CREATE SEQUENCE {}").format(sequence), + sql.SQL("GRANT USAGE ON SEQUENCE {} TO PUBLIC").format(sequence), sql.SQL( "CREATE FUNCTION {}() RETURNS trigger LANGUAGE plpgsql AS $fault$ " "BEGIN PERFORM nextval({}); " diff --git a/tests/integration/spend/test_shutdown_flush.py b/tests/integration/spend/test_shutdown_flush.py index b744c1f4b5f..5441f70d59e 100644 --- a/tests/integration/spend/test_shutdown_flush.py +++ b/tests/integration/spend/test_shutdown_flush.py @@ -196,12 +196,14 @@ def _proxy_with_one_seeded_row( gateway, tmp_path, { + "DATABASE_URL": os.environ["DATABASE_URL"], "LITELLM_LOG": "DEBUG", "GRACEFUL_SHUTDOWN_TIMEOUT": "1", "SCHEDULED_JOB_SHUTDOWN_FINISH_TIMEOUT_SECONDS": "1", "SCHEDULED_JOB_SHUTDOWN_CANCEL_TIMEOUT_SECONDS": str(cancel_timeout_seconds), }, config=_config_with_pool_limit(tmp_path, pool_limit), + remove_environment=("DATABASE_URL_READ_REPLICA",), workers=workers, ) as owned: key: Final = string_value( diff --git a/tests/unit/integration_support/__init__.py b/tests/unit/integration_support/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/tests/unit/integration_support/test_routing.py b/tests/unit/integration_support/test_routing.py new file mode 100644 index 00000000000..22a74423eb7 --- /dev/null +++ b/tests/unit/integration_support/test_routing.py @@ -0,0 +1,438 @@ +from __future__ import annotations + +import json +from collections.abc import Iterator +from pathlib import Path +from types import MappingProxyType +from typing import Final + +import pytest + +from tests.integration._support.routing import ( + DIFF_FILE, + OBSERVED_FILE, + READER_ROLE, + WRITER_ROLE, + Mismatch, + Observation, + compare, + delta, + dump_observation, + load_either_role, + load_observation, + main, + normalize, + render, + role_calls, +) + +TOKEN_QUERY: Final = 'UPDATE "LiteLLM_VerificationToken" SET token = $n WHERE token = $n' +NODE_ID: Final = "tests/integration/management/test_keys.py::test_generate" + + +def _routing(entries: dict[str, tuple[str, ...]]) -> MappingProxyType[str, frozenset[str]]: + return MappingProxyType({query: frozenset(roles) for query, roles in entries.items()}) + + +def _observation( + queries: dict[str, tuple[str, ...]], + tests: dict[str, dict[str, tuple[str, ...]]] | None = None, + calls: dict[str, int] | None = None, + dealloc: int = 0, +) -> Observation: + return Observation( + _routing(queries), + MappingProxyType({node: _routing(mapping) for node, mapping in (tests or {}).items()}), + MappingProxyType(calls if calls is not None else {"litellm_reader": 3, "litellm_writer": 7}), + dealloc, + ) + + +@pytest.mark.parametrize( + ("raw", "expected"), + [ + ("SELECT a\n FROM t", "SELECT a FROM t"), + ("SELECT * FROM t WHERE id IN ($1, $2, $3)", "SELECT * FROM t WHERE id IN ($n)"), + ("SELECT * FROM t WHERE id IN ($1,$2)", "SELECT * FROM t WHERE id IN ($n)"), + ("SELECT * FROM t WHERE id IN ($4)", "SELECT * FROM t WHERE id IN ($n)"), + ( + "INSERT INTO t VALUES ($1, $2) ON CONFLICT ($3, $4, $5) DO NOTHING", + "INSERT INTO t VALUES ($n) ON CONFLICT ($n) DO NOTHING", + ), + ], +) +def test_normalize_collapses_whitespace_and_placeholders(raw: str, expected: str) -> None: + assert normalize(raw) == expected + + +def test_compare_reports_global_role_mismatch() -> None: + base: Final = _observation({TOKEN_QUERY: ("litellm_reader",), "SELECT 1": ("litellm_writer",)}) + head: Final = _observation({TOKEN_QUERY: ("litellm_writer",), "SELECT 1": ("litellm_writer",)}) + report: Final = compare(base, head) + assert report.mismatches == (Mismatch(None, TOKEN_QUERY, ("litellm_reader",), ("litellm_writer",)),) + assert report.failures() == (f"global: {TOKEN_QUERY}: base [litellm_reader] head [litellm_writer]",) + + +def test_compare_reports_global_shrink_mismatch() -> None: + base: Final = _observation({TOKEN_QUERY: ("litellm_reader", "litellm_writer")}) + head: Final = _observation({TOKEN_QUERY: ("litellm_writer",)}) + report: Final = compare(base, head) + assert report.mismatches == ( + Mismatch(None, TOKEN_QUERY, ("litellm_reader", "litellm_writer"), ("litellm_writer",)), + ) + assert report.failures() == (f"global: {TOKEN_QUERY}: base [litellm_reader, litellm_writer] head [litellm_writer]",) + + +def test_compare_reports_per_test_mismatch_with_nodeid() -> None: + base: Final = _observation( + {TOKEN_QUERY: ("litellm_reader",)}, + {NODE_ID: {TOKEN_QUERY: ("litellm_reader",)}}, + ) + head: Final = _observation( + {TOKEN_QUERY: ("litellm_reader",)}, + {NODE_ID: {TOKEN_QUERY: ("litellm_writer",)}}, + ) + report: Final = compare(base, head) + assert report.mismatches == (Mismatch(NODE_ID, TOKEN_QUERY, ("litellm_reader",), ("litellm_writer",)),) + assert report.failures() == (f"{NODE_ID}: {TOKEN_QUERY}: base [litellm_reader] head [litellm_writer]",) + + +def test_compare_per_test_mismatch_ignores_global_observation() -> None: + base: Final = _observation( + {TOKEN_QUERY: ("litellm_reader", "litellm_writer")}, + {NODE_ID: {TOKEN_QUERY: ("litellm_reader",)}}, + ) + head: Final = _observation( + {TOKEN_QUERY: ("litellm_reader", "litellm_writer")}, + {NODE_ID: {TOKEN_QUERY: ("litellm_writer",)}}, + ) + report: Final = compare(base, head) + assert report.mismatches == (Mismatch(NODE_ID, TOKEN_QUERY, ("litellm_reader",), ("litellm_writer",)),) + assert report.failures() == (f"{NODE_ID}: {TOKEN_QUERY}: base [litellm_reader] head [litellm_writer]",) + + +def test_compare_reports_per_test_shrink_mismatch() -> None: + base: Final = _observation( + {TOKEN_QUERY: ("litellm_reader", "litellm_writer")}, + {NODE_ID: {TOKEN_QUERY: ("litellm_reader", "litellm_writer")}}, + ) + head: Final = _observation( + {TOKEN_QUERY: ("litellm_reader", "litellm_writer")}, + {NODE_ID: {TOKEN_QUERY: ("litellm_reader",)}}, + ) + report: Final = compare(base, head) + assert report.mismatches == ( + Mismatch(NODE_ID, TOKEN_QUERY, ("litellm_reader", "litellm_writer"), ("litellm_reader",)), + ) + assert report.failures() == ( + f"{NODE_ID}: {TOKEN_QUERY}: base [litellm_reader, litellm_writer] head [litellm_reader]", + ) + + +def test_compare_reports_global_gain_mismatch() -> None: + base: Final = _observation({TOKEN_QUERY: ("litellm_reader",)}) + head: Final = _observation({TOKEN_QUERY: ("litellm_reader", "litellm_writer")}) + report: Final = compare(base, head) + assert report.mismatches == ( + Mismatch(None, TOKEN_QUERY, ("litellm_reader",), ("litellm_reader", "litellm_writer")), + ) + assert report.failures() == (f"global: {TOKEN_QUERY}: base [litellm_reader] head [litellm_reader, litellm_writer]",) + + +def test_compare_reports_per_test_gain_mismatch() -> None: + base: Final = _observation( + {TOKEN_QUERY: ("litellm_reader",)}, + {NODE_ID: {TOKEN_QUERY: ("litellm_reader",)}}, + ) + head: Final = _observation( + {TOKEN_QUERY: ("litellm_reader",)}, + {NODE_ID: {TOKEN_QUERY: ("litellm_reader", "litellm_writer")}}, + ) + report: Final = compare(base, head) + assert report.mismatches == ( + Mismatch(NODE_ID, TOKEN_QUERY, ("litellm_reader",), ("litellm_reader", "litellm_writer")), + ) + assert report.failures() == ( + f"{NODE_ID}: {TOKEN_QUERY}: base [litellm_reader] head [litellm_reader, litellm_writer]", + ) + + +def test_compare_either_role_suppresses_and_reports_variance() -> None: + base: Final = _observation( + {TOKEN_QUERY: ("litellm_reader", "litellm_writer"), "SELECT quiet": ("litellm_reader",)}, + {NODE_ID: {TOKEN_QUERY: ("litellm_reader",)}}, + ) + head: Final = _observation( + {TOKEN_QUERY: ("litellm_writer",), "SELECT quiet": ("litellm_reader",)}, + {NODE_ID: {TOKEN_QUERY: ("litellm_writer",)}}, + ) + report: Final = compare(base, head, either_role=frozenset({TOKEN_QUERY, "SELECT quiet"})) + assert report.mismatches == () + assert report.failures() == () + assert report.either_role == (TOKEN_QUERY,) + assert "== either role ==\n" + TOKEN_QUERY + "\n" in render(report) + + +def test_compare_either_role_matches_exact_keys_only() -> None: + base: Final = _observation( + { + "SELECT $n": ("litellm_reader",), + "SELECT $n FROM x": ("litellm_reader",), + "SELECT $n FROM x WHERE y = $n": ("litellm_reader",), + } + ) + head: Final = _observation( + { + "SELECT $n": ("litellm_writer",), + "SELECT $n FROM x": ("litellm_writer",), + "SELECT $n FROM x WHERE y = $n": ("litellm_writer",), + } + ) + report: Final = compare(base, head, either_role=frozenset({"SELECT $n FROM x"})) + assert frozenset(mismatch.query for mismatch in report.mismatches) == frozenset( + {"SELECT $n", "SELECT $n FROM x WHERE y = $n"} + ) + other: Final = compare(base, head, either_role=frozenset({"SELECT $n"})) + assert frozenset(mismatch.query for mismatch in other.mismatches) == frozenset( + {"SELECT $n FROM x", "SELECT $n FROM x WHERE y = $n"} + ) + + +def test_compare_one_sided_queries_are_listed_not_failed() -> None: + base: Final = _observation({"SELECT a": ("litellm_reader",), "SELECT gone": ("litellm_writer",)}) + head: Final = _observation({"SELECT a": ("litellm_reader",), "SELECT new": ("litellm_writer",)}) + report: Final = compare(base, head) + assert report.only_base == ("SELECT gone",) + assert report.only_head == ("SELECT new",) + assert report.mismatches == () + assert report.failures() == () + + +def test_failures_flags_dealloc_evictions_on_base() -> None: + report: Final = compare(_observation({}, dealloc=1), _observation({})) + assert report.failures() == ("base: pg_stat_statements evicted 1 entries (dealloc > 0)",) + + +def test_failures_flags_dealloc_evictions_on_head() -> None: + report: Final = compare(_observation({}), _observation({}, dealloc=1)) + assert report.failures() == ("head: pg_stat_statements evicted 1 entries (dealloc > 0)",) + + +def test_failures_flags_silent_reader_on_base() -> None: + report: Final = compare( + _observation({}, calls={"litellm_reader": 0, "litellm_writer": 5}), + _observation({}), + ) + assert report.failures() == ("base: no litellm_reader calls observed",) + + +def test_failures_flags_silent_reader_on_head() -> None: + report: Final = compare( + _observation({}), + _observation({}, calls={"litellm_reader": 0, "litellm_writer": 5}), + ) + assert report.failures() == ("head: no litellm_reader calls observed",) + + +def test_failures_flags_silent_writer() -> None: + report: Final = compare( + _observation({}, calls={"litellm_reader": 5, "litellm_writer": 0}), + _observation({}), + ) + assert report.failures() == ("base: no litellm_writer calls observed",) + + +def test_failures_counts_missing_role_as_silent() -> None: + report: Final = compare(_observation({}), _observation({}, calls={"litellm_writer": 5})) + assert report.failures() == ("head: no litellm_reader calls observed",) + + +def test_compare_skips_per_test_mismatches_for_xdist_shape() -> None: + base: Final = _observation( + {TOKEN_QUERY: ("litellm_reader",)}, + {NODE_ID: {TOKEN_QUERY: ("litellm_reader",)}}, + ) + head: Final = _observation({TOKEN_QUERY: ("litellm_reader",)}) + assert head.tests == {} + report: Final = compare(base, head) + assert report.mismatches == () + assert report.failures() == () + + +class _WriterFirst(frozenset[str]): + def __iter__(self) -> Iterator[str]: + return iter((WRITER_ROLE, READER_ROLE)) + + +def test_dump_observation_sorts_role_lists_and_round_trips(tmp_path: Path) -> None: + queries: Final = [f"SELECT {index}" for index in range(4)] + observation: Final = Observation( + MappingProxyType({query: _WriterFirst({WRITER_ROLE, READER_ROLE}) for query in queries}), + MappingProxyType( + {NODE_ID: MappingProxyType({query: _WriterFirst({WRITER_ROLE, READER_ROLE}) for query in queries})} + ), + MappingProxyType({READER_ROLE: 1, WRITER_ROLE: 2}), + 0, + ) + expected: Final = ( + json.dumps( + { + "queries": {query: ["litellm_reader", "litellm_writer"] for query in queries}, + "tests": {NODE_ID: {query: ["litellm_reader", "litellm_writer"] for query in queries}}, + "calls": {"litellm_reader": 1, "litellm_writer": 2}, + "dealloc": 0, + }, + sort_keys=True, + indent=2, + ) + + "\n" + ) + dumped: Final = dump_observation(observation) + assert dumped == expected + path: Final = tmp_path / OBSERVED_FILE + path.write_text(dumped) + loaded: Final = load_observation(path) + assert loaded.queries == _routing({query: (WRITER_ROLE, READER_ROLE) for query in queries}) + assert loaded.tests == {NODE_ID: loaded.queries} + + +def _write_observed(results: Path, observation: Observation) -> None: + results.mkdir(parents=True, exist_ok=True) + (results / OBSERVED_FILE).write_text(dump_observation(observation)) + + +def test_main_check_returns_zero_for_matching_routes(tmp_path: Path) -> None: + base_dir: Final = tmp_path / "base" + head_dir: Final = tmp_path / "head" + observation: Final = _observation({TOKEN_QUERY: ("litellm_reader",)}) + _write_observed(base_dir, observation) + _write_observed(head_dir, observation) + assert main(["check", str(base_dir), str(head_dir)]) == 0 + diff: Final = (tmp_path / DIFF_FILE).read_text() + assert "== failures ==\nnone\n" in diff + + +def test_main_check_returns_one_and_writes_exact_diff(tmp_path: Path) -> None: + base_dir: Final = tmp_path / "parity" / "base" + head_dir: Final = tmp_path / "parity" / "head" + _write_observed( + base_dir, + _observation({TOKEN_QUERY: ("litellm_reader",), "SELECT absent": ("litellm_writer",)}), + ) + _write_observed( + head_dir, + _observation( + {TOKEN_QUERY: ("litellm_writer",)}, + calls={"litellm_reader": 0, "litellm_writer": 5}, + dealloc=2, + ), + ) + assert main(["check", str(base_dir), str(head_dir)]) == 1 + assert (head_dir.parent / DIFF_FILE).read_text() == ( + "== failures ==\n" + f"global: {TOKEN_QUERY}: base [litellm_reader] head [litellm_writer]\n" + "head: pg_stat_statements evicted 2 entries (dealloc > 0)\n" + "head: no litellm_reader calls observed\n" + "\n" + "== either role ==\n" + "none\n" + "\n" + "== queries only in base ==\n" + "SELECT absent\n" + "\n" + "== queries only in head ==\n" + "none\n" + "\n" + "== calls ==\n" + "base litellm_reader: 3\n" + "base litellm_writer: 7\n" + "base dealloc: 0\n" + "head litellm_reader: 0\n" + "head litellm_writer: 5\n" + "head dealloc: 2\n" + ) + + +def test_main_check_missing_observed_returns_one(tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None: + base_dir: Final = tmp_path / "base" + head_dir: Final = tmp_path / "head" + _write_observed(base_dir, _observation({})) + head_dir.mkdir() + assert main(["check", str(base_dir), str(head_dir)]) == 1 + assert "observed routing file missing" in capsys.readouterr().err + + +def test_main_check_either_role_suppresses_shrink(tmp_path: Path) -> None: + base_dir: Final = tmp_path / "base" + head_dir: Final = tmp_path / "head" + _write_observed(base_dir, _observation({TOKEN_QUERY: ("litellm_reader", "litellm_writer")})) + _write_observed(head_dir, _observation({TOKEN_QUERY: ("litellm_writer",)})) + argv: Final = ["check", str(base_dir), str(head_dir)] + allowlist: Final = tmp_path / "either.json" + allowlist.write_text(json.dumps({TOKEN_QUERY: "timer probe may use either pool"})) + assert main([*argv, "--either-role", str(allowlist)]) == 0 + assert "== either role ==\n" + TOKEN_QUERY + "\n" in (tmp_path / DIFF_FILE).read_text() + assert main(argv) == 1 + + +def test_delta_maps_positive_increases_per_role() -> None: + before: Final = MappingProxyType( + { + ("litellm_reader", "SELECT both"): 1, + ("litellm_writer", "SELECT both"): 2, + ("litellm_reader", "SELECT reader"): 3, + ("litellm_writer", "SELECT gone"): 4, + ("litellm_reader", "SELECT same"): 5, + } + ) + after: Final = MappingProxyType( + { + ("litellm_reader", "SELECT both"): 2, + ("litellm_writer", "SELECT both"): 5, + ("litellm_reader", "SELECT reader"): 6, + ("litellm_reader", "SELECT same"): 5, + ("litellm_writer", "SELECT writer"): 7, + } + ) + assert delta(before, after) == { + "SELECT both": frozenset({"litellm_reader", "litellm_writer"}), + "SELECT reader": frozenset({"litellm_reader"}), + "SELECT writer": frozenset({"litellm_writer"}), + } + + +def test_role_calls_sums_positive_increases_per_role() -> None: + before: Final = MappingProxyType( + { + ("litellm_reader", "SELECT a"): 10, + ("litellm_reader", "SELECT b"): 4, + ("litellm_writer", "SELECT a"): 1, + } + ) + after: Final = MappingProxyType( + { + ("litellm_reader", "SELECT a"): 11, + ("litellm_reader", "SELECT b"): 2, + ("litellm_writer", "SELECT a"): 1, + ("litellm_writer", "SELECT c"): 6, + } + ) + assert role_calls(before, after) == {"litellm_reader": 1, "litellm_writer": 6} + + +def test_load_either_role_missing_path_returns_empty(tmp_path: Path) -> None: + assert load_either_role(tmp_path / "absent.json") == frozenset() + + +def test_load_either_role_reads_query_keys(tmp_path: Path) -> None: + path: Final = tmp_path / "either.json" + path.write_text(json.dumps({"SELECT $n": "probe", "SELECT now()": "clock"})) + assert load_either_role(path) == frozenset({"SELECT $n", "SELECT now()"}) + + +def test_load_observation_reads_calls_and_dealloc(tmp_path: Path) -> None: + observation: Final = _observation({TOKEN_QUERY: ("litellm_reader",)}, dealloc=0) + path: Final = tmp_path / OBSERVED_FILE + path.write_text(dump_observation(observation)) + loaded: Final = load_observation(path) + assert loaded == observation