Files
ITWorx_MCP_Hub-Public/tests/test_contextforge_candidate_recipe.py
T

387 lines
17 KiB
Python

# The upstream TOML fixture retains exact source formatting.
# ruff: noqa: E501
from __future__ import annotations
import hashlib
import json
import re
import sys
from pathlib import Path
import pytest
ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT / "scripts"))
import contextforge_candidate_recipe as recipe # noqa: E402 - script path added above
LOCK_DIR = ROOT / "infrastructure" / "images" / "contextforge" / "candidate"
def _hash(data: bytes) -> str:
return hashlib.sha256(data).hexdigest()
def _containerfile_fixture() -> bytes:
return (
b"FROM base\n"
b'RUN if [ "$ENABLE_RUST" = "true" ]; then \\\n dnf upgrade -y; fi\n'
b"RUN set -euo pipefail \\\n && dnf install -y python3.12\n"
b"RUN microdnf upgrade -y \\\n && microdnf install -y python3.12\n"
b'RUN if [ "$ENABLE_FIPS" = "true" ]; then \\\n microdnf install -y crypto-policies; fi\n'
b"COPY pyproject.toml /app/\n"
+ recipe.INSTALL_START.encode("utf-8")
+ b"\n && /app/.venv/bin/uv pip install unpinned\n"
+ recipe.INSTALL_END.encode("utf-8")
+ b"COPY source /app\n"
)
def _source_fixture(tmp_path: Path, containerfile: bytes) -> tuple[Path, dict[str, str]]:
source = tmp_path / "upstream"
source.mkdir()
contents = {
"Containerfile": containerfile,
"pyproject.toml": b"[project]\nname='fixture'\n",
"uv.lock": b"version = 1\n",
}
expected = {name: _hash(value) for name, value in contents.items()}
for name, value in contents.items():
(source / name).write_bytes(value)
return source, expected
def _copy_candidate_locks(destination: Path) -> dict[str, str]:
destination.mkdir(parents=True)
requirements: dict[str, str] = {}
for name in ("production-requirements.lock", "build-tools.lock"):
data = (LOCK_DIR / name).read_bytes()
(destination / name).write_bytes(data)
requirements[name] = _hash(data)
return requirements
def test_candidate_lock_exports_are_hash_complete_and_exclude_build_tools() -> None:
manifest = json.loads((LOCK_DIR / "locks.json").read_text(encoding="utf-8"))
production = (LOCK_DIR / "production-requirements.lock").read_text(encoding="utf-8")
build_tools = (LOCK_DIR / "build-tools.lock").read_text(encoding="utf-8")
assert manifest["upstreamCommit"] == recipe.UPSTREAM_COMMIT
assert manifest["uvVersion"] == recipe.UV_VERSION == "0.12.10"
assert manifest["globalExcludeNewerUtc"] == "2026-09-22T00:00:00Z"
assert manifest["freshnessOverrideUtc"] == "2026-10-02T00:00:00Z"
assert manifest["extras"] == list(recipe.EXTRAS)
assert manifest["constraints"] == list(recipe.CONSTRAINTS)
for name in ("production-requirements.lock", "build-tools.lock"):
assert _hash((LOCK_DIR / name).read_bytes()) == manifest["requirements"][name]
for requirement in recipe.CONSTRAINTS:
package, version = requirement.split("==", 1)
assert re.search(rf"(?im)^{re.escape(package)}=={re.escape(version)}\s*\\", production)
for requirement in recipe.BUILD_TOOLS:
package, version = requirement.split("==", 1)
assert re.search(rf"(?im)^{re.escape(package)}=={re.escape(version)}\s*\\", build_tools)
package_blocks = re.split(r"(?m)(?=^[A-Za-z0-9_.-]+==)", production)
package_blocks = [
block for block in package_blocks if re.match(r"(?m)^[A-Za-z0-9_.-]+==", block)
]
assert len(package_blocks) > 100
assert all(re.search(r"--hash=sha256:[0-9a-f]{64}", block) for block in package_blocks)
assert not re.search(r"(?im)^(pip|setuptools|wheel|uv)==", production)
build_blocks = re.split(r"(?m)(?=^[A-Za-z0-9_.-]+==)", build_tools)
build_blocks = [block for block in build_blocks if re.match(r"(?m)^[A-Za-z0-9_.-]+==", block)]
assert all(re.search(r"--hash=sha256:[0-9a-f]{64}", block) for block in build_blocks)
assert "pip==26.2.1" in build_tools
def test_candidate_pyproject_freezes_relative_freshness_and_preserves_project_metadata() -> None:
source = """[tool.uv]
exclude-newer = "10 days"
[tool.uv.exclude-newer-package]
known-safe-package = "2026-07-01T00:00:00Z"
[tool.uv.sources]
compliance-reference-server = { path = "mcp-servers/python/compliance_reference_server", editable = true }
# Project table follows the local source table.
[project]
name = "contextforge"
dependencies = ["httpx>=0.27"]
[dependency-groups]
dev = ["pytest>=8"]
"""
generated = recipe.candidate_pyproject(source)
import tomllib
parsed = tomllib.loads(generated)
assert parsed["project"] == {
"name": "contextforge",
"dependencies": ["httpx>=0.27"],
}
assert "sources" not in parsed["tool"]["uv"]
assert parsed["tool"]["uv"]["exclude-newer"] == "2026-09-22T00:00:00Z"
assert parsed["tool"]["uv"]["constraint-dependencies"] == list(recipe.CONSTRAINTS)
assert parsed["tool"]["uv"]["exclude-newer-package"]["known-safe-package"] == (
"2026-07-01T00:00:00Z"
)
assert parsed["tool"]["uv"]["exclude-newer-package"]["pyjwt"] == "2026-10-02T00:00:00Z"
assert parsed["dependency-groups"]["build-tools"] == list(recipe.BUILD_TOOLS)
assert "dev" not in parsed["dependency-groups"]
def test_candidate_transform_is_exact_and_uses_hash_locked_runtime_install() -> None:
original = _containerfile_fixture()
transformed = recipe.transform_containerfile(original, expected_hash=_hash(original)).decode()
assert transformed.count("COPY contextforge-candidate-production-requirements.txt") == 1
assert transformed.count("COPY contextforge-candidate-build-tools.txt") == 1
assert (
"--require-hashes --only-binary=:all: -r /tmp/contextforge-production-requirements.txt"
in transformed
)
assert (
"--require-hashes --only-binary=:all: -r /tmp/contextforge-build-tools.txt" in transformed
)
assert (
'--no-deps --no-build-isolation ".[redis,postgres,observability,granian,plugins,llmchat]"'
in transformed
)
assert "uv pip install unpinned" not in transformed
assert "COPY source /app" in transformed
assert not re.search(r"\b(?:dnf|microdnf) (?:upgrade|install)\b", transformed)
assert 'test "$ENABLE_RUST" = "false"' in transformed
assert 'test "$ENABLE_FIPS" = "false"' in transformed
assert 'test "$ENABLE_PROFILING" = "false"' in transformed
assert 'test "$(id -u app):$(id -g app)" = "10001:10001"' in transformed
def test_candidate_os_freeze_rejects_missing_or_additional_transactions() -> None:
original = _containerfile_fixture().decode()
for changed in (
original.replace("dnf upgrade -y", "echo changed", 1),
original + "\nRUN dnf install -y unexpected\n",
):
with pytest.raises(recipe.CandidateRecipeError, match="RPM transaction count"):
recipe.freeze_os_layers(changed)
def test_candidate_transform_rejects_source_drift_and_unexpected_anchor_counts() -> None:
original = _containerfile_fixture()
digest = _hash(original)
with pytest.raises(recipe.CandidateRecipeError, match="Containerfile hash mismatch"):
recipe.transform_containerfile(original + b"# drift\n", expected_hash=digest)
with pytest.raises(recipe.CandidateRecipeError, match="anchor count changed"):
recipe.transform_containerfile(
original.replace(b"COPY pyproject.toml /app/\n", b""),
expected_hash=_hash(original.replace(b"COPY pyproject.toml /app/\n", b"")),
)
def test_source_verification_rejects_missing_and_tampered_pinned_inputs(tmp_path: Path) -> None:
source, expected = _source_fixture(tmp_path, _containerfile_fixture())
assert recipe.verify_upstream_source(source, expected) == {
name: (source / name).read_bytes() for name in expected
}
(source / "uv.lock").write_bytes(b"changed lock")
with pytest.raises(recipe.CandidateRecipeError, match="uv.lock hash mismatch"):
recipe.verify_upstream_source(source, expected)
(source / "uv.lock").unlink()
with pytest.raises(recipe.CandidateRecipeError, match="missing: uv.lock"):
recipe.verify_upstream_source(source, expected)
def test_prepare_copies_verified_locks_and_fails_before_writing_on_missing_lock(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
source, source_hashes = _source_fixture(tmp_path, _containerfile_fixture())
lock_dir = tmp_path / "candidate-locks"
lock_hashes = _copy_candidate_locks(lock_dir)
manifest = recipe._make_manifest(
source_hashes=source_hashes,
production_hash=lock_hashes["production-requirements.lock"],
build_tools_hash=lock_hashes["build-tools.lock"],
)
(lock_dir / "locks.json").write_bytes(recipe._canonical_json(manifest))
monkeypatch.setattr(recipe, "SOURCE_HASHES", source_hashes)
monkeypatch.setattr(recipe, "LOCK_DIR", lock_dir)
monkeypatch.setattr(recipe, "LOCK_MANIFEST", lock_dir / "locks.json")
recipe.prepare_source(source)
assert (source / "Containerfile.itworx-candidate").exists()
for name, target in recipe.LOCK_OUTPUT_NAMES.items():
assert (source / target).read_bytes() == (lock_dir / name).read_bytes()
(source / "Containerfile.itworx-candidate").unlink()
(lock_dir / "build-tools.lock").unlink()
with pytest.raises(recipe.CandidateRecipeError, match="missing: build-tools.lock"):
recipe.prepare_source(source)
assert not (source / "Containerfile.itworx-candidate").exists()
def test_candidate_build_args_pin_every_base_and_disable_optional_builds() -> None:
args = recipe.candidate_build_args()
assert set(args) == {
"UBI_BASE",
"NODEJS_IMAGE",
"UBI_MINIMAL",
"WHEELS_REF",
"PYTHON_VERSION",
"ENABLE_RUST",
"ENABLE_RUST_MCP_RMCP",
"ENABLE_PROFILING",
"ENABLE_FIPS",
}
assert all("@sha256:" in args[key] for key in ("UBI_BASE", "NODEJS_IMAGE", "UBI_MINIMAL"))
assert args["UBI_BASE"] == (
"gitea.example.com/jens/itworx-mcp-hub/contextforge-os-builder@"
"sha256:e23a733abd28376eb3328676a3ab1e6d153122228110de0c881bb5f3c38a4483"
)
assert args["UBI_MINIMAL"] == (
"gitea.example.com/jens/itworx-mcp-hub/contextforge-os-runtime@"
"sha256:9b23d8152f253b10e47ce077bb9c5f8c3c370f3e0d5533f8a4a0e986ff176236"
)
assert args["NODEJS_IMAGE"] == (
"registry.access.redhat.com/ubi10/nodejs-24@"
"sha256:0f110e68682f4a49188ba4db786eb4f207b119420913fc76bb5ba9512ffbef06"
)
assert args["WHEELS_REF"] == args["UBI_MINIMAL"]
assert args["PYTHON_VERSION"] == "3.12"
assert (
args["ENABLE_RUST"] == args["ENABLE_RUST_MCP_RMCP"] == args["ENABLE_PROFILING"] == "false"
)
assert args["ENABLE_FIPS"] == "false"
def test_frozen_base_manifest_binds_published_digests_and_rpm_inventories() -> None:
manifest = json.loads((LOCK_DIR / "locks.json").read_text(encoding="utf-8"))
bases = {item["arg"]: item for item in manifest["baseImages"]}
assert set(bases) == {"UBI_BASE", "NODEJS_IMAGE", "UBI_MINIMAL"}
assert bases["UBI_BASE"] == {
"arg": "UBI_BASE",
"sourceTag": (
"gitea.example.com/jens/itworx-mcp-hub/contextforge-os-builder:"
"sha256-c6bb70a7a340ed1047e7cb934aa55b086ee68a5eb9ba6b2ad2e6825c7184a3e8"
),
"reference": (
"gitea.example.com/jens/itworx-mcp-hub/contextforge-os-builder@"
"sha256:e23a733abd28376eb3328676a3ab1e6d153122228110de0c881bb5f3c38a4483"
),
"parentReference": (
"registry.access.redhat.com/ubi10@"
"sha256:1e32d829a4de713a835d702d3bc05f0c448a08590796281b7dec181912e9aeb8"
),
"imageId": "sha256:c6bb70a7a340ed1047e7cb934aa55b086ee68a5eb9ba6b2ad2e6825c7184a3e8",
"rpmInventorySha256": "6a8c0d4f66fc5a38008541ad15614711e7f536cd18c70c02d2eb0691552b88b0",
"platform": "linux/amd64",
"childManifestDigest": (
"sha256:e23a733abd28376eb3328676a3ab1e6d153122228110de0c881bb5f3c38a4483"
),
}
assert bases["UBI_MINIMAL"] == {
"arg": "UBI_MINIMAL",
"sourceTag": (
"gitea.example.com/jens/itworx-mcp-hub/contextforge-os-runtime:"
"sha256-fd94e2643dfc0401790d0528d3740bb2f0c04709e743f6c774a4e8a38606e40b"
),
"reference": (
"gitea.example.com/jens/itworx-mcp-hub/contextforge-os-runtime@"
"sha256:9b23d8152f253b10e47ce077bb9c5f8c3c370f3e0d5533f8a4a0e986ff176236"
),
"parentReference": (
"registry.access.redhat.com/ubi10/ubi-minimal@"
"sha256:afbfb5c18c628796fed3ad3122dcf096e0b2dec9dad3fbabe8620de06bc6f078"
),
"imageId": "sha256:fd94e2643dfc0401790d0528d3740bb2f0c04709e743f6c774a4e8a38606e40b",
"rpmInventorySha256": "a74d60ecb43d1aa561874305d1a9c62ac35c99a14c1dc275ba1656f14df7d034",
"platform": "linux/amd64",
"childManifestDigest": (
"sha256:9b23d8152f253b10e47ce077bb9c5f8c3c370f3e0d5533f8a4a0e986ff176236"
),
}
assert bases["NODEJS_IMAGE"]["reference"] == (
"registry.access.redhat.com/ubi10/nodejs-24@"
"sha256:0f110e68682f4a49188ba4db786eb4f207b119420913fc76bb5ba9512ffbef06"
)
assert "imageId" not in bases["NODEJS_IMAGE"]
def test_frozen_base_reference_drift_fails_before_candidate_build(
monkeypatch: pytest.MonkeyPatch,
) -> None:
original = recipe.BASE_IMAGES[0]
changed = dict(original)
changed["reference"] = changed["reference"].replace("e23a733", "f23a733", 1)
monkeypatch.setattr(recipe, "BASE_IMAGES", (changed, *recipe.BASE_IMAGES[1:]))
with pytest.raises(recipe.CandidateRecipeError, match="reviewed contract"):
recipe._read_lock_manifest()
def test_frozen_base_evidence_matches_candidate_pins_and_repository_inputs() -> None:
evidence_path = ROOT / "evidence" / "WP-400" / "CF_FROZEN_BASES.json"
evidence = json.loads(evidence_path.read_text(encoding="utf-8"))
manifest = json.loads((LOCK_DIR / "locks.json").read_text(encoding="utf-8"))
bases = {item["arg"]: item for item in manifest["baseImages"]}
assert evidence["schema"] == "itworx.contextforge-frozen-os-bases-evidence/v1"
assert evidence["status"] == "passed"
assert evidence["finalContextForgeCandidateBuilt"] is False
assert evidence["sourceInputs"]["upstreamCommit"] == manifest["upstreamCommit"]
assert evidence["sourceInputs"]["ContainerfileSha256"] == manifest["upstreamFiles"][
"Containerfile"
]
assert evidence["sourceInputs"]["pyprojectTomlSha256"] == manifest["upstreamFiles"][
"pyproject.toml"
]
assert evidence["sourceInputs"]["uvLockSha256"] == manifest["upstreamFiles"]["uv.lock"]
assert set(evidence["hubInputsSha256"]) == {
"scripts/contextforge_candidate_recipe.py",
"scripts/build_contextforge_compat_image.sh",
"infrastructure/images/contextforge/candidate/locks.json",
"infrastructure/images/contextforge/candidate/production-requirements.lock",
"infrastructure/images/contextforge/candidate/build-tools.lock",
"infrastructure/images/contextforge/os-bases/Dockerfile.builder",
"infrastructure/images/contextforge/os-bases/Dockerfile.runtime",
"infrastructure/compose/docker-compose.autodeploy.yml",
}
assert evidence["hubInputsSha256"] == {
name: _hash((ROOT / name).read_bytes()) for name in evidence["hubInputsSha256"]
}
for role, arg in (("builder", "UBI_BASE"), ("runtime", "UBI_MINIMAL")):
published = evidence["publishedBases"][role]
locked = bases[arg]
assert published["reference"] == locked["reference"]
assert published["imageId"] == locked["imageId"]
assert published["rpmInventorySha256"] == locked["rpmInventorySha256"]
assert published["platform"] == locked["platform"] == "linux/amd64"
assert published["publication"]["pulledByDigest"] is True
assert published["publication"]["manifestConfigImageIdMatches"] is True
assert published["publication"]["rpmInventoryMatchesAfterPull"] is True
assert published["rpmInventory"]["exactMatch"] is True
assert published["scan"]["secretFindings"] == 0
assert published["scan"]["vulnerabilityFindingRecords"] == 0
def test_candidate_manifest_rejects_changed_lock_bytes(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
lock_dir = tmp_path / "locks"
lock_hashes = _copy_candidate_locks(lock_dir)
manifest = recipe._make_manifest(
source_hashes=recipe.SOURCE_HASHES,
production_hash=lock_hashes["production-requirements.lock"],
build_tools_hash=lock_hashes["build-tools.lock"],
)
(lock_dir / "locks.json").write_bytes(recipe._canonical_json(manifest))
(lock_dir / "production-requirements.lock").write_bytes(b"changed")
monkeypatch.setattr(recipe, "LOCK_DIR", lock_dir)
monkeypatch.setattr(recipe, "LOCK_MANIFEST", lock_dir / "locks.json")
with pytest.raises(recipe.CandidateRecipeError, match="lock hash mismatch"):
recipe._read_lock_manifest()