65 lines
2.4 KiB
Python
65 lines
2.4 KiB
Python
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
PROD = ROOT / "infrastructure" / "compose" / "docker-compose.prod.yml"
|
|
ENTRYPOINT = ROOT / "infrastructure" / "images" / "contextforge" / "entrypoint.sh"
|
|
|
|
|
|
def _prod_contextforge_block() -> str:
|
|
source = PROD.read_text(encoding="utf-8")
|
|
return source.split(" contextforge:\n", 1)[1].split(" control-api:", 1)[0]
|
|
|
|
|
|
def test_strict_production_contract_cannot_be_weakened() -> None:
|
|
service = _prod_contextforge_block()
|
|
assert "image: itworx-mcp-hub/contextforge-schema:12d4a0c7789c" in service
|
|
assert "pull_policy: never" in service
|
|
expected = {
|
|
"GUNICORN_WORKERS": "1",
|
|
"USE_STATEFUL_SESSIONS": "false",
|
|
"MCPGATEWAY_SESSION_AFFINITY_ENABLED": "false",
|
|
"JSON_RESPONSE_ENABLED": "true",
|
|
"EMAIL_AUTH_ENABLED": "false",
|
|
"MCP_SESSION_POOL_ENABLED": "false",
|
|
"MCPGATEWAY_SKIP_MIGRATIONS": "true",
|
|
"ITWORX_CONTEXTFORGE_CONTRACT_MODE": "strict",
|
|
}
|
|
for name, value in expected.items():
|
|
assert f'{name}: "{value}"' in service
|
|
for mount in (
|
|
"../images/contextforge/entrypoint.sh:/opt/itworx/contextforge-entrypoint.sh:ro",
|
|
(
|
|
"../images/contextforge/runtime-contract.json:"
|
|
"/opt/itworx/contextforge-runtime-contract.json:ro"
|
|
),
|
|
(
|
|
"../../scripts/contextforge_runtime_contract.py:"
|
|
"/opt/itworx/contextforge-runtime-contract.py:ro"
|
|
),
|
|
):
|
|
assert f'"{mount}"' in service
|
|
assert (
|
|
'ITWORX_CONTEXTFORGE_EXPECTED_CONTRACT_SHA256: '
|
|
'"11345ea89e229e7bf164eb037b525b685a5dfb143c08a220544c0b6cdde411d9"'
|
|
in service
|
|
)
|
|
|
|
|
|
def test_entrypoint_runs_read_only_contract_gate_before_upstream_startup() -> None:
|
|
source = ENTRYPOINT.read_text(encoding="utf-8")
|
|
verify = "/app/.venv/bin/python /opt/itworx/contextforge-runtime-contract.py --startup"
|
|
upstream = "exec /app/docker-entrypoint.sh"
|
|
assert verify in source
|
|
assert source.index(verify) < source.index(upstream)
|
|
assert 'ITWORX_CONTEXTFORGE_CONTRACT_MODE:-relaxed' in source
|
|
|
|
|
|
def test_production_contextforge_healthcheck_combines_contract_and_http_health() -> None:
|
|
service = _prod_contextforge_block()
|
|
assert "contextforge-runtime-contract.py --health" in service
|
|
assert "http://127.0.0.1:4444/health" in service
|
|
assert "interval: 30s" in service
|
|
assert "retries: 3" in service
|