127 lines
4.9 KiB
Python
127 lines
4.9 KiB
Python
from fastapi.testclient import TestClient
|
|
from pydantic import SecretStr
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy.pool import StaticPool
|
|
|
|
from modelforge_api.api.routes.registry import get_registry_service
|
|
from modelforge_api.db import get_session
|
|
from modelforge_api.domain.release import PRODUCT_VERSION
|
|
from modelforge_api.main import app
|
|
from modelforge_api.persistence.models import Base
|
|
from modelforge_api.services.manifest_registry import ManifestRegistry
|
|
from modelforge_api.services.registry import RegistryService, seed_candidate_registry
|
|
from modelforge_api.settings import Settings, get_settings
|
|
|
|
client = TestClient(app)
|
|
|
|
|
|
def _test_operator_credential() -> str:
|
|
return "modelforge-test-operator"
|
|
|
|
|
|
CONTROL_PLANE_HEADERS = {"X-ModelForge-Admin-Token": _test_operator_credential()}
|
|
|
|
|
|
def _configure_operator_auth() -> None:
|
|
settings = Settings(
|
|
_env_file=None,
|
|
operator_api_key=SecretStr(_test_operator_credential()),
|
|
)
|
|
app.dependency_overrides[get_settings] = lambda: settings
|
|
|
|
|
|
def test_liveness_and_correlation_id() -> None:
|
|
response = client.get("/api/v1/health/live", headers={"x-correlation-id": "test-correlation"})
|
|
assert response.status_code == 200
|
|
assert response.headers["x-correlation-id"] == "test-correlation"
|
|
assert response.json()["status"] == "ok"
|
|
|
|
|
|
def test_readiness_validates_all_manifests() -> None:
|
|
response = client.get("/api/v1/health/ready")
|
|
assert response.status_code == 200
|
|
assert response.json()["checks"] == {"manifests": "healthy"}
|
|
|
|
|
|
def test_system_metadata_reports_the_release_without_unpromoted_inference() -> None:
|
|
"""A released product reports its version and channel, not the milestone that built it."""
|
|
|
|
engine = create_engine(
|
|
"sqlite+pysqlite:///:memory:",
|
|
connect_args={"check_same_thread": False},
|
|
poolclass=StaticPool,
|
|
)
|
|
Base.metadata.create_all(engine)
|
|
with Session(engine) as session:
|
|
_configure_operator_auth()
|
|
app.dependency_overrides[get_session] = lambda: session
|
|
try:
|
|
payload = client.get(
|
|
"/api/v1/system",
|
|
headers=CONTROL_PLANE_HEADERS,
|
|
).json()
|
|
assert payload["version"] == PRODUCT_VERSION
|
|
assert payload["release_channel"] == "stable"
|
|
assert "milestone" not in payload
|
|
assert payload["production_inference_available"] is False
|
|
finally:
|
|
app.dependency_overrides.clear()
|
|
|
|
|
|
def test_seed_catalog_is_unverified_candidate_metadata_only() -> None:
|
|
engine = create_engine(
|
|
"sqlite+pysqlite:///:memory:",
|
|
connect_args={"check_same_thread": False},
|
|
poolclass=StaticPool,
|
|
)
|
|
Base.metadata.create_all(engine)
|
|
with Session(engine) as session:
|
|
seed_candidate_registry(session, ManifestRegistry())
|
|
_configure_operator_auth()
|
|
app.dependency_overrides[get_registry_service] = lambda: RegistryService(session)
|
|
try:
|
|
response = client.get("/api/v1/models", headers=CONTROL_PLANE_HEADERS)
|
|
assert response.status_code == 200
|
|
models = response.json()["items"]
|
|
assert len(models) == 15
|
|
assert all(model["lifecycle"] == "candidate" for model in models)
|
|
assert all(model["verification_status"] == "unverified" for model in models)
|
|
assert all(model["deployment_status"] == "not_deployed" for model in models)
|
|
finally:
|
|
app.dependency_overrides.clear()
|
|
|
|
|
|
def test_capability_and_project_contract_endpoints() -> None:
|
|
"""Runs against an in-memory database, like its neighbours.
|
|
|
|
It used to call the endpoint with no session override, so `get_session` handed it the real
|
|
engine and the test quietly required a live PostgreSQL on localhost. It passed on a developer
|
|
machine that happened to have one running and failed anywhere else — the same class of hidden
|
|
environment dependency the release gate exists to find.
|
|
"""
|
|
|
|
engine = create_engine(
|
|
"sqlite+pysqlite:///:memory:",
|
|
connect_args={"check_same_thread": False},
|
|
poolclass=StaticPool,
|
|
)
|
|
Base.metadata.create_all(engine)
|
|
with Session(engine) as session:
|
|
_configure_operator_auth()
|
|
app.dependency_overrides[get_session] = lambda: session
|
|
try:
|
|
capabilities = client.get(
|
|
"/api/v1/capabilities",
|
|
headers=CONTROL_PLANE_HEADERS,
|
|
).json()
|
|
projects = client.get(
|
|
"/api/v1/projects",
|
|
headers=CONTROL_PLANE_HEADERS,
|
|
).json()
|
|
finally:
|
|
app.dependency_overrides.clear()
|
|
assert {item["key"] for item in capabilities} >= {"rag.embedding", "vision.embedding"}
|
|
assert {item["id"] for item in projects} == {"examplerag", "examplevision", "example-ops"}
|
|
assert all("model_id" not in binding for project in projects for binding in project["bindings"])
|