tegwick
ab936a1e98
Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a0230c-b06c-7641-808a-e191b6d1da49
271 lines
9.9 KiB
Python
271 lines
9.9 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
from copy import deepcopy
|
|
from typing import Any
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from jsonschema import Draft202012Validator, FormatChecker
|
|
from sqlalchemy.ext.asyncio import create_async_engine
|
|
|
|
from hub_core.contracts import workload_projection_contract_root
|
|
from hub_core.runtime.app import create_app
|
|
from hub_core.runtime.config import RuntimeSettings
|
|
from hub_core.runtime.postgres_store import PostgresPortStore
|
|
from hub_core.runtime.store import InMemoryPortStore
|
|
from hub_core.runtime.tables import (
|
|
runtime_metadata,
|
|
runtime_workload_projection_records,
|
|
runtime_workload_projection_state,
|
|
)
|
|
from hub_core.runtime.workload_projection import (
|
|
WorkloadCursorMismatch,
|
|
WorkloadProjectionRejected,
|
|
WorkloadProjectionService,
|
|
)
|
|
|
|
|
|
def fixture() -> dict[str, Any]:
|
|
resource = workload_projection_contract_root().joinpath(
|
|
"fixtures", "repo-manager-nine-workloads.json"
|
|
)
|
|
return json.loads(resource.read_text(encoding="utf-8"))
|
|
|
|
|
|
class PageClient:
|
|
def __init__(self, pages: list[dict[str, Any]]) -> None:
|
|
self.pages = pages
|
|
self.calls: list[str | None] = []
|
|
|
|
async def fetch_workload_page(self, cursor: str | None) -> dict[str, Any]:
|
|
self.calls.append(cursor)
|
|
for page in self.pages:
|
|
if page["snapshot"]["page_cursor"] == cursor:
|
|
return deepcopy(page)
|
|
raise RuntimeError(f"unexpected cursor {cursor!r}")
|
|
|
|
|
|
class FailingClient:
|
|
async def fetch_workload_page(self, cursor: str | None) -> dict[str, Any]:
|
|
raise RuntimeError("repo-manager workload index unavailable")
|
|
|
|
|
|
def next_snapshot(
|
|
page: dict[str, Any], *, marker: str, generated_at: str
|
|
) -> dict[str, Any]:
|
|
changed = deepcopy(page)
|
|
changed["snapshot"].update(
|
|
{"snapshot_id": marker * 64, "generated_at": generated_at}
|
|
)
|
|
return changed
|
|
|
|
|
|
def test_ingests_nine_records_and_resolves_only_exact_reference() -> None:
|
|
async def run() -> None:
|
|
store = InMemoryPortStore()
|
|
service = WorkloadProjectionService(client=PageClient([fixture()]), store=store)
|
|
|
|
result = await service.refresh()
|
|
resolved = await service.resolve(
|
|
rapp_id="rapp-qonto", name="qonto", deployable="rapp-qonto"
|
|
)
|
|
wrong_name = await service.resolve(
|
|
rapp_id="rapp-qonto", name="qonto-assistant"
|
|
)
|
|
wrong_deployable = await service.resolve(
|
|
rapp_id="rapp-openbao", name="openbao", deployable="rapp-openbao"
|
|
)
|
|
|
|
assert result.workload_count == 9
|
|
assert resolved is not None and resolved["status"] == "resolved"
|
|
assert wrong_name is not None and wrong_name["reason"] == "not_found"
|
|
assert wrong_deployable is not None
|
|
assert wrong_deployable["reason"] == "deployable_not_declared"
|
|
assert await service.readiness_checks() == {"workload_projection": "ok"}
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_deletion_and_full_rebuild_restore_equivalent_content() -> None:
|
|
async def run() -> None:
|
|
source = fixture()
|
|
store = InMemoryPortStore()
|
|
initial = WorkloadProjectionService(client=PageClient([source]), store=store)
|
|
original = await initial.refresh()
|
|
|
|
reduced = next_snapshot(
|
|
source, marker="a", generated_at="2026-08-22T12:05:00Z"
|
|
)
|
|
reduced["workloads"] = reduced["workloads"][:-1]
|
|
reduced["snapshot"]["total_workload_count"] = 8
|
|
await WorkloadProjectionService(
|
|
client=PageClient([reduced]), store=store
|
|
).refresh()
|
|
after_delete = await store.get_workload_projection()
|
|
|
|
restored = next_snapshot(
|
|
source, marker="b", generated_at="2026-08-22T12:10:00Z"
|
|
)
|
|
rebuilt = await WorkloadProjectionService(
|
|
client=PageClient([restored]), store=store
|
|
).refresh()
|
|
|
|
assert after_delete is not None and len(after_delete.workloads) == 8
|
|
assert rebuilt.workload_count == 9
|
|
assert rebuilt.content_hash == original.content_hash
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_invalid_non_utc_input_preserves_last_generation_as_stale() -> None:
|
|
async def run() -> None:
|
|
store = InMemoryPortStore()
|
|
await WorkloadProjectionService(
|
|
client=PageClient([fixture()]), store=store
|
|
).refresh()
|
|
before = await store.get_workload_projection()
|
|
invalid = next_snapshot(
|
|
fixture(), marker="c", generated_at="2026-08-22T14:00:00+02:00"
|
|
)
|
|
|
|
with pytest.raises(WorkloadProjectionRejected, match="invalid workload page"):
|
|
await WorkloadProjectionService(
|
|
client=PageClient([invalid]), store=store
|
|
).refresh()
|
|
after = await store.get_workload_projection()
|
|
|
|
assert before is not None and after is not None
|
|
assert after.content_hash == before.content_hash
|
|
assert after.workloads == before.workloads
|
|
assert after.projection_status == "stale"
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_query_order_cursor_and_generation_binding() -> None:
|
|
async def run() -> None:
|
|
store = InMemoryPortStore()
|
|
service = WorkloadProjectionService(client=PageClient([fixture()]), store=store)
|
|
await service.refresh()
|
|
first = await service.query(limit=4)
|
|
|
|
assert first is not None and first["next_cursor"]
|
|
assert [item["rapp_id"] for item in first["workloads"]] == sorted(
|
|
item["rapp_id"] for item in first["workloads"]
|
|
)
|
|
second = await service.query(cursor=first["next_cursor"], limit=4)
|
|
assert second is not None and len(second["workloads"]) == 4
|
|
|
|
changed = next_snapshot(
|
|
fixture(), marker="d", generated_at="2026-08-22T12:15:00Z"
|
|
)
|
|
await WorkloadProjectionService(
|
|
client=PageClient([changed]), store=store
|
|
).refresh()
|
|
with pytest.raises(WorkloadCursorMismatch, match="cursor_snapshot_mismatch"):
|
|
await service.query(cursor=first["next_cursor"], limit=4)
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_outage_without_generation_fails_readiness_closed() -> None:
|
|
async def run() -> None:
|
|
service = WorkloadProjectionService(
|
|
client=FailingClient(), store=InMemoryPortStore()
|
|
)
|
|
with pytest.raises(WorkloadProjectionRejected, match="unavailable"):
|
|
await service.refresh()
|
|
assert await service.readiness_checks() == {"workload_projection": "unavailable"}
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_http_and_mcp_backing_routes_are_read_only_and_schema_valid() -> None:
|
|
settings = RuntimeSettings(environment="test", backend="memory", allow_ephemeral=True)
|
|
app = create_app(
|
|
settings=settings,
|
|
port_store=InMemoryPortStore(),
|
|
workload_projection_client=PageClient([fixture()]),
|
|
)
|
|
with TestClient(app) as runtime:
|
|
response = runtime.get(
|
|
"/ports/projections/workloads", params={"deployable": "openbao"}
|
|
)
|
|
resolved = runtime.get(
|
|
"/ports/projections/workloads/resolve",
|
|
params={"rapp_id": "rapp-openbao", "name": "openbao"},
|
|
)
|
|
unknown = runtime.get(
|
|
"/ports/projections/workloads/resolve",
|
|
params={"rapp_id": "rapp-openbao", "name": "railiance-platform"},
|
|
)
|
|
ready = runtime.get("/readyz")
|
|
openapi = runtime.get("/openapi.json").json()
|
|
|
|
schema_root = workload_projection_contract_root().joinpath("schemas")
|
|
projection_schema = json.loads(
|
|
schema_root.joinpath("workload-projection.schema.json").read_text(encoding="utf-8")
|
|
)
|
|
resolution_schema = json.loads(
|
|
schema_root.joinpath("workload-resolution.schema.json").read_text(encoding="utf-8")
|
|
)
|
|
Draft202012Validator(projection_schema, format_checker=FormatChecker()).validate(
|
|
response.json()
|
|
)
|
|
Draft202012Validator(resolution_schema, format_checker=FormatChecker()).validate(
|
|
resolved.json()
|
|
)
|
|
Draft202012Validator(resolution_schema, format_checker=FormatChecker()).validate(
|
|
unknown.json()
|
|
)
|
|
assert response.status_code == resolved.status_code == unknown.status_code == 200
|
|
assert [record["rapp_id"] for record in response.json()["workloads"]] == [
|
|
"rapp-openbao"
|
|
]
|
|
assert resolved.json()["status"] == "resolved"
|
|
assert unknown.json()["status"] == "unknown"
|
|
assert ready.json()["checks"]["workload_projection"] == "ok"
|
|
paths = {
|
|
path: item
|
|
for path, item in openapi["paths"].items()
|
|
if path.startswith("/ports/projections/workloads")
|
|
}
|
|
assert paths and all(set(item) == {"get"} for item in paths.values())
|
|
|
|
|
|
def test_postgres_workload_projection_survives_reopen(tmp_path) -> None:
|
|
database_url = f"sqlite+aiosqlite:///{tmp_path / 'workloads.db'}"
|
|
|
|
async def run() -> None:
|
|
engine = create_async_engine(database_url)
|
|
async with engine.begin() as connection:
|
|
await connection.run_sync(runtime_metadata.create_all)
|
|
await engine.dispose()
|
|
first_store = PostgresPortStore.from_url(database_url)
|
|
result = await WorkloadProjectionService(
|
|
client=PageClient([fixture()]), store=first_store
|
|
).refresh()
|
|
await first_store.aclose()
|
|
second_store = PostgresPortStore.from_url(database_url)
|
|
projection = await second_store.get_workload_projection()
|
|
await second_store.aclose()
|
|
|
|
assert projection is not None
|
|
assert projection.content_hash == result.content_hash
|
|
assert len(projection.workloads) == 9
|
|
assert all(item["observed_at"].endswith("Z") for item in projection.workloads)
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
def test_projection_tables_and_module_have_no_private_repo_manager_coupling() -> None:
|
|
source = (
|
|
__import__("inspect")
|
|
.getsource(__import__("hub_core.runtime.workload_projection", fromlist=["*"]))
|
|
)
|
|
|
|
assert "repo_manager." not in source
|
|
assert not runtime_workload_projection_state.foreign_keys
|
|
assert not runtime_workload_projection_records.foreign_keys
|