ITWorx_MCP_Hub-Public/tests/test_contextforge_runtime_contract.py

from __future__ import annotations

import importlib.util
import json
from pathlib import Path
from types import SimpleNamespace

import pytest

ROOT = Path(__file__).resolve().parents[1]
MODULE_PATH = ROOT / "scripts" / "contextforge_runtime_contract.py"
CONTRACT_PATH = ROOT / "infrastructure" / "images" / "contextforge" / "runtime-contract.json"


def _load_module():
    spec = importlib.util.spec_from_file_location("contextforge_runtime_contract", MODULE_PATH)
    assert spec is not None and spec.loader is not None
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def _contract() -> dict[str, object]:
    return json.loads(CONTRACT_PATH.read_text(encoding="utf-8"))


def _environment() -> dict[str, str]:
    contract = _contract()
    return dict(contract["productionEnvironment"])  # type: ignore[arg-type]


def test_environment_contract_accepts_exact_production_tuple() -> None:
    module = _load_module()
    module.validate_environment(_contract(), _environment())


@pytest.mark.parametrize(
    ("name", "value"),
    [
        ("GUNICORN_WORKERS", "2"),
        ("USE_STATEFUL_SESSIONS", "true"),
        ("MCPGATEWAY_SESSION_AFFINITY_ENABLED", "true"),
        ("JSON_RESPONSE_ENABLED", "false"),
        ("EMAIL_AUTH_ENABLED", "true"),
        ("MCP_SESSION_POOL_ENABLED", "true"),
        ("MCPGATEWAY_SKIP_MIGRATIONS", "false"),
    ],
)
def test_environment_contract_rejects_transport_or_migration_regression(
    name: str,
    value: str,
) -> None:
    module = _load_module()
    environment = _environment()
    environment[name] = value
    with pytest.raises(module.RuntimeContractError, match=f"environment:{name}"):
        module.validate_environment(_contract(), environment)


def test_contract_digest_requires_exact_repository_contract() -> None:
    module = _load_module()
    expected = module.contract_sha256(CONTRACT_PATH)
    module.validate_contract_digest(
        CONTRACT_PATH,
        {"ITWORX_CONTEXTFORGE_EXPECTED_CONTRACT_SHA256": expected},
    )
    with pytest.raises(module.RuntimeContractError, match="contract-digest"):
        module.validate_contract_digest(
            CONTRACT_PATH,
            {"ITWORX_CONTEXTFORGE_EXPECTED_CONTRACT_SHA256": "0" * 64},
        )


def test_contract_digest_rejects_missing_or_malformed_configuration() -> None:
    module = _load_module()
    for value in ("", "deadbeef", "G" * 64):
        with pytest.raises(module.RuntimeContractError, match="contract-digest-config"):
            module.validate_contract_digest(
                CONTRACT_PATH,
                {"ITWORX_CONTEXTFORGE_EXPECTED_CONTRACT_SHA256": value},
            )


def test_mcp_runtime_rejects_wrong_sdk_version() -> None:
    module = _load_module()
    with pytest.raises(module.RuntimeContractError, match="mcp-version"):
        module.validate_mcp_runtime(
            _contract(),
            version_getter=lambda _: "2.0.0",
            symbol_loader=lambda _module, _symbol: object(),
        )


def test_mcp_runtime_rejects_missing_legacy_symbol() -> None:
    module = _load_module()

    def missing(_module: str, _symbol: str):
        raise AttributeError("missing")

    with pytest.raises(module.RuntimeContractError, match="mcp-symbol"):
        module.validate_mcp_runtime(
            _contract(),
            version_getter=lambda _: "1.30.0",
            symbol_loader=missing,
        )


def test_alembic_head_rejects_schema_mismatch() -> None:
    module = _load_module()
    with pytest.raises(module.RuntimeContractError, match="alembic-head"):
        module.validate_alembic_head(_contract(), ["deadbeef"])


def test_database_revision_rejects_schema_drift() -> None:
    module = _load_module()
    with pytest.raises(module.RuntimeContractError, match="database-revision"):
        module.validate_database_revision(_contract(), ["deadbeef"])


def test_database_revision_accepts_exact_single_head() -> None:
    module = _load_module()
    module.validate_database_revision(_contract(), ["12d4a0c7789c"])


def test_contract_shape_rejects_startup_migration_policy_weakening() -> None:
    module = _load_module()
    contract = _contract()
    contract["migrationPolicy"]["startupMigrations"] = "allowed"  # type: ignore[index]
    with pytest.raises(module.RuntimeContractError, match="migration-policy"):
        module.validate_contract_shape(contract)


def test_health_verifier_can_derive_database_url_from_mounted_secret(
    tmp_path: Path,
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    module = _load_module()
    secret = tmp_path / "contextforge_database_password"
    secret.write_text("p@ss:/word", encoding="utf-8")
    monkeypatch.setattr(module, "CONTEXTFORGE_DB_PASSWORD_FILE", secret)

    resolved = module.resolve_database_url(_environment())

    assert resolved.startswith("postgresql+psycopg://contextforge:")
    assert "@postgres:5432/contextforge" in resolved
    assert "p@ss:/word" not in resolved


def test_query_database_revisions_is_read_only(monkeypatch: pytest.MonkeyPatch) -> None:
    module = _load_module()
    calls: list[str] = []

    class Result:
        def scalars(self):
            return self

        def all(self):
            return ["12d4a0c7789c"]

    class Connection:
        def execute(self, statement):
            calls.append(str(statement))
            return Result()

        def __enter__(self):
            return self

        def __exit__(self, *_args):
            return False

    class Engine:
        def connect(self):
            return Connection()

        def dispose(self):
            calls.append("dispose")

    monkeypatch.setattr(module, "create_engine", lambda *_args, **_kwargs: Engine())
    assert module.query_database_revisions("postgresql://redacted") == ["12d4a0c7789c"]
    assert calls[0].strip().upper() == "SELECT VERSION_NUM FROM ALEMBIC_VERSION"
    blocked_tokens = ("UPDATE", "INSERT", "DELETE", "ALTER", "DROP")
    assert all(token not in calls[0].upper() for token in blocked_tokens)


def test_verify_runtime_combines_all_proofs(monkeypatch: pytest.MonkeyPatch) -> None:
    module = _load_module()
    monkeypatch.setattr(module, "installed_mcp_version", lambda: "1.30.0")
    monkeypatch.setattr(module, "load_mcp_symbol", lambda _module, _symbol: SimpleNamespace())
    monkeypatch.setattr(module, "query_image_alembic_heads", lambda: ["12d4a0c7789c"])
    monkeypatch.setattr(module, "query_database_revisions", lambda _url: ["12d4a0c7789c"])
    module.verify_runtime(
        _contract(),
        _environment() | {"DATABASE_URL": "postgresql://redacted"},
    )