"""The v1 release contract, enforced. A release that misdescribes itself is a release nobody can support. These tests hold the version to one source of truth, hold the compatibility ranges to explicit answers, and hold production configuration to rules a development environment is allowed to break. """ from __future__ import annotations import json import re import tomllib from pathlib import Path import pytest from pydantic import SecretStr from modelforge_api import __version__ from modelforge_api.domain.release import ( CURRENT_AGENT_PROTOCOL_VERSION, MINIMUM_POSTGRES_MAJOR, MINIMUM_UPGRADE_SOURCE, PRODUCT_VERSION, RELEASE_CHANNEL, SUPPORTED_AGENT_PROTOCOL_VERSIONS, SUPPORTED_SCHEMA_REVISIONS, SUPPORTED_UPGRADE_SOURCE_SCHEMA_REVISIONS, TARGET_SCHEMA_REVISION, Compatibility, SemanticVersion, agent_protocol_compatibility, build_identity, schema_compatibility, upgrade_required, ) from modelforge_api.services.startup_validation import ( StartupFailureCode, StartupValidationError, enforce_startup, validate_settings, ) from modelforge_api.settings import Settings ROOT = Path(__file__).resolve().parents[2] # --------------------------------------------------------------------------- one version def test_the_version_file_is_the_single_source_of_truth() -> None: declared = (ROOT / "VERSION").read_text("utf-8").strip() assert declared == PRODUCT_VERSION assert declared == __version__ @pytest.mark.parametrize( "manifest", ["backend/pyproject.toml", "node-agent/pyproject.toml", "runtime-worker/pyproject.toml"], ) def test_every_python_manifest_matches_the_version_file(manifest: str) -> None: """Four independent copies of the version is three chances to publish a wrong one.""" data = tomllib.loads((ROOT / manifest).read_text("utf-8")) assert data["project"]["version"] == PRODUCT_VERSION, ( f"{manifest} declares {data['project']['version']!r}, VERSION says {PRODUCT_VERSION!r}" ) def test_the_console_manifest_matches_the_version_file() -> None: data = json.loads((ROOT / "frontend" / "package.json").read_text("utf-8")) assert data["version"] == PRODUCT_VERSION @pytest.mark.parametrize( "module", [ "node-agent/src/modelforge_node_agent/__init__.py", "runtime-worker/src/modelforge_runtime_worker/__init__.py", ], ) def test_every_component_module_matches_the_version_file(module: str) -> None: text = (ROOT / module).read_text("utf-8") match = re.search(r'__version__ = "([^"]+)"', text) assert match is not None, f"{module} declares no __version__" assert match.group(1) == PRODUCT_VERSION def test_the_release_version_is_semantic_and_not_a_prerelease() -> None: parsed = SemanticVersion.parse(PRODUCT_VERSION) assert not parsed.is_prerelease assert parsed.major >= 1, "a v1 release cannot carry a 0.x major" @pytest.mark.parametrize("bad", ["1.0", "1.0.0.0", "v1.0.0", "one.0.0", ""]) def test_a_malformed_version_is_refused(bad: str) -> None: with pytest.raises(ValueError): SemanticVersion.parse(bad) # --------------------------------------------------------------------------- compatibility def test_the_target_schema_revision_is_supported() -> None: assert TARGET_SCHEMA_REVISION == "20260830_0024" assert TARGET_SCHEMA_REVISION in SUPPORTED_SCHEMA_REVISIONS assert schema_compatibility(TARGET_SCHEMA_REVISION) is Compatibility.COMPATIBLE assert "20260827_0021" in SUPPORTED_UPGRADE_SOURCE_SCHEMA_REVISIONS assert "20260828_0022" in SUPPORTED_UPGRADE_SOURCE_SCHEMA_REVISIONS assert "20260827_0021" not in SUPPORTED_SCHEMA_REVISIONS def test_the_rc_schema_has_one_explicit_direct_upgrade_contract() -> None: # The version bump is a later release tranche. This schema tranche must already describe every # accepted source honestly so an intermediate protected merge cannot migrate the wrong target. assert PRODUCT_VERSION == "1.2.2" assert MINIMUM_UPGRADE_SOURCE == "v1.0.0" assert SUPPORTED_UPGRADE_SOURCE_SCHEMA_REVISIONS == ( "20260827_0021", "20260828_0022", "20260830_0023", "20260830_0024", ) def test_v1_1_documents_the_destructive_downgrade_boundary_truthfully() -> None: migration = ( ROOT / "backend" / "alembic" / "versions" / "20260828_0022_node_decommission.py" ).read_text("utf-8") upgrade_runbook = (ROOT / "docs" / "UPGRADE.md").read_text("utf-8") upgrade_tool = (ROOT / "scripts" / "upgrade.py").read_text("utf-8") assert "def downgrade()" in migration assert "tombstone-free" in upgrade_runbook assert "not the production rollback promise" in upgrade_runbook assert "rehearsal-only" in upgrade_tool @pytest.mark.parametrize( ("revision", "expected"), [ ("20260830_0024", Compatibility.COMPATIBLE), ("20260830_0023", Compatibility.TOO_OLD), ("20260828_0022", Compatibility.TOO_OLD), ("20260827_0021", Compatibility.TOO_OLD), ("20260101_0001", Compatibility.TOO_OLD), ("20990101_0099", Compatibility.TOO_NEW), (None, Compatibility.UNKNOWN), ], ) def test_schema_compatibility_is_explicit(revision: str | None, expected: Compatibility) -> None: assert schema_compatibility(revision) is expected @pytest.mark.parametrize( ("protocol", "expected"), [ (1, Compatibility.COMPATIBLE), (0, Compatibility.TOO_OLD), (2, Compatibility.TOO_NEW), (None, Compatibility.UNKNOWN), ], ) def test_agent_protocol_compatibility_is_explicit( protocol: int | None, expected: Compatibility ) -> None: assert agent_protocol_compatibility(protocol) is expected def test_only_a_compatible_answer_requires_no_operator_action() -> None: assert upgrade_required(Compatibility.COMPATIBLE) is None for other in (Compatibility.TOO_OLD, Compatibility.TOO_NEW, Compatibility.UNKNOWN): message = upgrade_required(other) assert message and message.strip(), f"{other} must tell the operator what to do" def test_the_control_plane_speaks_a_protocol_it_supports() -> None: assert CURRENT_AGENT_PROTOCOL_VERSION in SUPPORTED_AGENT_PROTOCOL_VERSIONS def test_the_declared_protocol_matches_the_wire_constant() -> None: from modelforge_api.domain.agent_protocol import AGENT_PROTOCOL_VERSION assert AGENT_PROTOCOL_VERSION == CURRENT_AGENT_PROTOCOL_VERSION # --------------------------------------------------------------------------- build identity def test_build_identity_reports_absent_values_as_null_rather_than_guessing() -> None: identity = build_identity() assert identity.version == PRODUCT_VERSION assert identity.source_commit is None assert identity.built_at is None assert identity.image_digest is None assert identity.channel == RELEASE_CHANNEL assert identity.schema_revision == TARGET_SCHEMA_REVISION @pytest.mark.parametrize("placeholder", ["", " ", "unknown", "UNKNOWN", "none", "null"]) def test_an_unsubstituted_build_argument_is_not_reported_as_a_fact(placeholder: str) -> None: """A Dockerfile ARG that was never passed must not become a claimed commit.""" identity = build_identity(source_commit=placeholder, image_digest=placeholder) assert identity.source_commit is None assert identity.image_digest is None def test_build_identity_carries_a_real_commit_when_given_one() -> None: identity = build_identity(source_commit="657ad91", built_at="2026-08-27T12:00:00Z") assert identity.source_commit == "657ad91" assert identity.built_at == "2026-08-27T12:00:00Z" assert "source_commit" in identity.as_dict() # --------------------------------------------------------------------------- configuration def _production(**overrides: object) -> Settings: base: dict[str, object] = { "env": "production", "operator_api_key": SecretStr("k" * 48), "backup_encryption_key": SecretStr("x" * 44), "database_url": "postgresql+psycopg://modelforge_runtime:aVeryLongProductionSecret@db:5432/mf", "redis_url": "redis://cache:6379/0", "cors_origins": "https://console.example.test", "allow_remote_code": False, } base.update(overrides) return Settings(**base) # type: ignore[arg-type] def _codes(settings: Settings) -> set[str]: return {str(problem.code) for problem in validate_settings(settings).problems} def test_production_without_an_operator_key_is_refused() -> None: settings = _production(operator_api_key=None) assert StartupFailureCode.MISSING_REQUIRED_SETTING in _codes(settings) def test_production_with_a_short_operator_key_is_refused() -> None: assert StartupFailureCode.INSECURE_PRODUCTION_SETTING in _codes( _production(operator_api_key=SecretStr("short")) ) def test_production_with_a_development_database_password_is_refused() -> None: settings = _production( database_url="postgresql+psycopg://modelforge:modelforge@db:5432/modelforge" ) assert StartupFailureCode.INSECURE_PRODUCTION_SETTING in _codes(settings) def test_production_never_permits_remote_code_execution() -> None: assert StartupFailureCode.INSECURE_PRODUCTION_SETTING in _codes( _production(allow_remote_code=True) ) def test_production_without_a_backup_encryption_key_is_refused() -> None: assert StartupFailureCode.MISSING_REQUIRED_SETTING in _codes( _production(backup_encryption_key=None) ) def test_production_refuses_a_wildcard_cors_origin() -> None: assert StartupFailureCode.INSECURE_PRODUCTION_SETTING in _codes(_production(cors_origins="*")) @pytest.mark.parametrize( "url", ["", "not-a-url", "mysql://db/mf", "postgresql+psycopg:///mf"] ) def test_an_invalid_database_url_is_refused(url: str) -> None: assert StartupFailureCode.INVALID_URL in _codes(_production(database_url=url)) @pytest.mark.parametrize("url", ["", "http://cache:6379", "memcached://cache"]) def test_an_invalid_redis_url_is_refused(url: str) -> None: assert StartupFailureCode.INVALID_URL in _codes(_production(redis_url=url)) def test_a_missing_storage_root_is_refused(tmp_path: Path) -> None: settings = _production(artifact_root=str(tmp_path / "absent")) problems = validate_settings(settings).problems assert any( problem.code is StartupFailureCode.INVALID_STORAGE_ROOT and problem.setting == "MODELFORGE_ARTIFACT_ROOT" for problem in problems ) def test_a_storage_root_that_is_a_file_is_refused(tmp_path: Path) -> None: target = tmp_path / "not-a-directory" target.write_text("", encoding="utf-8") settings = _production(artifact_root=str(target)) assert StartupFailureCode.INVALID_STORAGE_ROOT in _codes(settings) def test_a_writable_storage_root_is_accepted(tmp_path: Path) -> None: settings = _production( artifact_root=str(tmp_path), quarantine_root=str(tmp_path), backup_root=tmp_path, ) roots = { problem.setting for problem in validate_settings(settings).problems if problem.code is StartupFailureCode.INVALID_STORAGE_ROOT } assert roots == set() @pytest.mark.parametrize( "overrides", [ {"scheduler_safety_reserve_percentage": 0.5}, {"node_stale_after_seconds": 90, "node_offline_after_seconds": 90}, {"gateway_queue_timeout_seconds": 45, "gateway_request_timeout_seconds": 45}, ], ) def test_an_impossible_policy_is_refused(overrides: dict[str, object]) -> None: assert StartupFailureCode.IMPOSSIBLE_POLICY in _codes(_production(**overrides)) def test_a_sound_production_configuration_has_no_problems(tmp_path: Path) -> None: settings = _production( artifact_root=str(tmp_path), quarantine_root=str(tmp_path), backup_root=tmp_path ) report = validate_settings(settings) assert report.ok, [str(problem) for problem in report.problems] def test_production_refuses_to_start_while_development_defaults_survive(tmp_path: Path) -> None: settings = _production( operator_api_key=None, artifact_root=str(tmp_path), quarantine_root=str(tmp_path), backup_root=tmp_path, ) with pytest.raises(StartupValidationError) as raised: enforce_startup(settings) assert "MODELFORGE_OPERATOR_API_KEY" in str(raised.value) def test_development_reports_the_same_problems_without_refusing_to_start(tmp_path: Path) -> None: """A developer with no backup key should still be able to run the API — and be told.""" settings = Settings( env="development", operator_api_key=None, artifact_root=str(tmp_path), quarantine_root=str(tmp_path), backup_root=tmp_path, ) report = enforce_startup(settings) assert report.ok or report.problems is not None # never raises outside production def test_the_minimum_postgres_major_is_stated() -> None: assert MINIMUM_POSTGRES_MAJOR >= 16 def test_the_node_agent_pins_the_control_plane_at_the_product_version() -> None: """They ship as one product, so a stale pin makes the agent image unbuildable at release time. Found exactly that way: bumping the product version to 1.0.0 left this at 0.1.0 and the Node Agent image failed to resolve its dependencies during the fresh-install rehearsal. """ data = tomllib.loads((ROOT / "node-agent" / "pyproject.toml").read_text("utf-8")) pins = [ dependency for dependency in data["project"]["dependencies"] if dependency.startswith("modelforge-api") ] assert pins == [f"modelforge-api=={PRODUCT_VERSION}"], ( f"the Node Agent pins {pins}, but the product version is {PRODUCT_VERSION}" )