from __future__ import annotations import json import os import shutil import subprocess import tempfile import unittest from pathlib import Path import yaml ROOT = Path(__file__).resolve().parents[1] COMPOSE = ROOT / "infrastructure" / "compose" / "docker-compose.yml" class ComposeContractTests(unittest.TestCase): @classmethod def setUpClass(cls) -> None: cls.document = yaml.safe_load(COMPOSE.read_text(encoding="utf-8")) def assert_database_candidate_pin( self, name: str, image_ref: str, image_evidence: dict, publication: dict, publication_candidate: dict, ) -> None: artifact = image_evidence["artifact"] published = artifact["publishedImage"] self.assertRegex( image_ref, rf"^gitea\.itworx\.tech/jens/itworx-mcp-hub/{name}:[^@]+@sha256:[0-9a-f]{{64}}$", ) image_name, digest = image_ref.split("@", maxsplit=1) self.assertEqual(published["reference"], image_ref) self.assertEqual(published["taggedReference"], image_name) self.assertEqual(published["manifestDigest"], digest) self.assertEqual(published["imageId"], artifact["imageId"]) self.assertTrue(published["immutablePullConfigMatchedScannedImage"]) self.assertEqual(published["sourceReceiptSha256"], publication["sourceReceiptSha256"]) self.assertEqual(publication_candidate["reference"], image_ref) self.assertEqual(publication_candidate["manifestDigest"], digest) self.assertEqual(publication_candidate["scannedImageId"], artifact["imageId"]) self.assertEqual(publication_candidate["pulledImageId"], artifact["imageId"]) self.assertTrue(publication_candidate["immutablePullConfigMatchedScannedImage"]) self.assertEqual(image_evidence["humanRiskAcceptance"], "pending") self.assertFalse(image_evidence["productionDeployment"]) def test_all_images_are_immutable_or_local_dev(self) -> None: for name, service in self.document["services"].items(): image = service.get("image", "") self.assertNotIn(":latest", image, name) if not image.startswith("local/"): self.assertIn("@sha256:", image, name) def test_socket_is_only_mounted_in_dedicated_broker(self) -> None: for name, service in self.document["services"].items(): mounts = json.dumps(service.get("volumes", [])) if "/var/run/docker.sock" in mounts: self.assertEqual(name, "docker-read-broker") def test_only_edge_services_publish_ports(self) -> None: for name, service in self.document["services"].items(): if service.get("ports"): self.assertIn(name, {"web", "dataplane-router", "inspector"}) def test_broker_is_internal_and_only_socket_consumer(self) -> None: broker = self.document["services"]["docker-read-broker"] connector = self.document["services"]["connector-unraid"] self.assertNotIn("ports", broker) self.assertTrue( any("/var/run/docker.sock" in str(value) for value in broker.get("volumes", [])) ) self.assertFalse( any("/var/run/docker.sock" in str(value) for value in connector.get("volumes", [])) ) self.assertIn("docker_broker_service_token", broker.get("secrets", [])) self.assertIn("docker_broker_service_token", connector.get("secrets", [])) def test_wp050_integration_compose_is_isolated(self) -> None: path = ROOT / "infrastructure/compose/docker-compose.wp050-integration.yml" document = yaml.safe_load(path.read_text(encoding="utf-8")) self.assertEqual(document["name"], "itworx-mcp-hub-wp050") self.assertIn("@sha256:", document["services"]["contextforge"]["image"]) self.assertEqual( document["services"]["contextforge"]["ports"], ["127.0.0.1:${WP050_CONTEXTFORGE_PORT:-18450}:4444"], ) self.assertNotIn("ports", document["services"]["postgres"]) self.assertNotIn("ports", document["services"]["redis"]) def test_integration_uses_the_production_database_artifacts(self) -> None: path = ROOT / "infrastructure/compose/docker-compose.wp050-integration.yml" integration = yaml.safe_load(path.read_text(encoding="utf-8")) for name in ("postgres", "redis"): with self.subTest(service=name): image = self.document["services"][name]["image"] self.assertRegex( image, r"^gitea\.itworx\.tech/jens/itworx-mcp-hub/" rf"{name}:[^@]+@sha256:[0-9a-f]{{64}}$", ) self.assertEqual(integration["services"][name]["image"], image) def test_database_candidate_pins_bind_to_published_scan_identities(self) -> None: evidence_path = ROOT / "evidence/WP-400/DATABASE_REMEDIATION_SCANS.json" evidence = json.loads(evidence_path.read_text(encoding="utf-8")) publication = evidence["publication"] publication_proof = json.loads( (ROOT / "evidence/WP-400/DATABASE_CANDIDATE_PUBLICATION_20261003.json").read_text( encoding="utf-8" ) ) self.assertEqual( publication, {key: value for key, value in publication_proof.items() if key != "candidates"}, ) self.assertFalse(evidence["productionDeployment"]) self.assertFalse(publication["productionDeployment"]) self.assertFalse(publication["cveRiskAccepted"]) self.assertEqual(publication["status"], "published_digest_verified") self.assertRegex(publication["sourceReceiptSha256"], r"^[0-9a-f]{64}$") integration = yaml.safe_load( (ROOT / "infrastructure/compose/docker-compose.wp050-integration.yml").read_text( encoding="utf-8" ) ) security_publication = json.loads( (ROOT / "evidence/WP-400/SECURITY_BASE_PUBLICATION.json").read_text(encoding="utf-8") ) self.assertEqual(security_publication["schema"], "itworx.security-base-preparation/v1") self.assertFalse(security_publication["candidateEmitted"]) self.assertEqual(security_publication["reportCount"], 5) for name in ("postgres", "redis"): with self.subTest(service=name): image_evidence = evidence["images"][name] image_ref = image_evidence["artifact"]["publishedImage"]["reference"] candidate_proof = publication_proof["candidates"][name] self.assert_database_candidate_pin( name, image_ref, image_evidence, publication, candidate_proof, ) if name == "postgres": prepared = next( row for row in security_publication["preparedImages"] if row["service"] == name ) self.assertEqual(prepared["parentRef"], image_ref) self.assertTrue(prepared["runtimeVerified"]) self.assertTrue(prepared["vulnerabilityAndSecretVerified"]) self.assertRegex(prepared["imageId"], r"^sha256:[0-9a-f]{64}$") image_ref = prepared["imageRef"] self.assertEqual(image_ref, self.document["services"][name]["image"]) self.assertEqual(image_ref, integration["services"][name]["image"]) def test_database_candidate_pin_contract_rejects_tag_only_and_config_drift(self) -> None: evidence = json.loads( (ROOT / "evidence/WP-400/DATABASE_REMEDIATION_SCANS.json").read_text(encoding="utf-8") ) publication_proof = json.loads( (ROOT / "evidence/WP-400/DATABASE_CANDIDATE_PUBLICATION_20261003.json").read_text( encoding="utf-8" ) ) for name in ("postgres", "redis"): with self.subTest(service=name): artifact = evidence["images"][name]["artifact"] published = artifact["publishedImage"] valid_ref = published["reference"] tag_only = valid_ref.split("@", maxsplit=1)[0] with self.assertRaises(AssertionError): self.assert_database_candidate_pin( name, tag_only, evidence["images"][name], evidence["publication"], publication_proof["candidates"][name], ) mismatched_image = dict(published, imageId="sha256:" + "0" * 64) drifted_evidence = dict(evidence["images"][name]) drifted_evidence["artifact"] = dict(artifact, publishedImage=mismatched_image) with self.assertRaises(AssertionError): self.assert_database_candidate_pin( name, valid_ref, drifted_evidence, evidence["publication"], publication_proof["candidates"][name], ) def test_inspector_is_lan_only_optional_profile(self) -> None: inspector = self.document["services"]["inspector"] self.assertEqual(inspector.get("profiles"), ["inspector"]) self.assertIn( "INSPECTOR_BIND_IP:?Set an explicit LAN-only Inspector bind address", str(inspector["ports"]), ) routing = (ROOT / "infrastructure/compose/NPM_CONNECTIVITY.md").read_text(encoding="utf-8") self.assertIn("does not receive a public DNS route", routing) def test_contextforge_and_hub_have_distinct_database_credentials(self) -> None: init_script = (ROOT / "infrastructure/postgres/init/001-create-databases.sh").read_text( encoding="utf-8" ) self.assertIn("itworx_mcp_hub", init_script) self.assertIn("contextforge", init_script) self.assertIn("hub_database_password", init_script) self.assertIn("contextforge_database_password", init_script) def test_contextforge_admin_password_and_runtime_token_are_separate(self) -> None: contextforge = self.document["services"]["contextforge"] self.assertIn("contextforge_admin_credential", contextforge["secrets"]) self.assertNotIn("contextforge_api_token", contextforge["secrets"]) self.assertIn( "contextforge-control", contextforge["networks"]["hub-control"]["aliases"], ) for name in ("control-api", "worker", "dataplane-router"): service = self.document["services"][name] self.assertIn("contextforge_api_token", service["secrets"]) self.assertEqual(service["environment"]["HUB_CONTEXTFORGE_AUTH_SCHEME"], "Bearer") self.assertIn( "contextforge-control:4444", service["environment"]["HUB_CONTEXTFORGE_URL"] ) def test_dataplane_owns_mcp_admission_control(self) -> None: contextforge = self.document["services"]["contextforge"] router = self.document["services"]["dataplane-router"] self.assertEqual(contextforge["environment"]["RATE_LIMITING_ENABLED"], "false") self.assertEqual( router["environment"]["HUB_RATE_LIMIT_PER_MINUTE"], "${HUB_RATE_LIMIT_PER_MINUTE:-600}", ) self.assertEqual( router["environment"]["HUB_UPSTREAM_MAX_CONCURRENCY"], "${HUB_UPSTREAM_MAX_CONCURRENCY:-24}", ) self.assertEqual( router["environment"]["HUB_UPSTREAM_QUEUE_TIMEOUT_SECONDS"], "${HUB_UPSTREAM_QUEUE_TIMEOUT_SECONDS:-5}", ) def test_public_oauth_contract_is_shared_by_web_and_dataplane(self) -> None: web = self.document["services"]["web"]["environment"] router = self.document["services"]["dataplane-router"]["environment"] self.assertEqual(web["HUB_PUBLIC_MCP_URL"], router["HUB_PUBLIC_MCP_URL"]) for environment in (web, router): self.assertIn( "${HUB_MCP_ADDITIONAL_OIDC_ISSUER:-", environment["HUB_MCP_ADDITIONAL_OIDC_ISSUER"], ) def test_worker_health_requires_runtime_and_public_surface_readiness(self) -> None: healthcheck = self.document["services"]["worker"]["healthcheck"] self.assertIn("/health/ready", " ".join(healthcheck["test"])) self.assertEqual(healthcheck["interval"], "10s") self.assertEqual(healthcheck["retries"], 18) web = self.document["services"]["web"] self.assertEqual(web["environment"]["HUB_WORKER_URL"], "http://worker:8080") self.assertEqual(web["depends_on"]["worker"]["condition"], "service_healthy") def test_demo_token_revocations_require_an_operator_owned_host_file(self) -> None: router = self.document["services"]["dataplane-router"] self.assertEqual(router["user"], "${PUID:-10001}:${PGID:-10001}") mount = next(value for value in router["volumes"] if ":/etc/itworx:ro" in value) self.assertIn("DEMO_TOKEN_REVOCATIONS_DIR_HOST:?", mount) self.assertNotIn("demo-token-revocations.example.json", mount) def test_operator_tools_are_explicit_and_socket_free(self) -> None: tools = self.document["services"]["operator-tools"] self.assertEqual(tools["profiles"], ["operator-tools"]) self.assertEqual(tools["user"], "${PUID:-10001}:${PGID:-10001}") self.assertEqual(tools["secrets"], ["demo_token_signing_key"]) self.assertNotIn("ports", tools) self.assertNotIn("/var/run/docker.sock", json.dumps(tools["volumes"])) self.assertIn("DEMO_TOKEN_REVOCATIONS_DIR_HOST:?", tools["volumes"][0]) self.assertIn(":/etc/itworx:rw", tools["volumes"][0]) prod = (ROOT / "infrastructure/compose/docker-compose.prod.yml").read_text(encoding="utf-8") self.assertIn( 'image: "${HUB_DATAPLANE_IMAGE:?Set an immutable dataplane image digest}"', prod ) self.assertIn(" operator-tools:\n", prod) self.assertIn(" build: !reset null\n", prod) def test_real_compose_config_has_no_build_for_owned_services(self) -> None: docker = shutil.which("docker") if docker is None: self.skipTest("docker CLI is not available in this environment") prod_compose = ROOT / "infrastructure" / "compose" / "docker-compose.prod.yml" env_file = ROOT / "tests" / "fixtures" / "wp200-compose-config.env" required_secrets = ( "postgres_superuser_password", "hub_database_password", "contextforge_database_password", "contextforge_auth_encryption_secret", "contextforge_jwt_secret", "contextforge_admin_credential", "contextforge_api_token", "docker_broker_service_token", "connector_identity_signing_key", "demo_token_signing_key", "authentik_oidc_client_secret", "hub_session_signing_key", "gitea_mcp_token", "gitea_mcp_write_token", "stitch_api_key", "redis_password", ) with tempfile.TemporaryDirectory() as secrets_dir: for name in required_secrets: (Path(secrets_dir) / name).write_text("wp200-fixture-secret\n", encoding="utf-8") env = {**os.environ, "SECRETS_DIR": secrets_dir} completed = subprocess.run( # noqa: S603 [ docker, "compose", "-f", str(COMPOSE), "-f", str(prod_compose), "--env-file", str(env_file), "config", "--format", "json", ], capture_output=True, text=True, check=False, env=env, ) self.assertEqual(completed.returncode, 0, completed.stderr) rendered = json.loads(completed.stdout) # Derived from the base compose file's own `build:` keys (not a hand-maintained # duplicate list) so this test automatically covers any future service that # gains a `build:` section, rather than silently skipping it. See # evidence/WP-200/ADVERSARIAL_REVIEW.md item (c). owned_with_build = tuple( name for name, service in self.document["services"].items() if "build" in service and not service.get("profiles") ) self.assertTrue(owned_with_build, "no service in the base compose file defines build:") for name in owned_with_build: self.assertIn( name, rendered["services"], f"{name} is missing from the rendered topology" ) self.assertNotIn( "build", rendered["services"][name], f"{name} retains a build definition; local build fallback is possible", ) def test_autodeploy_builds_every_owned_service_with_controller_provenance(self) -> None: overlay = yaml.safe_load( (ROOT / "infrastructure/compose/docker-compose.autodeploy.yml").read_text( encoding="utf-8" ) ) owned = {name for name, service in self.document["services"].items() if "build" in service} self.assertTrue(owned) for name in owned: with self.subTest(service=name): self.assertIn(name, overlay["services"]) service = overlay["services"][name] self.assertIn("${GITEA_COMMIT_SHA:?", service["image"]) arguments = service["build"].get("args", {}) self.assertEqual( arguments.get("GIT_REVISION"), "${GITEA_COMMIT_SHA:?GITEA_COMMIT_SHA is required}", ) self.assertEqual( arguments.get("BUILD_DATE"), "${APPOPS_BUILD_DATE:?APPOPS_BUILD_DATE is required}", ) for argument, value in { "GIT_SOURCE_TREE": "${APPOPS_SOURCE_TREE:?APPOPS_SOURCE_TREE is required}", "IMAGE_VERSION": "${APPOPS_BUILD_ID:?APPOPS_BUILD_ID is required}", "SOURCE_DATE_EPOCH": "${SOURCE_DATE_EPOCH:?SOURCE_DATE_EPOCH is required}", }.items(): self.assertEqual(arguments.get(argument), value, name) original_arguments = self.document["services"][name]["build"].get("args", {}) for key, value in original_arguments.items(): self.assertEqual(arguments.get(key), value) def test_build_dockerfiles_emit_controller_tree_version_and_epoch_labels(self) -> None: dockerfiles = ( "infrastructure/images/python-service/Dockerfile", "infrastructure/images/docker-read-broker/Dockerfile", "infrastructure/images/observability/Dockerfile", ) expected_labels = { "org.opencontainers.image.source-tree": "GIT_SOURCE_TREE", "org.opencontainers.image.version": "IMAGE_VERSION", "org.opencontainers.image.source-date-epoch": "SOURCE_DATE_EPOCH", } for relative_path in dockerfiles: text = (ROOT / relative_path).read_text(encoding="utf-8") with self.subTest(dockerfile=relative_path): for label, argument in expected_labels.items(): self.assertIn(f"ARG {argument}", text) self.assertIn(f'{label}="${{{argument}}}"', text) def test_real_autodeploy_preserves_intake_digest_and_owned_build_revision(self) -> None: docker = shutil.which("docker") if docker is None: self.skipTest("docker CLI is not available in this environment") revision = "a" * 40 built_at = "2026-10-01T18:00:00Z" source_tree = "b" * 40 image_version = revision[:12] source_date_epoch = "1790888863" with tempfile.TemporaryDirectory() as secrets_dir: for name in self.document["secrets"]: (Path(secrets_dir) / name).write_text("test-fixture-secret\n", encoding="utf-8") command = [docker, "compose", "--profile", "knowledge-intake"] for filename in ( "docker-compose.yml", "docker-compose.prod.yml", "docker-compose.autodeploy.yml", "docker-compose.ingestion.prod.yml", ): command.extend(["-f", str(COMPOSE.parent / filename)]) command.extend( [ "--env-file", str(ROOT / "tests/fixtures/wp200-compose-config.env"), "config", "--format", "json", ] ) completed = subprocess.run( # noqa: S603 command, capture_output=True, text=True, check=False, env={ **os.environ, "SECRETS_DIR": secrets_dir, "COMPOSE_PROFILES": "", "CONNECTOR_INGESTION_IMAGE": f"registry.example.test/intake@sha256:{'b' * 64}", "GITEA_COMMIT_SHA": revision, "APPOPS_BUILD_DATE": built_at, "APPOPS_SOURCE_TREE": source_tree, "APPOPS_BUILD_ID": image_version, "SOURCE_DATE_EPOCH": source_date_epoch, }, ) self.assertEqual(completed.returncode, 0, completed.stderr) rendered = json.loads(completed.stdout) self.assertEqual(len(rendered["services"]), 17) self.assertEqual( rendered["services"]["connector-ingestion"]["image"], f"registry.example.test/intake@sha256:{'b' * 64}", ) self.assertNotIn("build", rendered["services"]["connector-ingestion"]) for name, service in rendered["services"].items(): if "build" not in service: continue self.assertEqual(service["build"]["args"]["GIT_REVISION"], revision, name) self.assertEqual(service["build"]["args"]["BUILD_DATE"], built_at, name) self.assertEqual(service["build"]["args"]["GIT_SOURCE_TREE"], source_tree, name) self.assertEqual(service["build"]["args"]["IMAGE_VERSION"], image_version, name) self.assertEqual(service["build"]["args"]["SOURCE_DATE_EPOCH"], source_date_epoch, name) self.assertTrue(service["image"].endswith(f":{revision}"), name) def test_unraid_folder_view_groups_the_complete_production_stack(self) -> None: definition_path = ROOT / "infrastructure" / "unraid" / "folder-view3-itworx-mcp-hub.json" definition = json.loads(definition_path.read_text(encoding="utf-8")) folder = definition["ITWorxMCPHub"] expected_containers = { "itworx-mcp-hub-postgres-1", "itworx-mcp-hub-redis-1", "itworx-mcp-hub-contextforge-1", "itworx-mcp-hub-control-api-1", "itworx-mcp-hub-worker-1", "itworx-mcp-hub-dataplane-router-1", "itworx-mcp-hub-web-1", "itworx-mcp-hub-connector-knowledge-1", "itworx-mcp-hub-connector-gitea-1", "itworx-mcp-hub-connector-unraid-1", "itworx-mcp-hub-connector-mobilityops-1", "itworx-mcp-hub-connector-projectbrain-1", "itworx-mcp-hub-connector-appops-1", "itworx-mcp-hub-connector-stitch-1", "itworx-mcp-hub-docker-read-broker-1", } self.assertEqual(folder["name"], "ITWorx MCP Hub") self.assertEqual(folder["regex"], "^itworx-mcp-hub-") self.assertEqual(set(folder["containers"]), expected_containers) self.assertTrue(folder["settings"]["folder_webui"]) self.assertEqual(folder["settings"]["folder_webui_url"], "http://198.18.10.150:1100") self.assertEqual(folder["icon"], "https://mcp.example.com/itworx-mcp-hub-icon.png") icon_path = ROOT / "apps" / "web" / "public" / "itworx-mcp-hub-icon.png" self.assertTrue(icon_path.is_file()) self.assertEqual(icon_path.read_bytes()[:8], b"\x89PNG\r\n\x1a\n") self.assertLess(icon_path.stat().st_size, 2 * 1024 * 1024) if __name__ == "__main__": unittest.main()