diff --git a/.github/scripts/compare_versions_sources.py b/.github/scripts/compare_versions_sources.py new file mode 100644 index 000000000..ad6563dc5 --- /dev/null +++ b/.github/scripts/compare_versions_sources.py @@ -0,0 +1,382 @@ +"""Compare the old gateway's ``/versions`` answers with the v2 manifest's. + +The Simulation Entrypoint forwards ``GET /versions`` and +``GET /versions/{kind}`` to the old gateway, which answers from v1's routing +state. It serves the answers the v2 version manifest gives at +``GET /internal/v2/versions`` and ``GET /internal/v2/versions/{kind}``. A +staging deploy runs this script against its candidate revision and records +every difference between the two in the job summary, before ``/versions`` +stops forwarding. + +The report is for information only. The script exits 0 whatever it finds, +including when a route cannot be read, so it never fails a deploy. It exits 2 +only when it is run without a base URL. + +Application names are not compared directly: the old gateway names each +version's v1 application (``policyengine-simulation-py…``) and the manifest +its v2 application (``policyengine-simulation-v2-py…``). Each route is +compared instead by the PolicyEngine version whose ``policyengine`` route, in +the same answer, names the same application. + +Usage: + compare_versions_sources.py BASE_URL + +Environment: + GITHUB_STEP_SUMMARY optional; a file the report is appended to (set by the + GitHub Actions runner) +""" + +from __future__ import annotations + +import json +import os +import sys +from dataclasses import dataclass +from typing import TYPE_CHECKING +from urllib.error import HTTPError +from urllib.request import Request, urlopen + +if TYPE_CHECKING: + from collections.abc import Callable + +KINDS = ("policyengine", "us", "uk") +# A kind neither source supports; both must reject it the same way. +UNKNOWN_KIND = "unknown-kind" +LATEST_KEY = "latest" +HTTP_OK = 200 +EXPECTED_ARGUMENT_COUNT = 2 +USAGE_ERROR = 2 +REQUEST_TIMEOUT_SECONDS = 30 +MAX_CELL_LENGTH = 160 + + +class _Absent: + """Marks a key that one answer has and the other does not.""" + + def __repr__(self) -> str: + return "(absent)" + + +ABSENT = _Absent() + + +@dataclass(frozen=True) +class Answer: + """One route's HTTP status and JSON body, or why there is none.""" + + status: int | None + body: object = None + error: str | None = None + + def describe(self) -> str: + """Return the status, or the error that prevented one.""" + if self.status is None: + return f"no answer ({self.error})" + return f"HTTP {self.status}" + + +@dataclass(frozen=True) +class Difference: + """One place where the old gateway's answer and the manifest's differ.""" + + where: str + key: str + gateway: object + manifest: object + + +type Fetch = Callable[[str], Answer] + + +def fetch(url: str) -> Answer: + """GET ``url`` and return its status and parsed JSON body.""" + request = Request(url, headers={"Accept": "application/json"}) # noqa: S310 # the URL is the deploy's own candidate revision, from its argument + try: + with urlopen(request, timeout=REQUEST_TIMEOUT_SECONDS) as response: # noqa: S310 # the request built above + status, content = response.status, response.read() + except HTTPError as error: + status, content = error.code, error.read() + except OSError as error: + return Answer(status=None, error=type(error).__name__) + try: + body = json.loads(content) + except ValueError: + body = None + return Answer(status=status, body=body) + + +def _version_sort_key(key: str) -> tuple[bool, tuple[int, ...], str]: + parts = key.split(".") + if all(part.isascii() and part.isdigit() for part in parts): + return (True, tuple(int(part) for part in parts), key) + return (False, (), key) + + +def application_versions(answer: object) -> dict[str, str]: + """Map each application to the PolicyEngine version whose route names it.""" + routes = answer.get("policyengine") if isinstance(answer, dict) else None + if not isinstance(routes, dict): + return {} + versions: dict[str, str] = {} + for version, application in sorted( + routes.items(), key=lambda entry: _version_sort_key(str(entry[0])) + ): + if version != LATEST_KEY and isinstance(application, str): + versions.setdefault(application, str(version)) + return versions + + +def _served_version(application: object, versions: dict[str, str]) -> str: + if isinstance(application, str) and application in versions: + return f"serves policyengine {versions[application]}" + return f"application with no policyengine route: {application!r}" + + +def compare_routes( + where: str, + gateway_routes: object, + manifest_routes: object, + *, + gateway_versions: dict[str, str], + manifest_versions: dict[str, str], +) -> list[Difference]: + """Compare two route maps by the PolicyEngine version each route serves.""" + + def normalized(routes: object, versions: dict[str, str]) -> dict[str, object]: + if not isinstance(routes, dict): + return {} + return { + str(key): value if key == LATEST_KEY else _served_version(value, versions) + for key, value in routes.items() + } + + gateway = normalized(gateway_routes, gateway_versions) + manifest = normalized(manifest_routes, manifest_versions) + return [ + Difference(where, key, gateway.get(key, ABSENT), manifest.get(key, ABSENT)) + for key in sorted(set(gateway) | set(manifest), key=_version_sort_key) + if gateway.get(key, ABSENT) != manifest.get(key, ABSENT) + ] + + +def compare_values( + where: str, path: tuple[str, ...], gateway: object, manifest: object +) -> list[Difference]: + """Compare two JSON values, descending into objects key by key.""" + if isinstance(gateway, dict) and isinstance(manifest, dict): + differences: list[Difference] = [] + for key in sorted(set(gateway) | set(manifest), key=_version_sort_key): + differences.extend( + compare_values( + where, + (*path, str(key)), + gateway.get(key, ABSENT), + manifest.get(key, ABSENT), + ) + ) + return differences + if gateway == manifest: + return [] + return [Difference(where, ".".join(path), gateway, manifest)] + + +def compare_whole_answers(gateway: dict, manifest: dict) -> list[Difference]: + """Compare two ``GET /versions`` answers field by field.""" + gateway_versions = application_versions(gateway) + manifest_versions = application_versions(manifest) + differences: list[Difference] = [] + for field in sorted(set(gateway) | set(manifest)): + where = f"GET /versions: {field}" + if field in KINDS: + differences.extend( + compare_routes( + where, + gateway.get(field, ABSENT), + manifest.get(field, ABSENT), + gateway_versions=gateway_versions, + manifest_versions=manifest_versions, + ) + ) + else: + differences.extend( + compare_values( + where, (), gateway.get(field, ABSENT), manifest.get(field, ABSENT) + ) + ) + return differences + + +def _status_difference(where: str, gateway: Answer, manifest: Answer) -> Difference: + return Difference(where, "status", gateway.describe(), manifest.describe()) + + +def _describe_body(body: object) -> str: + if isinstance(body, dict): + return "JSON object" + return f"not a JSON object ({type(body).__name__})" + + +def _uncomparable(where: str, gateway: Answer, manifest: Answer) -> Difference | None: + """Return why two answers cannot be compared as JSON objects, or None. + + Both must be HTTP 200 with a JSON object body. Two answers that fail the + same way were still not compared, so they are reported too. + """ + if gateway.status != HTTP_OK or manifest.status != HTTP_OK: + return _status_difference(where, gateway, manifest) + if not isinstance(gateway.body, dict) or not isinstance(manifest.body, dict): + return Difference( + where, "body", _describe_body(gateway.body), _describe_body(manifest.body) + ) + return None + + +def compare_sources(base_url: str, fetch_answer: Fetch = fetch) -> list[Difference]: + """Fetch both sources' answers from ``base_url`` and list their differences.""" + differences: list[Difference] = [] + gateway_whole = fetch_answer(f"{base_url}/versions") + manifest_whole = fetch_answer(f"{base_url}/internal/v2/versions") + whole_problem = _uncomparable("GET /versions", gateway_whole, manifest_whole) + if whole_problem is not None: + differences.append(whole_problem) + elif isinstance(gateway_whole.body, dict) and isinstance(manifest_whole.body, dict): + differences.extend( + compare_whole_answers(gateway_whole.body, manifest_whole.body) + ) + + kind_answers = { + kind: ( + fetch_answer(f"{base_url}/versions/{kind}"), + fetch_answer(f"{base_url}/internal/v2/versions/{kind}"), + ) + for kind in KINDS + } + # Each source's own policyengine routes name the version each of its + # applications serves, even when its whole answer is unavailable. + gateway_policyengine, manifest_policyengine = kind_answers["policyengine"] + gateway_versions = application_versions({"policyengine": gateway_policyengine.body}) + manifest_versions = application_versions( + {"policyengine": manifest_policyengine.body} + ) + for kind, (gateway_kind, manifest_kind) in kind_answers.items(): + where = f"GET /versions/{kind}" + kind_problem = _uncomparable(where, gateway_kind, manifest_kind) + if kind_problem is not None: + differences.append(kind_problem) + continue + differences.extend( + compare_routes( + where, + gateway_kind.body, + manifest_kind.body, + gateway_versions=gateway_versions, + manifest_versions=manifest_versions, + ) + ) + + where = f"GET /versions/{UNKNOWN_KIND}" + gateway_unknown = fetch_answer(f"{base_url}/versions/{UNKNOWN_KIND}") + manifest_unknown = fetch_answer(f"{base_url}/internal/v2/versions/{UNKNOWN_KIND}") + if ( + gateway_unknown.status != manifest_unknown.status + or gateway_unknown.status is None + ): + # Two routes that both gave no answer were not compared, so they are + # reported too. + differences.append(_status_difference(where, gateway_unknown, manifest_unknown)) + else: + differences.extend( + compare_values( + where, ("body",), gateway_unknown.body, manifest_unknown.body + ) + ) + return differences + + +def _cell(value: object) -> str: + if value is ABSENT: + text = repr(ABSENT) + elif isinstance(value, str): + text = value + else: + text = json.dumps(value, sort_keys=True) + if len(text) > MAX_CELL_LENGTH: + text = text[: MAX_CELL_LENGTH - 1] + "…" + return text.replace("|", "\\|").replace("\n", " ") + + +_REPORT_HEADING = "## `/versions`: the old gateway compared with the v2 manifest" +_REPORT_NOTE = ( + "This comparison is for information only and does not fail the deploy. " + "Routes are compared by the PolicyEngine version each application serves, " + "because the old gateway names v1 applications and the manifest names v2 " + "applications." +) + + +def render_report(base_url: str, differences: list[Difference]) -> str: + """Return the Markdown report of ``differences`` found at ``base_url``.""" + lines = [ + _REPORT_HEADING, + "", + f"Compared at `{base_url}`. {_REPORT_NOTE}", + "", + ] + if not differences: + lines.append("**The answers match.**") + return "\n".join(lines) + "\n" + count = len(differences) + lines.extend( + [ + f"**{count} difference{'s' if count != 1 else ''}.**", + "", + "| Where | Key | Old gateway | v2 manifest |", + "| --- | --- | --- | --- |", + ] + ) + lines.extend( + f"| {_cell(difference.where)} | {_cell(difference.key)} | " + f"{_cell(difference.gateway)} | {_cell(difference.manifest)} |" + for difference in differences + ) + return "\n".join(lines) + "\n" + + +def render_failure(base_url: str, error: BaseException) -> str: + """Return the Markdown report for a comparison that could not finish.""" + return ( + f"{_REPORT_HEADING}\n\nCompared at `{base_url}`. {_REPORT_NOTE}\n\n" + f"**The comparison could not finish** ({type(error).__name__}).\n" + ) + + +def write_summary(report: str) -> None: + """Append ``report`` to the job summary, when the runner names one.""" + summary_path = os.environ.get("GITHUB_STEP_SUMMARY") + if not summary_path: + return + try: + with open(summary_path, "a", encoding="utf-8") as summary: # noqa: PTH123 # the runner names this file by path + summary.write(report) + except OSError as error: + print(f"Could not write the job summary: {type(error).__name__}") + + +def main(argv: list[str]) -> int: + """Compare the sources at the base URL in ``argv`` and report; return 0.""" + if len(argv) != EXPECTED_ARGUMENT_COUNT or not argv[1].strip(): + print("Usage: compare_versions_sources.py BASE_URL", file=sys.stderr) + return USAGE_ERROR + base_url = argv[1].strip().rstrip("/") + try: + report = render_report(base_url, compare_sources(base_url)) + # The report is for information only and must never fail the deploy. + except Exception as error: # noqa: BLE001 # any failure is reported, not raised + report = render_failure(base_url, error) + print(report) + write_summary(report) + return 0 + + +if __name__ == "__main__": + sys.exit(main(sys.argv)) diff --git a/.github/workflows/simulation-deploy.reusable.yml b/.github/workflows/simulation-deploy.reusable.yml index 9de0a3eff..afe5cc406 100644 --- a/.github/workflows/simulation-deploy.reusable.yml +++ b/.github/workflows/simulation-deploy.reusable.yml @@ -746,6 +746,17 @@ jobs: EXPECTED_LATEST: ${{ inputs.deploy_existing_stack && needs.update_routing.outputs.advances_latest == 'true' && 'true' || '' }} run: .github/scripts/cloud-run-simulation-entry-smoke.sh "${{ needs.deploy_entrypoint.outputs.candidate_url }}" "${{ needs.deploy_entrypoint.outputs.revision }}" + # /versions still forwards to the old gateway, which reads v1's routing + # state; /internal/v2/versions gives the v2 version manifest's answers. + # The report goes to the job summary and never fails the deploy: the + # sources are known to differ until the manifest records everything + # the gateway reports. + - name: Compare /versions with the v2 manifest's answers (report only) + if: ${{ inputs.deployment_environment == 'staging' }} + env: + CANDIDATE_URL: ${{ needs.deploy_entrypoint.outputs.candidate_url }} + run: python .github/scripts/compare_versions_sources.py "${CANDIDATE_URL}" + - name: Download the numerical release sentinel if: ${{ inputs.deploy_existing_stack }} uses: actions/download-artifact@v4 diff --git a/projects/policyengine-simulation-entry/README.md b/projects/policyengine-simulation-entry/README.md index 3a3f72815..a0495b9ab 100644 --- a/projects/policyengine-simulation-entry/README.md +++ b/projects/policyengine-simulation-entry/README.md @@ -41,7 +41,18 @@ Auth0 permission: | `GET /internal/stage12/reports/{evaluation_id}` | `stage12:operate` | `GET /health`, `GET /ready`, `GET /versions`, `GET /versions/{kind}` and -`POST /ping` stay public. +`POST /ping` stay public. `GET /health` and `POST /ping` are answered by the +entrypoint itself: `/ping` returns the request's `value` plus one as +`incremented`, as the old gateway does, without calling it. + +`GET /internal/v2/versions` and `GET /internal/v2/versions/{kind}` are public +too, and are not in the OpenAPI document. They give the answers `/versions` +and `/versions/{kind}` would give from the v2 version manifest +(`policyengine_simulation_entry.v2_versions`), in the same response models, +while `/versions` still forwards to the old gateway. A manifest that cannot be +read gives a 503. Each staging deploy compares the candidate's two answers +with `.github/scripts/compare_versions_sources.py` and writes the differences +to the job summary; the comparison never fails the deploy. A caller holds a permission when it appears in the token's `permissions` claim (Auth0 RBAC). For a client-credentials (machine) token, an entry in its diff --git a/projects/policyengine-simulation-entry/src/policyengine_simulation_entry/app.py b/projects/policyengine-simulation-entry/src/policyengine_simulation_entry/app.py index 9bb6d3af9..dfbbcadc9 100644 --- a/projects/policyengine-simulation-entry/src/policyengine_simulation_entry/app.py +++ b/projects/policyengine-simulation-entry/src/policyengine_simulation_entry/app.py @@ -73,11 +73,20 @@ TemporaryStage12DispatchFailed, TemporaryStage12UnsupportedRequestError, ) +from policyengine_simulation_entry.v2_versions import ( + ModalV2VersionsSource, + UnknownVersionKindError, + V2VersionsSource, + is_version_kind, + version_map_from_manifest, + versions_from_manifest, +) if TYPE_CHECKING: from collections.abc import AsyncIterator from policyengine_observability import ObservabilityRuntime + from policyengine_simulation_contract.stage12_manifest import V2VersionManifest from starlette.middleware.base import RequestResponseEndpoint logger = logging.getLogger(__name__) @@ -87,6 +96,9 @@ _json_object_adapter = TypeAdapter(JsonObject) type ResponseIdentifier = Literal["job_id", "batch_job_id"] STAGE12_MODAL_SUBMISSION_TIMEOUT_SECONDS = 5.0 +# How long the internal v2 version routes wait for the manifest's Modal Dict. +V2_MANIFEST_READ_TIMEOUT_SECONDS = 5.0 +V2_MANIFEST_UNAVAILABLE_DETAIL = "The v2 version manifest is unavailable." class ComparisonBackend(Protocol): @@ -645,6 +657,72 @@ async def get_temporary_stage12_report( ) +def _v2_manifest_unavailable_response() -> JSONResponse: + """Return the 503 the internal v2 version routes send without a manifest.""" + return JSONResponse( + status_code=503, + content={"detail": V2_MANIFEST_UNAVAILABLE_DETAIL}, + headers={"Retry-After": "10"}, + ) + + +def _register_v2_version_routes( + app: FastAPI, + versions_source: V2VersionsSource | None, +) -> None: + """Add the internal routes that answer ``/versions`` from the v2 manifest. + + They answer as ``GET /versions`` and ``GET /versions/{kind}`` would from + the v2 version manifest, so a staging deploy can compare those answers + with the old gateway's while ``/versions`` still forwards to it. Like + ``/versions`` they are public, and they are left out of OpenAPI. They + never call the old gateway. + """ + + async def read_manifest(request: Request) -> V2VersionManifest | None: + if versions_source is None: + return None + try: + return await asyncio.wait_for( + versions_source.load_manifest(), + timeout=V2_MANIFEST_READ_TIMEOUT_SECONDS, + ) + # A missing, invalid or unreachable manifest, or a timeout, becomes a + # 503 rather than an unhandled error. + except Exception as error: # noqa: BLE001 # every manifest read failure becomes a 503 + logger.error( # noqa: TRY400 # the error type describes the failure; a traceback adds only Modal client internals + "v2_version_manifest_read_failed", + extra={ + "request_id": request.state.request_id, + "error_type": type(error).__name__, + }, + ) + return None + + @app.get("/internal/v2/versions", include_in_schema=False) + async def list_v2_manifest_versions(request: Request) -> Response: + manifest = await read_manifest(request) + if manifest is None: + return _v2_manifest_unavailable_response() + return JSONResponse(versions_from_manifest(manifest).model_dump(mode="json")) + + @app.get("/internal/v2/versions/{kind}", include_in_schema=False) + async def get_v2_manifest_kind_versions(kind: str, request: Request) -> Response: + # As in the old gateway, an unknown kind is a 404 whether or not the + # manifest can be read. + if not is_version_kind(kind): + return JSONResponse( + status_code=404, + content={"detail": UnknownVersionKindError(kind).detail}, + ) + manifest = await read_manifest(request) + if manifest is None: + return _v2_manifest_unavailable_response() + return JSONResponse( + version_map_from_manifest(manifest, kind).model_dump(mode="json") + ) + + def _register_simulation_routes( app: FastAPI, entry: _EntryRuntime, @@ -845,10 +923,10 @@ async def ready() -> Response: operation_id="ping_ping_post", response_model=PingResponse, ) - async def ping(body: PingRequest, request: Request) -> Response: - return await entry.forward( - request, _BackendCall("POST", "/ping", _model_json(body)) - ) + async def ping(body: PingRequest) -> PingResponse: + # The entrypoint answers itself, exactly as the old gateway does: the + # request's value plus one. The old gateway is not called. + return PingResponse(incremented=body.value + 1) def create_app( @@ -857,8 +935,13 @@ def create_app( backend: SimulationBackend | None = None, auth_dependency: AuthenticationDependency | None = None, comparison_backend: ComparisonBackend | None = None, + v2_versions_source: V2VersionsSource | None = None, ) -> FastAPI: - """Build the app with injectable auth/backend seams for hermetic tests.""" + """Build the app with injectable auth/backend seams for hermetic tests. + + Without ``v2_versions_source``, the internal v2 version routes read the + manifest Dict the settings name, when they name one. + """ runtime_settings = settings or Settings.from_env() @asynccontextmanager @@ -919,6 +1002,10 @@ async def lifespan(_: FastAPI) -> AsyncIterator[None]: reader=simulation_reader, ) _register_service_routes(app, entry) + runtime_versions_source = v2_versions_source + if runtime_versions_source is None: + runtime_versions_source = ModalV2VersionsSource.from_settings(runtime_settings) + _register_v2_version_routes(app, runtime_versions_source) return app diff --git a/projects/policyengine-simulation-entry/src/policyengine_simulation_entry/v2_versions.py b/projects/policyengine-simulation-entry/src/policyengine_simulation_entry/v2_versions.py new file mode 100644 index 000000000..f63385db0 --- /dev/null +++ b/projects/policyengine-simulation-entry/src/policyengine_simulation_entry/v2_versions.py @@ -0,0 +1,260 @@ +"""``/versions`` answers built from the v2 version manifest. + +The old gateway answers ``GET /versions`` and ``GET /versions/{kind}`` from +v1's routing state. These functions build answers with the same response +models from the v2 version manifest instead, for the v2 side of the +entrypoint. They share no code with the routes that forward to the old +gateway. + +What the manifest can and cannot say: + +- Each route names the v2 worker application (``policyengine-simulation-v2-py…``) + that serves a version, where the old gateway names the v1 application. +- A country model version that several bundles contain routes to the newest + of them by version order. The old gateway routes it to the last bundle + published, which is the newest unless an older bundle was republished. +- ``latest`` is the manifest's default version, and each country's ``latest`` + is that bundle's model version. +- ``spm_capabilities`` is always empty: the manifest records no SPM + capability. + +``ModalV2VersionsSource`` reads the manifest from the Modal Dict that the +deploy publishes it to, the same one the Stage 12 comparison path reads. +""" + +from __future__ import annotations + +import asyncio +from concurrent.futures import ThreadPoolExecutor +from typing import TYPE_CHECKING, Protocol + +import modal +from policyengine_simulation_contract.dataset_uri import runtime_dataset_uri +from policyengine_simulation_contract.gateway_models import ( + PolicyEngineBundle, + VersionMap, + VersionsResponse, +) +from policyengine_simulation_contract.stage12_manifest import V2ManifestLoader + +if TYPE_CHECKING: + from policyengine_simulation_contract.stage12_bundle import ( + Stage12BundleManifest, + Stage12CountryBundle, + ) + from policyengine_simulation_contract.stage12_manifest import ( + V2ManifestStore, + V2VersionManifest, + V2WorkerVersion, + ) + + from policyengine_simulation_entry.config import Settings + +POLICYENGINE_KIND = "policyengine" +COUNTRY_KINDS = ("us", "uk") +VERSION_KINDS = (POLICYENGINE_KIND, *COUNTRY_KINDS) +LATEST_KEY = "latest" + + +class UnknownVersionKindError(LookupError): + """Raised for a ``/versions/{kind}`` kind that is not policyengine, us or uk.""" + + def __init__(self, kind: str) -> None: + """Record the response detail the old gateway sends for this kind.""" + self.detail = f"Unknown version kind: {kind}" + super().__init__(self.detail) + + +def is_version_kind(kind: str) -> bool: + """Return whether ``kind`` is policyengine, us or uk, in any case.""" + return kind.lower() in VERSION_KINDS + + +class V2VersionsSource(Protocol): + """Where the entrypoint reads the v2 version manifest from.""" + + async def load_manifest(self) -> V2VersionManifest: + """Return the current manifest; raise if none can be read.""" + ... + + +class ModalV2VersionsSource: + """Reads the v2 version manifest from its Modal Dict. + + The loader keeps the last valid manifest it read, and returns it if the + stored value later goes missing or becomes invalid. With no valid + manifest read yet, ``load_manifest`` raises ValueError. + """ + + def __init__(self, store: V2ManifestStore) -> None: + """Read from ``store``; nothing is read until ``load_manifest``.""" + self._loader = V2ManifestLoader(store) + # The Modal client blocks, so reads run on a thread. One of this + # source's own is enough, and keeps a stalled read from holding + # threads of the event loop's default executor, which the Stage 12 + # comparison path uses. One thread also means one read at a time, as + # the loader expects. + self._read_thread = ThreadPoolExecutor( + max_workers=1, thread_name_prefix="v2-version-manifest" + ) + + @classmethod + def from_settings(cls, settings: Settings) -> ModalV2VersionsSource | None: + """Open the configured manifest Dict, or return None if none is configured. + + The deploy sets the manifest's Modal environment in staging and + production. Opening the Dict is lazy: nothing is read here, and the + Dict is never created. + """ + if not settings.stage12_v2_manifest_environment: + return None + store = modal.Dict.from_name( + settings.stage12_v2_manifest_name, + environment_name=settings.stage12_v2_manifest_environment, + ) + return cls(store) + + async def load_manifest(self) -> V2VersionManifest: + """Read the manifest on this source's thread.""" + return await asyncio.get_running_loop().run_in_executor( + self._read_thread, self._loader.load + ) + + +def _version_order(policyengine_version: str) -> tuple[bool, tuple[int, ...], str]: + """Order versions oldest first. + + A version made of dot-separated numbers, such as ``6.10.0``, is ordered by + those numbers. Any other version sorts before every such version, by its + text. + """ + parts = policyengine_version.split(".") + if all(part.isascii() and part.isdigit() for part in parts): + return (True, tuple(int(part) for part in parts), policyengine_version) + return (False, (), policyengine_version) + + +def _oldest_first( + manifest: V2VersionManifest, +) -> list[tuple[str, V2WorkerVersion]]: + return sorted( + manifest.versions.items(), + key=lambda entry: _version_order(entry[0]), + ) + + +def _country_bundle( + worker: V2WorkerVersion, country: str +) -> Stage12CountryBundle | None: + return next( + ( + country_bundle + for country_bundle in worker.bundle.countries + if country_bundle.country == country + ), + None, + ) + + +def _policyengine_routes(manifest: V2VersionManifest) -> dict[str, str]: + routes = { + version: worker.application_name for version, worker in _oldest_first(manifest) + } + routes[LATEST_KEY] = manifest.default_version + return routes + + +def _country_routes(manifest: V2VersionManifest, country: str) -> dict[str, str]: + routes: dict[str, str] = {} + # Oldest first, so a model version that several bundles contain ends up + # routed to the newest of them. + for _, worker in _oldest_first(manifest): + country_bundle = _country_bundle(worker, country) + if country_bundle is not None: + routes[country_bundle.country_package_version] = worker.application_name + default_bundle = _country_bundle( + manifest.versions[manifest.default_version], country + ) + if default_bundle is not None: + routes[LATEST_KEY] = default_bundle.country_package_version + return routes + + +def version_map_from_manifest(manifest: V2VersionManifest, kind: str) -> VersionMap: + """Return ``kind``'s routes and latest version, as ``GET /versions/{kind}`` does. + + ``kind`` is matched without regard to case. Raises UnknownVersionKindError + for any kind other than policyengine, us or uk. + """ + kind_lower = kind.lower() + if kind_lower not in VERSION_KINDS: + raise UnknownVersionKindError(kind) + if kind_lower == POLICYENGINE_KIND: + return VersionMap(_policyengine_routes(manifest)) + return VersionMap(_country_routes(manifest, kind_lower)) + + +def _default_dataset_sha256(country_bundle: Stage12CountryBundle) -> str | None: + return next( + ( + dataset.sha256 + for dataset in country_bundle.datasets + if dataset.identity == country_bundle.default_dataset + ), + None, + ) + + +def _runtime_dataset(country_bundle: Stage12CountryBundle) -> str | None: + """Return the default dataset's runtime URI, or None if it cannot be resolved. + + As in the old gateway, one entry with an unresolvable dataset is reported + without one rather than failing the whole answer. ``validate_hf=False`` + trusts the bundle and makes no network call. + """ + try: + return runtime_dataset_uri( + country_bundle.default_dataset_uri, + default_revision=country_bundle.data_package_version, + artifact_revision=country_bundle.data_artifact_revision, + validate_hf=False, + ) + except ValueError: + # HuggingFaceDatasetReferenceError is a ValueError. + return None + + +def _country_identity( + bundle: Stage12BundleManifest, + country_bundle: Stage12CountryBundle, +) -> PolicyEngineBundle: + return PolicyEngineBundle( + model_version=country_bundle.country_package_version, + policyengine_version=bundle.policyengine_version, + data_version=country_bundle.data_release_version, + dataset=_runtime_dataset(country_bundle), + dataset_sha256=_default_dataset_sha256(country_bundle), + ) + + +def _bundle_identities( + manifest: V2VersionManifest, +) -> dict[str, dict[str, PolicyEngineBundle]]: + return { + version: { + country_bundle.country: _country_identity(worker.bundle, country_bundle) + for country_bundle in worker.bundle.countries + } + for version, worker in _oldest_first(manifest) + } + + +def versions_from_manifest(manifest: V2VersionManifest) -> VersionsResponse: + """Return the whole ``GET /versions`` answer for ``manifest``.""" + return VersionsResponse( + spm_capabilities={}, + policyengine=version_map_from_manifest(manifest, POLICYENGINE_KIND), + us=version_map_from_manifest(manifest, "us"), + uk=version_map_from_manifest(manifest, "uk"), + bundles=_bundle_identities(manifest), + ) diff --git a/projects/policyengine-simulation-entry/tests/test_app.py b/projects/policyengine-simulation-entry/tests/test_app.py index 200ea4f41..b168374fd 100644 --- a/projects/policyengine-simulation-entry/tests/test_app.py +++ b/projects/policyengine-simulation-entry/tests/test_app.py @@ -819,15 +819,57 @@ def test_budget_window_routes_use_original_batch_id(client, backend): assert backend.requests[-1].path == "/budget-window-jobs/batch-123" -def test_versions_and_ping_are_public_proxy_routes(client, backend): +def test_versions_are_public_proxy_routes(client, backend): backend.responses[("GET", "/versions/us")] = response( 200, {"latest": "1.2.3"}, ) - backend.responses[("POST", "/ping")] = response(200, {"incremented": 2}) assert client.get("/versions/us").json() == {"latest": "1.2.3"} - assert client.post("/ping", json={"value": 1}).json() == {"incremented": 2} + assert backend.requests[-1].path == "/versions/us" + + +@pytest.mark.parametrize("value", [-3, 0, 1, 2**70]) +def test_ping_is_answered_by_the_entrypoint_itself(client, backend, value): + # The old gateway's /ping returns the request's value plus one; the + # entrypoint now gives that answer without calling it. + result = client.post("/ping", json={"value": value}) + + assert result.status_code == 200 + assert result.json() == {"incremented": value + 1} + assert result.headers["content-type"] == "application/json" + assert "x-policyengine-simulation-backend" not in result.headers + assert backend.requests == [] + + +def test_ping_answers_while_the_old_gateway_is_unavailable(): + class UnavailableBackend(FakeBackend): + @override + async def request(self, *args, **kwargs): + msg = "unavailable" + raise BackendUnavailableError(msg) + + app = create_app( + settings=make_settings(), + backend=UnavailableBackend(), + auth_dependency=lambda: None, + ) + + from fastapi.testclient import TestClient + + with TestClient(app) as test_client: + result = test_client.post("/ping", json={"value": 41}) + + assert result.status_code == 200 + assert result.json() == {"incremented": 42} + + +@pytest.mark.parametrize("body", [{}, {"value": "one"}, {"value": 1.5}]) +def test_ping_rejects_a_body_without_an_integer_value(client, backend, body): + result = client.post("/ping", json=body) + + assert result.status_code == 422 + assert backend.requests == [] def test_request_id_is_propagated_logged_and_returned( diff --git a/projects/policyengine-simulation-entry/tests/test_compare_versions_sources.py b/projects/policyengine-simulation-entry/tests/test_compare_versions_sources.py new file mode 100644 index 000000000..b2e3f3600 --- /dev/null +++ b/projects/policyengine-simulation-entry/tests/test_compare_versions_sources.py @@ -0,0 +1,399 @@ +"""Tests for .github/scripts/compare_versions_sources.py. + +The staging deploy runs the script against its candidate revision to compare +the old gateway's ``/versions`` answers, which the entrypoint forwards, with +the v2 manifest's answers on ``/internal/v2/versions``. Tests of the +comparison load the script in this process; tests of a whole run start it as +the workflow does, against a local HTTP server, so no test leaves this +machine. +""" + +from __future__ import annotations + +import copy +import importlib.util +import json +import subprocess +import sys +import threading +from contextlib import contextmanager +from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer +from pathlib import Path +from typing import TYPE_CHECKING, ClassVar + +import pytest +import yaml +from workflow_script_support import SCRIPTS_DIR, WORKFLOWS_DIR + +if TYPE_CHECKING: + from collections.abc import Iterator + from types import ModuleType + +COMPARE_SCRIPT = SCRIPTS_DIR / "compare_versions_sources.py" +REUSABLE_DEPLOY_WORKFLOW = WORKFLOWS_DIR / "simulation-deploy.reusable.yml" +COMPARE_STEP = "Compare /versions with the v2 manifest's answers (report only)" +# The gateway's real /versions answer for two published releases; the +# gateway project's test_versions_fixture.py keeps it current. +GATEWAY_ANSWER = json.loads( + (Path(__file__).parent / "fixtures" / "versions_body.json").read_text() +) +UNKNOWN_KIND_BODY = {"detail": "Unknown version kind: unknown-kind"} + + +def manifest_answer_for(gateway_answer: dict) -> dict: + """Return ``gateway_answer`` as the manifest gives it: v2 application names.""" + answer = copy.deepcopy(gateway_answer) + for kind in ("policyengine", "us", "uk"): + answer[kind] = { + key: value + if key == "latest" + else value.replace( + "policyengine-simulation-py", "policyengine-simulation-v2-py" + ) + for key, value in answer[kind].items() + } + return answer + + +def route_answers(gateway_answer: dict, manifest_answer: dict) -> dict: + """Return each route's (status, body) for two whole answers.""" + routes = { + "/versions": (200, gateway_answer), + "/internal/v2/versions": (200, manifest_answer), + "/versions/unknown-kind": (404, UNKNOWN_KIND_BODY), + "/internal/v2/versions/unknown-kind": (404, UNKNOWN_KIND_BODY), + } + for kind in ("policyengine", "us", "uk"): + routes[f"/versions/{kind}"] = (200, gateway_answer[kind]) + routes[f"/internal/v2/versions/{kind}"] = (200, manifest_answer[kind]) + return routes + + +@pytest.fixture(scope="module") +def compare_module() -> ModuleType: + """Load the comparison script as a module in this process.""" + spec = importlib.util.spec_from_file_location( + "compare_versions_sources", COMPARE_SCRIPT + ) + assert spec is not None + assert spec.loader is not None + module = importlib.util.module_from_spec(spec) + # Dataclasses resolve their module through sys.modules. + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +def differences_between(compare_module, routes: dict) -> list: + def fetch_answer(url: str): + path = url.removeprefix("https://candidate.example") + status, body = routes[path] + return compare_module.Answer(status=status, body=body) + + return compare_module.compare_sources("https://candidate.example", fetch_answer) + + +def test_answers_that_differ_only_in_application_names_match(compare_module): + routes = route_answers(GATEWAY_ANSWER, manifest_answer_for(GATEWAY_ANSWER)) + + assert differences_between(compare_module, routes) == [] + + +def test_a_version_only_the_gateway_routes_is_a_difference(compare_module): + manifest_answer = manifest_answer_for(GATEWAY_ANSWER) + del manifest_answer["policyengine"]["6.2.6"] + del manifest_answer["uk"]["2.125.1"] + del manifest_answer["bundles"]["6.2.6"] + + differences = differences_between( + compare_module, route_answers(GATEWAY_ANSWER, manifest_answer) + ) + + found = {(difference.where, difference.key) for difference in differences} + assert ("GET /versions: policyengine", "6.2.6") in found + assert ("GET /versions: uk", "2.125.1") in found + assert ("GET /versions/uk", "2.125.1") in found + assert ("GET /versions: bundles", "6.2.6") in found + [policyengine_difference] = [ + difference + for difference in differences + if difference.where == "GET /versions: policyengine" + ] + assert policyengine_difference.gateway == "serves policyengine 6.2.6" + assert policyengine_difference.manifest is compare_module.ABSENT + + +def test_a_model_version_routed_to_another_bundle_is_a_difference(compare_module): + manifest_answer = manifest_answer_for(GATEWAY_ANSWER) + manifest_answer["us"]["1.700.0"] = "policyengine-simulation-v2-py6-2-6" + + differences = differences_between( + compare_module, route_answers(GATEWAY_ANSWER, manifest_answer) + ) + + assert [ + (difference.where, difference.key, difference.gateway, difference.manifest) + for difference in differences + ] == [ + ( + "GET /versions: us", + "1.700.0", + "serves policyengine 6.2.7", + "serves policyengine 6.2.6", + ), + ( + "GET /versions/us", + "1.700.0", + "serves policyengine 6.2.7", + "serves policyengine 6.2.6", + ), + ] + + +def test_a_different_latest_version_is_a_difference(compare_module): + manifest_answer = manifest_answer_for(GATEWAY_ANSWER) + manifest_answer["policyengine"]["latest"] = "6.2.6" + + differences = differences_between( + compare_module, route_answers(GATEWAY_ANSWER, manifest_answer) + ) + + assert { + (difference.where, difference.key, difference.gateway, difference.manifest) + for difference in differences + } == { + ("GET /versions: policyengine", "latest", "6.2.7", "6.2.6"), + ("GET /versions/policyengine", "latest", "6.2.7", "6.2.6"), + } + + +def test_bundle_fields_and_spm_capabilities_are_compared_value_by_value( + compare_module, +): + gateway_answer = copy.deepcopy(GATEWAY_ANSWER) + capability = {"contract_version": "canonical-spm-v1", "defaults": {}} + gateway_answer["spm_capabilities"] = {"6.2.7": capability} + manifest_answer = manifest_answer_for(GATEWAY_ANSWER) + manifest_answer["bundles"]["6.2.6"]["us"]["dataset_sha256"] = "e" * 64 + + differences = differences_between( + compare_module, route_answers(gateway_answer, manifest_answer) + ) + + assert { + (difference.where, difference.key, json.dumps(difference.gateway, default=repr)) + for difference in differences + } == { + ("GET /versions: bundles", "6.2.6.us.dataset_sha256", "null"), + ( + "GET /versions: spm_capabilities", + "6.2.7", + json.dumps(capability), + ), + } + + +def test_a_route_that_does_not_answer_is_reported_by_status(compare_module): + routes = route_answers(GATEWAY_ANSWER, manifest_answer_for(GATEWAY_ANSWER)) + routes["/internal/v2/versions"] = ( + 503, + {"detail": "The v2 version manifest is unavailable."}, + ) + + differences = differences_between(compare_module, routes) + + assert [ + (difference.where, difference.key, difference.gateway, difference.manifest) + for difference in differences + ] == [("GET /versions", "status", "HTTP 200", "HTTP 503")] + + +@pytest.mark.parametrize( + ("gateway_body", "manifest_body", "gateway_description", "manifest_description"), + [ + ([], [], "not a JSON object (list)", "not a JSON object (list)"), + (None, None, "not a JSON object (NoneType)", "not a JSON object (NoneType)"), + ({"latest": "1.700.0"}, "text", "JSON object", "not a JSON object (str)"), + ], +) +def test_a_kind_answer_that_is_not_a_json_object_is_never_a_match( + compare_module, + gateway_body, + manifest_body, + gateway_description, + manifest_description, +): + routes = route_answers(GATEWAY_ANSWER, manifest_answer_for(GATEWAY_ANSWER)) + routes["/versions/us"] = (200, gateway_body) + routes["/internal/v2/versions/us"] = (200, manifest_body) + + differences = differences_between(compare_module, routes) + + assert [ + (difference.where, difference.key, difference.gateway, difference.manifest) + for difference in differences + ] == [("GET /versions/us", "body", gateway_description, manifest_description)] + + +def test_a_whole_answer_that_is_not_a_json_object_is_reported_by_body( + compare_module, +): + routes = route_answers(GATEWAY_ANSWER, manifest_answer_for(GATEWAY_ANSWER)) + routes["/versions"] = (200, []) + + differences = differences_between(compare_module, routes) + + assert [ + (difference.where, difference.key, difference.gateway, difference.manifest) + for difference in differences + ] == [("GET /versions", "body", "not a JSON object (list)", "JSON object")] + + +def test_an_unknown_kind_must_be_rejected_the_same_way(compare_module): + routes = route_answers(GATEWAY_ANSWER, manifest_answer_for(GATEWAY_ANSWER)) + routes["/internal/v2/versions/unknown-kind"] = (404, {"detail": "Not Found"}) + + differences = differences_between(compare_module, routes) + + assert [(difference.where, difference.key) for difference in differences] == [ + ("GET /versions/unknown-kind", "body.detail") + ] + + +def test_the_report_lists_each_difference(compare_module): + report = compare_module.render_report( + "https://candidate.example", + [ + compare_module.Difference( + "GET /versions: policyengine", + "4.18.3", + "serves policyengine 4.18.3", + compare_module.ABSENT, + ) + ], + ) + + assert "Compared at `https://candidate.example`." in report + assert "does not fail the deploy" in report + assert "**1 difference.**" in report + assert ( + "| GET /versions: policyengine | 4.18.3 | serves policyengine 4.18.3 " + "| (absent) |" + ) in report + + +class _CannedRoutes(BaseHTTPRequestHandler): + routes: ClassVar[dict] = {} + + def do_GET(self): + status, body = self.routes.get(self.path, (404, {"detail": "Not Found"})) + content = json.dumps(body).encode() + self.send_response(status) + self.send_header("Content-Type", "application/json") + self.send_header("Content-Length", str(len(content))) + self.end_headers() + self.wfile.write(content) + + def log_message(self, *args): + return + + +@contextmanager +def serving(routes: dict) -> Iterator[str]: + """Serve ``routes`` on a loopback port and yield the base URL.""" + handler = type("CannedRoutes", (_CannedRoutes,), {"routes": routes}) + server = ThreadingHTTPServer(("127.0.0.1", 0), handler) + thread = threading.Thread(target=server.serve_forever, daemon=True) + thread.start() + try: + yield f"http://127.0.0.1:{server.server_address[1]}" + finally: + server.shutdown() + server.server_close() + + +def run_comparison(*args: str, summary: Path) -> subprocess.CompletedProcess[str]: + """Run the script as the workflow step does.""" + return subprocess.run( + [sys.executable, str(COMPARE_SCRIPT), *args], + capture_output=True, + text=True, + check=False, + env={"GITHUB_STEP_SUMMARY": str(summary), "PATH": "/usr/bin:/bin"}, + timeout=60, + ) + + +def test_a_run_with_matching_answers_reports_a_match(tmp_path): + summary = tmp_path / "summary.md" + routes = route_answers(GATEWAY_ANSWER, manifest_answer_for(GATEWAY_ANSWER)) + + with serving(routes) as base_url: + result = run_comparison(f"{base_url}/", summary=summary) + + assert result.returncode == 0, result.stderr + assert "**The answers match.**" in result.stdout + assert summary.read_text(encoding="utf-8") in result.stdout + assert f"Compared at `{base_url}`." in summary.read_text(encoding="utf-8") + + +def test_a_run_with_differences_reports_them_and_still_succeeds(tmp_path): + summary = tmp_path / "summary.md" + manifest_answer = manifest_answer_for(GATEWAY_ANSWER) + manifest_answer["spm_capabilities"] = {"6.2.7": {"contract_version": "x"}} + + with serving(route_answers(GATEWAY_ANSWER, manifest_answer)) as base_url: + result = run_comparison(base_url, summary=summary) + + assert result.returncode == 0, result.stderr + assert "**1 difference.**" in summary.read_text(encoding="utf-8") + + +def test_a_run_where_no_route_answers_reports_it_and_still_succeeds(tmp_path): + summary = tmp_path / "summary.md" + with serving({}) as base_url: + unreachable_url = base_url + # The server has stopped, so every request is refused. + + result = run_comparison(unreachable_url, summary=summary) + + assert result.returncode == 0, result.stderr + report = summary.read_text(encoding="utf-8") + assert "| GET /versions | status | no answer (" in report + assert "**5 differences.**" in report + + +@pytest.mark.parametrize("args", [(), ("",), ("https://a.example", "extra")]) +def test_a_run_without_exactly_one_base_url_is_a_usage_error(tmp_path, args): + summary = tmp_path / "summary.md" + + result = run_comparison(*args, summary=summary) + + assert result.returncode == 2 + assert "Usage: compare_versions_sources.py BASE_URL" in result.stderr + assert not summary.exists() + + +def _integration_steps() -> list[dict]: + workflow = yaml.safe_load(REUSABLE_DEPLOY_WORKFLOW.read_text(encoding="utf-8")) + return workflow["jobs"]["integration"]["steps"] + + +def test_the_staging_deploy_compares_the_candidates_two_answers(): + steps = _integration_steps() + names = [step.get("name") for step in steps] + [compare_step] = [step for step in steps if step.get("name") == COMPARE_STEP] + + # After the candidate's own smoke check, before the numerical checks. + assert names.index(COMPARE_STEP) == ( + names.index("Verify the complete deployed request path") + 1 + ) + assert compare_step["if"] == "${{ inputs.deployment_environment == 'staging' }}" + assert compare_step["env"] == { + "CANDIDATE_URL": "${{ needs.deploy_entrypoint.outputs.candidate_url }}" + } + assert compare_step["run"] == ( + 'python .github/scripts/compare_versions_sources.py "${CANDIDATE_URL}"' + ) + # A report-only step: nothing may turn its result into a pass. + assert "continue-on-error" not in compare_step diff --git a/projects/policyengine-simulation-entry/tests/test_v2_versions.py b/projects/policyengine-simulation-entry/tests/test_v2_versions.py new file mode 100644 index 000000000..eaff75e73 --- /dev/null +++ b/projects/policyengine-simulation-entry/tests/test_v2_versions.py @@ -0,0 +1,232 @@ +"""The ``/versions`` answers built from the v2 version manifest. + +The old gateway answers ``GET /versions`` and ``GET /versions/{kind}`` from +v1's routing state; these tests pin the same answers, in the same shapes, +built from the v2 version manifest instead. +""" + +from __future__ import annotations + +import json +from pathlib import Path + +import pytest +from policyengine_simulation_contract.gateway_models import VersionsResponse +from policyengine_simulation_entry.v2_versions import ( + UnknownVersionKindError, + version_map_from_manifest, + versions_from_manifest, +) +from v2_versions_support import ( + TWO_RELEASES, + manifest_worker, + version_manifest, + with_country_bundle, +) + +# The gateway's real /versions answer for two published releases; the +# gateway project's test_versions_fixture.py keeps it current. +GATEWAY_VERSIONS_ANSWER = json.loads( + (Path(__file__).parent / "fixtures" / "versions_body.json").read_text() +) +TEST_DIGEST = "c" * 64 + + +def test_policyengine_routes_each_version_to_its_application_and_latest_to_default(): + answer = version_map_from_manifest(TWO_RELEASES, "policyengine") + + assert answer.root == { + "6.2.6": "policyengine-simulation-v2-py6-2-6", + "6.2.7": "policyengine-simulation-v2-py6-2-7", + "latest": "6.2.7", + } + + +def test_country_routes_each_model_version_and_latest_to_the_default_bundles_model(): + assert version_map_from_manifest(TWO_RELEASES, "uk").root == { + "2.125.1": "policyengine-simulation-v2-py6-2-6", + "2.126.8": "policyengine-simulation-v2-py6-2-7", + "latest": "2.126.8", + } + assert version_map_from_manifest(TWO_RELEASES, "us").root == { + "1.700.0": "policyengine-simulation-v2-py6-2-7", + "latest": "1.700.0", + } + + +def test_a_model_version_in_several_bundles_routes_to_the_newest_bundle(): + # 6.10.0 is newer than 6.7.5 by version order, though not as text. The + # old gateway routes a shared model version to the last bundle published, + # which in staging and production is the newest. + manifest = version_manifest( + manifest_worker("6.10.0", us_version="2.2.1", uk_version="2.132.6"), + manifest_worker("6.7.5", us_version="2.2.1", uk_version="2.130.1"), + manifest_worker("6.2.1", us_version="2.2.1", uk_version="2.102.3"), + default_version="6.10.0", + ) + + assert version_map_from_manifest(manifest, "us").root == { + "2.2.1": "policyengine-simulation-v2-py6-10-0", + "latest": "2.2.1", + } + + +def test_a_version_that_is_not_only_numbers_counts_as_older_than_releases(): + manifest = version_manifest( + manifest_worker("6.3.0", us_version="2.2.1", uk_version="2.132.6"), + manifest_worker("6.4.0rc1", us_version="2.2.1", uk_version="2.132.7"), + default_version="6.3.0", + ) + + assert version_map_from_manifest(manifest, "us").root["2.2.1"] == ( + "policyengine-simulation-v2-py6-3-0" + ) + + +def test_latest_follows_the_default_version_rather_than_the_newest(): + manifest = version_manifest( + manifest_worker("6.2.6", us_version="1.700.0", uk_version="2.125.1"), + manifest_worker("6.2.7", us_version="1.701.0", uk_version="2.126.8"), + default_version="6.2.6", + ) + + answer = versions_from_manifest(manifest) + + assert answer.policyengine.root["latest"] == "6.2.6" + assert answer.us.root["latest"] == "1.700.0" + assert answer.uk.root["latest"] == "2.125.1" + + +def test_bundles_report_each_countrys_identity(): + release = with_country_bundle( + manifest_worker("6.2.7", us_version="1.700.0", uk_version="2.126.8"), + "uk", + data_package_version="0.1.0", + data_release_version="uk-release-1", + data_artifact_revision="uk-release-1", + default_dataset="frs_2024", + default_dataset_uri="hf://policyengine/data/frs_2024.h5@uk-release-1", + datasets=[ + { + "identity": "frs_2024", + "uri": "hf://policyengine/data/frs_2024.h5@uk-release-1", + "artifact_revision": "uk-release-1", + "sha256": TEST_DIGEST, + "repo_type": "dataset", + }, + { + "identity": "other_dataset", + "uri": "hf://policyengine/data/other_dataset.h5@uk-release-1", + "artifact_revision": "uk-release-1", + "sha256": "d" * 64, + "repo_type": "dataset", + }, + ], + ) + + answer = versions_from_manifest(version_manifest(release, default_version="6.2.7")) + + assert answer.bundles["6.2.7"]["uk"].model_dump(mode="json") == { + "model_version": "2.126.8", + "policyengine_version": "6.2.7", + "data_version": "uk-release-1", + "dataset": "hf://policyengine/data/frs_2024.h5@uk-release-1", + "dataset_sha256": TEST_DIGEST, + "spm": None, + } + assert set(answer.bundles["6.2.7"]) == {"us", "uk"} + + +def test_a_default_dataset_without_a_recorded_digest_reports_none(): + release = with_country_bundle( + manifest_worker("6.2.7", us_version="1.700.0", uk_version="2.126.8"), + "us", + default_dataset="dataset_the_bundle_lists_no_digest_for", + ) + + answer = versions_from_manifest(version_manifest(release, default_version="6.2.7")) + + assert answer.bundles["6.2.7"]["us"].dataset_sha256 is None + + +def test_a_dataset_that_cannot_be_resolved_is_reported_without_one(): + # The old gateway reports such an entry without a dataset rather than + # failing the whole answer. + release = with_country_bundle( + manifest_worker("6.2.7", us_version="1.700.0", uk_version="2.126.8"), + "us", + default_dataset_uri="hf://policyengine/missing-path", + ) + + answer = versions_from_manifest(version_manifest(release, default_version="6.2.7")) + + assert answer.bundles["6.2.7"]["us"].dataset is None + assert answer.bundles["6.2.7"]["us"].model_version == "1.700.0" + + +def test_the_manifest_records_no_spm_capabilities(): + # Known difference from the old gateway, which reports each bundle's SPM + # capability from v1's routing state. The v2 manifest does not record one. + assert versions_from_manifest(TWO_RELEASES).spm_capabilities == {} + + +@pytest.mark.parametrize("kind", ["US", "Uk", "POLICYENGINE"]) +def test_kind_is_matched_without_regard_to_case(kind): + assert version_map_from_manifest(TWO_RELEASES, kind) == version_map_from_manifest( + TWO_RELEASES, kind.lower() + ) + + +@pytest.mark.parametrize("kind", ["ca", "latest", ""]) +def test_an_unknown_kind_is_rejected_with_the_old_gateways_detail(kind): + with pytest.raises(UnknownVersionKindError) as raised: + version_map_from_manifest(TWO_RELEASES, kind) + + assert raised.value.detail == f"Unknown version kind: {kind}" + + +def test_each_kind_answer_equals_that_kinds_map_in_the_whole_answer(): + whole_answer = versions_from_manifest(TWO_RELEASES) + + for kind in ("policyengine", "us", "uk"): + assert version_map_from_manifest(TWO_RELEASES, kind) == getattr( + whole_answer, kind + ) + + +def _json_shape(value: object) -> object: + """Describe a JSON value by its keys and value types, ignoring the values.""" + if isinstance(value, dict): + return {key: _json_shape(child) for key, child in value.items()} + return type(value).__name__ + + +def test_the_answer_has_the_shape_of_the_gateways_answer(): + # The same two releases as the gateway fixture: 6.2.6 and 6.2.7, latest + # 6.2.7, with the same model versions. + answer = versions_from_manifest(TWO_RELEASES).model_dump(mode="json") + gateway_answer = GATEWAY_VERSIONS_ANSWER + + assert set(answer) == set(gateway_answer) + for kind in ("policyengine", "us", "uk"): + assert set(answer[kind]) == set(gateway_answer[kind]) + assert all(isinstance(value, str) for value in answer[kind].values()) + assert answer[kind]["latest"] == gateway_answer[kind]["latest"] + assert set(answer["bundles"]) == set(gateway_answer["bundles"]) + for version, countries in gateway_answer["bundles"].items(): + assert set(answer["bundles"][version]) == set(countries) + for country, identity in countries.items(): + built_identity = answer["bundles"][version][country] + assert set(built_identity) == set(identity) + assert built_identity["model_version"] == identity["model_version"] + assert ( + built_identity["policyengine_version"] + == identity["policyengine_version"] + ) + assert _json_shape(built_identity["spm"]) == _json_shape(identity["spm"]) + # Both answers read back as the same response model, unchanged. + for whole_answer in (answer, gateway_answer): + assert ( + VersionsResponse.model_validate(whole_answer).model_dump(mode="json") + == whole_answer + ) diff --git a/projects/policyengine-simulation-entry/tests/test_v2_versions_routes.py b/projects/policyengine-simulation-entry/tests/test_v2_versions_routes.py new file mode 100644 index 000000000..19a95517e --- /dev/null +++ b/projects/policyengine-simulation-entry/tests/test_v2_versions_routes.py @@ -0,0 +1,286 @@ +"""The internal routes that answer ``/versions`` from the v2 version manifest. + +``GET /internal/v2/versions`` and ``GET /internal/v2/versions/{kind}`` give +the answers ``/versions`` would give from the v2 version manifest, so a +staging deploy can compare them with the old gateway's answers while +``/versions`` still forwards to it. +""" + +from __future__ import annotations + +import asyncio +import threading + +import pytest +from conftest import FakeBackend, make_settings +from fastapi import HTTPException +from fastapi.testclient import TestClient +from policyengine_simulation_contract.stage12_manifest import ( + V2_VERSION_MANIFEST_NAME, + publish_v2_manifest, +) +from policyengine_simulation_entry import app as app_module +from policyengine_simulation_entry import v2_versions +from policyengine_simulation_entry.app import create_app +from policyengine_simulation_entry.v2_versions import ( + ModalV2VersionsSource, + version_map_from_manifest, + versions_from_manifest, +) +from v2_versions_support import TWO_RELEASES, manifest_worker + +UNAVAILABLE_DETAIL = "The v2 version manifest is unavailable." + + +class FakeVersionsSource: + """A v2 versions source that returns one manifest, or raises ``error``.""" + + def __init__(self, manifest=TWO_RELEASES, error: BaseException | None = None): + self.manifest = manifest + self.error = error + self.loads = 0 + + async def load_manifest(self): + self.loads += 1 + if self.error is not None: + raise self.error + return self.manifest + + +def versions_client(source, *, auth_dependency=lambda: None): + """Return a test client whose app reads the v2 manifest from ``source``.""" + backend = FakeBackend() + app = create_app( + settings=make_settings(), + backend=backend, + auth_dependency=auth_dependency, + v2_versions_source=source, + ) + return TestClient(app), backend + + +def test_the_whole_answer_comes_from_the_manifest(): + source = FakeVersionsSource() + client, backend = versions_client(source) + + with client: + result = client.get("/internal/v2/versions") + + assert result.status_code == 200 + assert result.json() == versions_from_manifest(TWO_RELEASES).model_dump(mode="json") + # Fields the answer leaves empty are sent as null, as the gateway sends them. + assert result.json()["bundles"]["6.2.7"]["us"]["spm"] is None + assert backend.requests == [] + + +@pytest.mark.parametrize( + ("kind", "expected_kind"), + [("policyengine", "policyengine"), ("us", "us"), ("UK", "uk")], +) +def test_each_kind_answer_comes_from_the_manifest(kind, expected_kind): + client, backend = versions_client(FakeVersionsSource()) + + with client: + result = client.get(f"/internal/v2/versions/{kind}") + + assert result.status_code == 200 + assert result.json() == version_map_from_manifest( + TWO_RELEASES, expected_kind + ).model_dump(mode="json") + assert backend.requests == [] + + +def test_an_unknown_kind_is_404_with_the_gateways_detail_before_any_read(): + source = FakeVersionsSource() + client, _ = versions_client(source) + + with client: + result = client.get("/internal/v2/versions/ca") + + assert result.status_code == 404 + assert result.json() == {"detail": "Unknown version kind: ca"} + assert source.loads == 0 + + +@pytest.mark.parametrize( + "error", + [ + ValueError("no valid v2 version manifest is available"), + RuntimeError("modal unavailable"), + TimeoutError(), + ], +) +@pytest.mark.parametrize("path", ["/internal/v2/versions", "/internal/v2/versions/us"]) +def test_an_unreadable_manifest_is_503(error, path): + client, _ = versions_client(FakeVersionsSource(error=error)) + + with client: + result = client.get(path) + + assert result.status_code == 503 + assert result.json() == {"detail": UNAVAILABLE_DETAIL} + assert result.headers["retry-after"] == "10" + + +@pytest.mark.parametrize("path", ["/internal/v2/versions", "/internal/v2/versions/uk"]) +def test_without_a_configured_manifest_the_routes_are_503(path): + client, _ = versions_client(None) + + with client: + result = client.get(path) + + assert result.status_code == 503 + assert result.json() == {"detail": UNAVAILABLE_DETAIL} + + +def test_a_manifest_read_that_does_not_finish_is_503(monkeypatch): + class StalledVersionsSource(FakeVersionsSource): + async def load_manifest(self): + await asyncio.Event().wait() + + monkeypatch.setattr(app_module, "V2_MANIFEST_READ_TIMEOUT_SECONDS", 0.001) + client, _ = versions_client(StalledVersionsSource()) + + with client: + result = client.get("/internal/v2/versions") + + assert result.status_code == 503 + assert result.json() == {"detail": UNAVAILABLE_DETAIL} + + +def test_the_routes_are_public_like_versions(): + def reject_every_caller(): + raise HTTPException(status_code=403, detail="Not authenticated") + + client, _ = versions_client( + FakeVersionsSource(), auth_dependency=reject_every_caller + ) + + with client: + whole = client.get("/internal/v2/versions") + kind = client.get("/internal/v2/versions/us") + + assert whole.status_code == 200 + assert kind.status_code == 200 + + +def test_the_routes_are_not_in_the_openapi_document(): + client, _ = versions_client(FakeVersionsSource()) + + with client: + paths = client.get("/openapi.json").json()["paths"] + + assert not [path for path in paths if path.startswith("/internal/v2")] + + +def test_the_modal_source_reads_the_published_manifest(): + store: dict[str, object] = {} + worker_version = manifest_worker("6.2.7", us_version="1.700.0", uk_version="2.1.0") + published = publish_v2_manifest(store=store, worker=worker_version) + + manifest = asyncio.run(ModalV2VersionsSource(store).load_manifest()) + + assert manifest == published + + +def test_the_modal_source_raises_when_no_manifest_was_published(): + with pytest.raises(ValueError, match="no valid v2 version manifest"): + asyncio.run(ModalV2VersionsSource({}).load_manifest()) + + +def test_the_modal_source_opens_the_configured_manifest_dict(monkeypatch): + opened = [] + + def open_dict(name, *, environment_name): + opened.append((name, environment_name)) + return {} + + monkeypatch.setattr(v2_versions.modal.Dict, "from_name", open_dict) + + source = ModalV2VersionsSource.from_settings( + make_settings(stage12_v2_manifest_environment="staging") + ) + + assert isinstance(source, ModalV2VersionsSource) + assert opened == [(V2_VERSION_MANIFEST_NAME, "staging")] + + +def test_no_modal_source_without_a_manifest_environment(monkeypatch): + def refuse_to_open(*_args, **_kwargs): + msg = "no manifest environment is configured" + raise AssertionError(msg) + + monkeypatch.setattr(v2_versions.modal.Dict, "from_name", refuse_to_open) + + assert ModalV2VersionsSource.from_settings(make_settings()) is None + + +def test_the_app_builds_the_modal_source_from_its_settings(monkeypatch): + opened = [] + + def open_dict(name, *, environment_name): + opened.append((name, environment_name)) + return {} + + monkeypatch.setattr(v2_versions.modal.Dict, "from_name", open_dict) + + # The deploy sets the manifest environment together with the other Stage + # 12 resources, which settings validation requires. + app = create_app( + settings=make_settings( + stage12_v2_manifest_environment="main", + stage12_database_url=( + "postgresql://policyengine_v2_runtime:secret@" + "test.pooler.supabase.com:5432/postgres?sslmode=require" + ), + stage12_artifact_bucket="policyengine-stage12-production", + ), + backend=FakeBackend(), + auth_dependency=lambda: None, + # A stand-in, so only the version source opens a manifest Dict here. + comparison_backend=object(), + ) + with TestClient(app) as client: + result = client.get("/internal/v2/versions") + + assert opened == [(V2_VERSION_MANIFEST_NAME, "main")] + # The opened store is empty, so no manifest has been published to it. + assert result.status_code == 503 + + +def test_the_modal_source_reads_on_its_own_thread(): + # A stalled Modal read must not hold threads from the event loop's default + # executor, which the Stage 12 comparison path uses. + reading_threads = [] + + class RecordingStore(dict): + def get(self, key, default=None): + reading_threads.append(threading.current_thread().name) + return super().get(key, default) + + store = RecordingStore() + publish_v2_manifest( + store=store, + worker=manifest_worker("6.2.7", us_version="1.700.0", uk_version="2.1.0"), + ) + reading_threads.clear() + + asyncio.run(ModalV2VersionsSource(store).load_manifest()) + + assert reading_threads + assert all(name.startswith("v2-version-manifest") for name in reading_threads) + + +def test_an_injected_source_is_used_even_when_it_is_falsy(): + class EmptySizedVersionsSource(FakeVersionsSource): + def __len__(self): + return 0 + + source = EmptySizedVersionsSource() + client, _ = versions_client(source) + + with client: + result = client.get("/internal/v2/versions") + + assert result.status_code == 200 + assert source.loads == 1 diff --git a/projects/policyengine-simulation-entry/tests/v2_versions_support.py b/projects/policyengine-simulation-entry/tests/v2_versions_support.py new file mode 100644 index 000000000..dce40065c --- /dev/null +++ b/projects/policyengine-simulation-entry/tests/v2_versions_support.py @@ -0,0 +1,75 @@ +"""v2 version manifests for the tests of the manifest's ``/versions`` answers. + +Shared by the tests of the answers themselves and of the internal routes that +serve them. +""" + +from __future__ import annotations + +from policyengine_simulation_contract.stage12_manifest import ( + V2VersionManifest, + V2WorkerVersion, + v2_application_name, +) +from stage12_fixtures import worker + + +def manifest_worker( + policyengine_version: str, + *, + us_version: str, + uk_version: str, +) -> V2WorkerVersion: + """Return the fixture worker, rebuilt for these package versions.""" + worker_document = worker().model_dump(mode="json") + application_name = v2_application_name(policyengine_version) + worker_document["application_name"] = application_name + worker_document["validation"]["application_name"] = application_name + worker_document["bundle"]["policyengine_version"] = policyengine_version + worker_document["bundle"]["policyengine_requirement"] = ( + f"policyengine=={policyengine_version}" + ) + for country_bundle in worker_document["bundle"]["countries"]: + package_version = {"us": us_version, "uk": uk_version}[ + country_bundle["country"] + ] + country_bundle["country_package_version"] = package_version + country_bundle["country_package_requirement"] = ( + f"{country_bundle['country_package_name']}=={package_version}" + ) + return V2WorkerVersion.model_validate(worker_document) + + +def version_manifest( + *workers: V2WorkerVersion, + default_version: str, +) -> V2VersionManifest: + """Return a manifest holding ``workers``, keyed by their bundle versions.""" + return V2VersionManifest( + generation="test-generation", + default_version=default_version, + versions={ + worker_version.bundle.policyengine_version: worker_version + for worker_version in workers + }, + ) + + +def with_country_bundle( + worker_version: V2WorkerVersion, + country: str, + **fields: object, +) -> V2WorkerVersion: + """Return ``worker_version`` with ``fields`` replaced in one country's bundle.""" + worker_document = worker_version.model_dump(mode="json") + for country_bundle in worker_document["bundle"]["countries"]: + if country_bundle["country"] == country: + country_bundle.update(fields) + return V2WorkerVersion.model_validate(worker_document) + + +TWO_RELEASES = version_manifest( + manifest_worker("6.2.6", us_version="1.700.0", uk_version="2.125.1"), + manifest_worker("6.2.7", us_version="1.700.0", uk_version="2.126.8"), + default_version="6.2.7", +)