fix: serve legacy route aliases during cutover
Some checks failed
CI Smoke / host-smoke (push) Successful in 0s
CI Smoke / pytest-smoke (push) Failing after 2s

This commit is contained in:
tegwick 2026-08-21 17:45:25 +02:00
parent 8ab1d0c09a
commit 14befd0b50
4 changed files with 57 additions and 2 deletions

View file

@ -9,6 +9,7 @@ from uuid import uuid4
import sqlalchemy as sa
from fastapi import APIRouter, Header, HTTPException, Request, status
from fastapi.routing import APIRoute
from fastapi.responses import HTMLResponse, PlainTextResponse, RedirectResponse
from hub_core.runtime.compat_catalogs import (
@ -504,6 +505,25 @@ def create_compatibility_router() -> APIRouter:
"<p>Compatibility runtime candidate; projections are authoritative.</p>"
)
# Core Hub historically served the same compatibility operations both below
# /api/v2 and at the root. Keep the aliases as real routes (not merely
# OpenAPI aliases) so a route-group cutover cannot turn existing clients into
# 404s. The canonical /api/v2 routes remain the only schema-bearing routes;
# openapi_json adds the documented alias paths deterministically.
for route in list(router.routes):
if not isinstance(route, APIRoute) or not route.path.startswith("/api/v2/"):
continue
alias = route.path.removeprefix("/api/v2")
router.add_api_route(
alias,
route.endpoint,
methods=sorted(route.methods or []),
status_code=route.status_code,
response_class=route.response_class,
include_in_schema=False,
name=f"legacy_{route.name}",
)
return router

View file

@ -69,11 +69,24 @@ class RuntimeSettings:
def readiness_checks(self, store_backend: str) -> dict[str, str]:
ephemeral_allowed = store_backend != "memory" or self.allow_ephemeral
protected_groups = self.v2_groups & {
"registry",
"credentials",
"interaction",
"deferred",
"operator",
}
authorization_ready = (
not protected_groups
or bool(self.api_token)
or store_backend == "postgresql"
)
return {
"environment": self.environment,
"configured_backend": self.backend,
"active_backend": store_backend,
"ephemeral_backend": "allowed" if ephemeral_allowed else "not_allowed",
"authorization": "ok" if authorization_ready else "unavailable",
"contract": "helixforge.hub-extension/0.1.0",
"v2_groups": ",".join(sorted(self.v2_groups)) or "none",
"v2_write_groups": ",".join(sorted(self.v2_write_groups)) or "none",
@ -81,6 +94,9 @@ class RuntimeSettings:
}
def is_ready(self, store_backend: str) -> bool:
return self.backend == store_backend and (
store_backend != "memory" or self.allow_ephemeral
checks = self.readiness_checks(store_backend)
return (
self.backend == store_backend
and (store_backend != "memory" or self.allow_ephemeral)
and checks["authorization"] == "ok"
)

View file

@ -62,6 +62,10 @@ def test_system_compatibility_and_protected_auth(tmp_path) -> None:
document = client.get("/api/v2/openapi.json").json()
assert "/api/v2/hubs" in document["paths"]
assert "/hubs" in document["paths"]
assert client.get("/widget-types").status_code == 200
assert client.get("/hubs").status_code == 401
assert client.get("/hubs", headers=HEADERS).status_code == 200
assert client.get("/console", headers=HEADERS).status_code == 200
asyncio.run(store.aclose())

View file

@ -58,6 +58,21 @@ def test_production_readiness_fails_closed_for_ephemeral_backend() -> None:
assert response.json()["checks"]["ephemeral_backend"] == "not_allowed"
def test_protected_compatibility_readiness_requires_authorization_dependency() -> None:
settings = RuntimeSettings(
environment="test",
backend="memory",
allow_ephemeral=True,
v2_groups=frozenset({"registry"}),
)
response = TestClient(
create_app(settings=settings, port_store=InMemoryPortStore())
).get("/readyz")
assert response.status_code == 503
assert response.json()["checks"]["authorization"] == "unavailable"
def test_registry_validates_and_registers_idempotently() -> None:
runtime = client()
package = ops_hub_package()