557 lines
22 KiB
Python
557 lines
22 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import importlib
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import types
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
BROKER_DIR = ROOT / "ops" / "gitea-deploy-broker"
|
|
|
|
|
|
@pytest.fixture
|
|
def modules(monkeypatch):
|
|
if "fcntl" not in sys.modules:
|
|
shim = types.ModuleType("fcntl")
|
|
shim.LOCK_EX, shim.LOCK_UN, shim.flock = 2, 8, lambda *_args: None
|
|
monkeypatch.setitem(sys.modules, "fcntl", shim)
|
|
monkeypatch.syspath_prepend(str(BROKER_DIR))
|
|
sys.modules.pop("deploy", None)
|
|
sys.modules.pop("prepare_mcp_hub_candidate", None)
|
|
controller = importlib.import_module("deploy")
|
|
utility = importlib.import_module("prepare_mcp_hub_candidate")
|
|
monkeypatch.setattr(utility, "_is_host_root", lambda: True)
|
|
yield controller, utility
|
|
sys.modules.pop("prepare_mcp_hub_candidate", None)
|
|
sys.modules.pop("deploy", None)
|
|
|
|
|
|
@pytest.fixture
|
|
def harness(modules, tmp_path, monkeypatch):
|
|
controller, utility = modules
|
|
revision = "a" * 40
|
|
epoch = "1790888863"
|
|
now = (
|
|
datetime.now(timezone.utc)
|
|
.isoformat(timespec="milliseconds")
|
|
.replace("+00:00", "Z")
|
|
)
|
|
metadata = {
|
|
"id": revision[:12],
|
|
"repository": controller.MCP_HUB_REPOSITORY,
|
|
"revision": revision,
|
|
"tree": "b" * 40,
|
|
"treeDigest": hashlib.sha256(("git-tree:" + "b" * 40).encode()).hexdigest(),
|
|
"builtAt": now,
|
|
}
|
|
repo = tmp_path / "hub.git"
|
|
project = {
|
|
"branch": "main",
|
|
"slug": controller.MCP_HUB_PROJECT,
|
|
"compose_project": controller.MCP_HUB_PROJECT,
|
|
"containers": list(controller.MCP_HUB_CONTAINERS),
|
|
"_mcp_hub_contract": controller.MCP_HUB_TOPOLOGY_CONTRACT,
|
|
"_repository_path": str(repo),
|
|
"repo_name": "itworx_mcp_hub",
|
|
}
|
|
state = types.SimpleNamespace(
|
|
source=tmp_path / "releases" / revision,
|
|
compose_marker="baseline",
|
|
images={},
|
|
mutations=[],
|
|
inspect_calls=[],
|
|
)
|
|
state.source.mkdir(parents=True)
|
|
(state.source / "scripts").mkdir()
|
|
draft = state.source / "scripts" / "draft_cve_exceptions.py"
|
|
draft.write_text(
|
|
"import argparse, json\n"
|
|
"from datetime import date, timedelta\n"
|
|
"from pathlib import Path\n"
|
|
"p=argparse.ArgumentParser()\n"
|
|
"p.add_argument('--scan-summary', required=True)\n"
|
|
"p.add_argument('--image-digests', required=True)\n"
|
|
"p.add_argument('--evidence-reference', required=True)\n"
|
|
"p.add_argument('--output', required=True)\n"
|
|
"a=p.parse_args()\n"
|
|
"scan=json.loads(Path(a.scan_summary).read_text())\n"
|
|
"digest=json.loads(Path(a.image_digests).read_text())['images']\n"
|
|
"exceptions=[]\n"
|
|
"for row in scan['unresolvedCriticalOrHigh']:\n"
|
|
" refs=set(row['affectedImages'])\n"
|
|
" affected={s:v['imageId'] for s,v in digest.items() if v['tag'] in refs}\n"
|
|
" for cve in row['cves']:\n"
|
|
" exceptions.append({'id':row['package']+'-'+cve,'package':row['package'],'cve':cve,\n"
|
|
" 'installedVersion':row['installedVersion'],'severity':row['severity'],\n"
|
|
" 'approvalStatus':'pending','riskOwner':'','approver':'','affectedImageDigests':affected,\n"
|
|
" 'evidenceReference':a.evidence_reference})\n"
|
|
"Path(a.output).write_text(json.dumps({'schemaVersion':'1.0','exceptions':exceptions}))\n"
|
|
"",
|
|
encoding="utf-8",
|
|
)
|
|
raw_project = dict(project)
|
|
monkeypatch.setattr(controller, "STATE_ROOT", tmp_path)
|
|
monkeypatch.setattr(controller, "MCP_HUB_RELEASES_ROOT", tmp_path / "releases")
|
|
monkeypatch.setattr(controller, "_mcp_hub_require_trusted_state_root", lambda: None)
|
|
monkeypatch.setattr(controller, "_mcp_hub_require_no_posix_acl", lambda _path: None)
|
|
monkeypatch.setattr(controller, "_root_owned", lambda _path: True)
|
|
monkeypatch.setattr(
|
|
controller,
|
|
"load_projects",
|
|
lambda: {controller.MCP_HUB_REPOSITORY: raw_project},
|
|
)
|
|
monkeypatch.setattr(
|
|
controller, "effective_mcp_hub_project", lambda repository, value: dict(value)
|
|
)
|
|
monkeypatch.setattr(
|
|
controller, "repository_path", lambda _repository, _project: repo
|
|
)
|
|
monkeypatch.setattr(controller, "output", lambda _args: revision)
|
|
monkeypatch.setattr(
|
|
controller,
|
|
"validate_commit",
|
|
lambda _repo, branch, commit: commit if branch == "main" else "",
|
|
)
|
|
monkeypatch.setattr(
|
|
controller, "deployment_metadata", lambda *_args, **_kwargs: dict(metadata)
|
|
)
|
|
monkeypatch.setattr(controller, "mcp_hub_source_date_epoch", lambda *_args: epoch)
|
|
|
|
def stage(_project, _repo, commit):
|
|
assert commit == revision
|
|
return state.source
|
|
|
|
monkeypatch.setattr(controller, "mcp_hub_stage_release_source", stage)
|
|
monkeypatch.setattr(
|
|
controller, "exact_source_matches_revision", lambda *_args: True
|
|
)
|
|
monkeypatch.setattr(
|
|
controller, "_mcp_hub_verify_release_tree_metadata", lambda *_args: None
|
|
)
|
|
monkeypatch.setattr(controller, "mcp_hub_validate_secret_files", lambda _model: 21)
|
|
|
|
def rendered(project_arg, commit, repository):
|
|
assert commit == revision
|
|
assert repository == controller.MCP_HUB_REPOSITORY
|
|
services = {}
|
|
for index, service in enumerate(controller.MCP_HUB_SERVICES, 1):
|
|
if service in controller.MCP_HUB_OWNED_IMAGE_REPOSITORIES:
|
|
local = controller.MCP_HUB_LOCAL_BUILD_IMAGE_REPOSITORIES.get(service)
|
|
reference = (
|
|
f"{local}:{revision[:12]}"
|
|
if local
|
|
else f"{controller.MCP_HUB_OWNED_IMAGE_REPOSITORIES[service]}@sha256:{index:064x}"
|
|
)
|
|
elif service == "contextforge":
|
|
reference = f"itworx-mcp-hub/contextforge-schema:{revision}"
|
|
elif service == "ingestion-clamd":
|
|
reference = controller.MCP_HUB_INTAKE_CLAMAV_IMAGE
|
|
elif service == "postgres":
|
|
reference = controller.MCP_HUB_POSTGRES_IMAGE
|
|
elif service == "redis":
|
|
reference = controller.MCP_HUB_REDIS_IMAGE
|
|
else:
|
|
reference = f"fixture/{service}@sha256:{index:064x}"
|
|
services[service] = {"image": reference}
|
|
return {"services": services, "testMarker": state.compose_marker}
|
|
|
|
monkeypatch.setattr(controller, "mcp_hub_compose_model", rendered)
|
|
|
|
def inspect(reference):
|
|
state.inspect_calls.append(reference)
|
|
return state.images.get(reference)
|
|
|
|
monkeypatch.setattr(controller, "inspect_image", inspect)
|
|
monkeypatch.setattr(
|
|
controller, "run", lambda *_args, **_kwargs: state.mutations.append("run")
|
|
)
|
|
monkeypatch.setattr(utility, "_is_host_root", lambda: True)
|
|
|
|
plan = utility.plan_candidate(controller.MCP_HUB_REPOSITORY, revision)
|
|
candidate_dir = tmp_path / "mcp-hub-release-candidates" / revision
|
|
plan_doc = json.loads((candidate_dir / "build.json").read_text())
|
|
model = rendered(project, revision, controller.MCP_HUB_REPOSITORY)
|
|
labels = {
|
|
"org.opencontainers.image.source": controller.MCP_HUB_OCI_IMAGE_SOURCE,
|
|
"org.opencontainers.image.revision": revision,
|
|
"org.opencontainers.image.source-tree": metadata["tree"],
|
|
"org.opencontainers.image.version": metadata["id"],
|
|
"org.opencontainers.image.created": metadata["builtAt"],
|
|
"org.opencontainers.image.source-date-epoch": epoch,
|
|
}
|
|
for index, service in enumerate(controller.MCP_HUB_SERVICES, 1):
|
|
reference = model["services"][service]["image"]
|
|
image_id = "sha256:" + f"{index:064x}"
|
|
config = {"User": "10001:10001"}
|
|
if (
|
|
service in controller.MCP_HUB_OWNED_IMAGE_REPOSITORIES
|
|
or service == "contextforge"
|
|
):
|
|
config["Labels"] = dict(labels)
|
|
state.images[reference] = {"Id": image_id, "Config": config}
|
|
|
|
def write_input(name: str, value: object) -> Path:
|
|
path = candidate_dir / name
|
|
data = (
|
|
json.dumps(value, sort_keys=True, separators=(",", ":")) + "\n"
|
|
).encode()
|
|
descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
|
|
with os.fdopen(descriptor, "wb") as handle:
|
|
handle.write(data)
|
|
path.chmod(0o600)
|
|
return path
|
|
|
|
for index, service in enumerate(controller.MCP_HUB_SERVICES, 1):
|
|
image_id = "sha256:" + f"{index:064x}"
|
|
vuln = {
|
|
"SchemaVersion": 2,
|
|
"Trivy": {"Version": "0.73.0"},
|
|
"CreatedAt": now,
|
|
"Metadata": {
|
|
"ImageID": image_id,
|
|
"ImageConfig": {"architecture": "amd64", "os": "linux"},
|
|
},
|
|
"Results": [
|
|
{
|
|
"Class": "os-pkgs",
|
|
"Packages": [{"Name": "fixture", "Version": "1"}],
|
|
"Vulnerabilities": [],
|
|
}
|
|
],
|
|
}
|
|
secret = {
|
|
"ImageID": image_id,
|
|
"CreatedAt": now,
|
|
"TrivyVersion": "0.73.0",
|
|
"FindingCount": 0,
|
|
"SensitiveFindingDetailsPersisted": False,
|
|
"Results": [{"FindingCount": 0}],
|
|
}
|
|
write_input(f"vulnerability-{service}.json", vuln)
|
|
write_input(f"secret-{service}.json", secret)
|
|
write_input("trivy-db-metadata.json", {"UpdatedAt": now, "DownloadedAt": now})
|
|
|
|
# The reports must match the IDs in the final inspect inventory. Keep a
|
|
# convenient writer for intentional drift cases in each test.
|
|
monkeypatch.setattr(utility, "_test_write_input", write_input, raising=False)
|
|
yield types.SimpleNamespace(
|
|
controller=controller,
|
|
utility=utility,
|
|
revision=revision,
|
|
epoch=epoch,
|
|
metadata=metadata,
|
|
project=project,
|
|
state=state,
|
|
candidate_dir=candidate_dir,
|
|
model=model,
|
|
write_input=write_input,
|
|
now=now,
|
|
plan=plan,
|
|
plan_doc=plan_doc,
|
|
)
|
|
|
|
|
|
def test_plan_freezes_protected_metadata_without_candidate_or_approval(harness):
|
|
h = harness
|
|
document = json.loads((h.candidate_dir / "build.json").read_text())
|
|
assert document["schema"] == "itworx.mcp-hub-build-plan/v1"
|
|
assert document["repository"] == h.controller.MCP_HUB_REPOSITORY
|
|
assert document["revision"] == h.revision
|
|
assert document["build"] == h.metadata
|
|
assert document["sourceDateEpoch"] == h.epoch
|
|
assert len(document["composeSha256"]) == 64
|
|
assert not (h.candidate_dir / "candidate.json").exists()
|
|
assert not (h.candidate_dir / "approval.json").exists()
|
|
assert h.state.mutations == []
|
|
|
|
|
|
def test_wrong_repository_and_nonroot_fail_before_controller_access(
|
|
modules, monkeypatch
|
|
):
|
|
controller, utility = modules
|
|
monkeypatch.setattr(utility, "_is_host_root", lambda: True)
|
|
monkeypatch.setattr(
|
|
controller, "load_projects", lambda: pytest.fail("must reject repo first")
|
|
)
|
|
with pytest.raises(controller.DeployError, match="fixed MCP Hub repository"):
|
|
utility.plan_candidate("someone/else", "a" * 40)
|
|
monkeypatch.setattr(utility, "_is_host_root", lambda: False)
|
|
with pytest.raises(controller.DeployError, match="root"):
|
|
utility.plan_candidate(controller.MCP_HUB_REPOSITORY, "a" * 40)
|
|
|
|
|
|
def test_non_exact_main_head_is_denied_before_staging(modules, monkeypatch, tmp_path):
|
|
controller, utility = modules
|
|
monkeypatch.setattr(utility, "_is_host_root", lambda: True)
|
|
project = {
|
|
"branch": "main",
|
|
"_mcp_hub_contract": controller.MCP_HUB_TOPOLOGY_CONTRACT,
|
|
}
|
|
monkeypatch.setattr(
|
|
controller, "load_projects", lambda: {controller.MCP_HUB_REPOSITORY: project}
|
|
)
|
|
monkeypatch.setattr(
|
|
controller, "effective_mcp_hub_project", lambda _repo, value: value
|
|
)
|
|
monkeypatch.setattr(
|
|
controller, "repository_path", lambda *_args: tmp_path / "hub.git"
|
|
)
|
|
monkeypatch.setattr(controller, "output", lambda _args: "c" * 40)
|
|
monkeypatch.setattr(
|
|
controller,
|
|
"mcp_hub_stage_release_source",
|
|
lambda *_args: pytest.fail("must not stage"),
|
|
)
|
|
with pytest.raises(controller.DeployError, match="exact protected main head"):
|
|
utility.plan_candidate(controller.MCP_HUB_REPOSITORY, "a" * 40)
|
|
|
|
|
|
def test_capture_emits_candidate_and_pending_only_envelope_without_apply_calls(
|
|
harness, monkeypatch
|
|
):
|
|
h = harness
|
|
calls = []
|
|
original_run = h.utility.subprocess.run
|
|
|
|
def observe_draft(command, **kwargs):
|
|
calls.append((command, kwargs))
|
|
return original_run(command, **kwargs)
|
|
|
|
monkeypatch.setattr(h.utility.subprocess, "run", observe_draft)
|
|
result = h.utility.capture_candidate(h.controller.MCP_HUB_REPOSITORY, h.revision)
|
|
candidate = json.loads((h.candidate_dir / "candidate.json").read_text())
|
|
exceptions = json.loads((h.candidate_dir / "exceptions.json").read_text())
|
|
digests = json.loads((h.candidate_dir / "image-digests.json").read_text())
|
|
assert result["status"] == "pending_review"
|
|
assert set(candidate["images"]) == set(h.controller.MCP_HUB_SERVICES)
|
|
assert digests["images"]["web"]["tag"] == candidate["images"]["web"]["imageRef"]
|
|
assert digests["images"]["web"]["imageId"] == candidate["images"]["web"]["imageId"]
|
|
candidate_bytes = (h.candidate_dir / "candidate.json").read_bytes()
|
|
exceptions_bytes = (h.candidate_dir / "exceptions.json").read_bytes()
|
|
assert result["candidateSha256"] == hashlib.sha256(candidate_bytes).hexdigest()
|
|
assert result["exceptionsSha256"] == hashlib.sha256(exceptions_bytes).hexdigest()
|
|
assert not (h.candidate_dir / "approval.json").exists()
|
|
assert all(item["approvalStatus"] == "pending" for item in exceptions["exceptions"])
|
|
assert h.state.mutations == []
|
|
assert len(h.state.inspect_calls) == 17
|
|
assert len(calls) == 1
|
|
assert (
|
|
Path(calls[0][0][1]) == h.state.source / "scripts" / "draft_cve_exceptions.py"
|
|
)
|
|
assert calls[0][1]["stdout"] is subprocess.DEVNULL
|
|
assert calls[0][1]["stderr"] is subprocess.DEVNULL
|
|
assert set(calls[0][1]["env"]) == {
|
|
"PATH",
|
|
"PYTHONDONTWRITEBYTECODE",
|
|
"PYTHONIOENCODING",
|
|
}
|
|
|
|
|
|
@pytest.mark.parametrize("mutation", ["label", "root"])
|
|
def test_capture_rejects_contextforge_provenance_drift_without_candidate(
|
|
harness, mutation
|
|
):
|
|
h = harness
|
|
reference = h.model["services"]["contextforge"]["image"]
|
|
config = h.state.images[reference]["Config"]
|
|
if mutation == "label":
|
|
config["Labels"]["org.opencontainers.image.source-tree"] = "wrong"
|
|
else:
|
|
config["User"] = "0:10001"
|
|
with pytest.raises(h.controller.DeployError, match="provenance|non-root runtime"):
|
|
h.utility.capture_candidate(h.controller.MCP_HUB_REPOSITORY, h.revision)
|
|
assert not (h.candidate_dir / "candidate.json").exists()
|
|
assert not (h.candidate_dir / "approval.json").exists()
|
|
assert h.state.mutations == []
|
|
|
|
|
|
def test_pending_high_findings_are_drafted_from_the_exact_staged_source(harness):
|
|
h = harness
|
|
report_path = h.candidate_dir / "vulnerability-web.json"
|
|
report = json.loads(report_path.read_text())
|
|
report["Results"][0]["Vulnerabilities"] = [
|
|
{
|
|
"PkgName": "libfixture",
|
|
"VulnerabilityID": "CVE-2026-12345",
|
|
"InstalledVersion": "1.2.3",
|
|
"Severity": "HIGH",
|
|
}
|
|
]
|
|
report_path.unlink()
|
|
h.write_input(report_path.name, report)
|
|
result = h.utility.capture_candidate(h.controller.MCP_HUB_REPOSITORY, h.revision)
|
|
candidate = json.loads((h.candidate_dir / "candidate.json").read_text())
|
|
summary = json.loads((h.candidate_dir / "scan-summary.json").read_text())
|
|
exceptions = json.loads((h.candidate_dir / "exceptions.json").read_text())[
|
|
"exceptions"
|
|
]
|
|
row = summary["unresolvedCriticalOrHigh"][0]
|
|
assert row["affectedImages"] == [candidate["images"]["web"]["imageRef"]]
|
|
assert exceptions[0]["approvalStatus"] == "pending"
|
|
assert exceptions[0]["package"] == "libfixture"
|
|
assert exceptions[0]["cve"] == "CVE-2026-12345"
|
|
assert exceptions[0]["affectedImageDigests"] == {
|
|
"web": candidate["images"]["web"]["imageId"]
|
|
}
|
|
assert result["status"] == "pending_review"
|
|
assert not h.state.mutations
|
|
|
|
|
|
def test_findings_for_same_cve_at_distinct_versions_remain_distinct(harness):
|
|
h = harness
|
|
for service, version in (("web", "1.2.3"), ("worker", "4.5.6")):
|
|
path = h.candidate_dir / f"vulnerability-{service}.json"
|
|
report = json.loads(path.read_text())
|
|
report["Results"][0]["Vulnerabilities"] = [
|
|
{
|
|
"PkgName": "libfixture",
|
|
"VulnerabilityID": "CVE-2026-12345",
|
|
"InstalledVersion": version,
|
|
"Severity": "HIGH",
|
|
}
|
|
]
|
|
path.unlink()
|
|
h.write_input(path.name, report)
|
|
h.utility.capture_candidate(h.controller.MCP_HUB_REPOSITORY, h.revision)
|
|
summary = json.loads((h.candidate_dir / "scan-summary.json").read_text())
|
|
assert {row["installedVersion"] for row in summary["unresolvedCriticalOrHigh"]} == {
|
|
"1.2.3",
|
|
"4.5.6",
|
|
}
|
|
|
|
|
|
def test_same_package_cve_version_with_conflicting_severity_fails_closed(harness):
|
|
h = harness
|
|
for service, severity in (("web", "HIGH"), ("worker", "CRITICAL")):
|
|
path = h.candidate_dir / f"vulnerability-{service}.json"
|
|
report = json.loads(path.read_text())
|
|
report["Results"][0]["Vulnerabilities"] = [
|
|
{
|
|
"PkgName": "libfixture",
|
|
"VulnerabilityID": "CVE-2026-12345",
|
|
"InstalledVersion": "1.2.3",
|
|
"Severity": severity,
|
|
}
|
|
]
|
|
path.unlink()
|
|
h.write_input(path.name, report)
|
|
with pytest.raises(h.controller.DeployError, match="conflicting scan severities"):
|
|
h.utility.capture_candidate(h.controller.MCP_HUB_REPOSITORY, h.revision)
|
|
assert not (h.candidate_dir / "candidate.json").exists()
|
|
|
|
|
|
def test_draft_for_all_images_is_rejected_when_finding_scope_is_narrower(harness):
|
|
h = harness
|
|
path = h.candidate_dir / "vulnerability-web.json"
|
|
report = json.loads(path.read_text())
|
|
report["Results"][0]["Vulnerabilities"] = [
|
|
{
|
|
"PkgName": "libfixture",
|
|
"VulnerabilityID": "CVE-2026-12345",
|
|
"InstalledVersion": "1.2.3",
|
|
"Severity": "HIGH",
|
|
}
|
|
]
|
|
path.unlink()
|
|
h.write_input(path.name, report)
|
|
draft_script = h.state.source / "scripts" / "draft_cve_exceptions.py"
|
|
draft_script.write_text(
|
|
draft_script.read_text().replace("if v['tag'] in refs", "if True"),
|
|
encoding="utf-8",
|
|
)
|
|
with pytest.raises(h.controller.DeployError, match="pending-only document"):
|
|
h.utility.capture_candidate(h.controller.MCP_HUB_REPOSITORY, h.revision)
|
|
assert not (h.candidate_dir / "candidate.json").exists()
|
|
|
|
|
|
def test_missing_scan_report_fails_without_publishing_candidate(harness):
|
|
h = harness
|
|
(h.candidate_dir / "secret-web.json").unlink()
|
|
with pytest.raises(h.controller.DeployError):
|
|
h.utility.capture_candidate(h.controller.MCP_HUB_REPOSITORY, h.revision)
|
|
assert not (h.candidate_dir / "candidate.json").exists()
|
|
assert not (h.candidate_dir / "approval.json").exists()
|
|
assert h.state.mutations == []
|
|
|
|
|
|
def test_frozen_compose_drift_and_image_identity_drift_fail_closed(harness):
|
|
h = harness
|
|
h.state.compose_marker = "changed"
|
|
with pytest.raises(h.controller.DeployError, match="Compose"):
|
|
h.utility.capture_candidate(h.controller.MCP_HUB_REPOSITORY, h.revision)
|
|
h.state.compose_marker = "baseline"
|
|
web_ref = h.model["services"]["web"]["image"]
|
|
h.state.images[web_ref]["Id"] = "sha256:" + "f" * 64
|
|
with pytest.raises(h.controller.DeployError):
|
|
h.utility.capture_candidate(h.controller.MCP_HUB_REPOSITORY, h.revision)
|
|
assert not (h.candidate_dir / "candidate.json").exists()
|
|
|
|
|
|
def test_fixable_critical_or_high_finding_is_not_captured(harness):
|
|
h = harness
|
|
path = h.candidate_dir / "vulnerability-web.json"
|
|
report = json.loads(path.read_text())
|
|
report["Results"][0]["Vulnerabilities"] = [
|
|
{
|
|
"PkgName": "libfixture",
|
|
"VulnerabilityID": "CVE-2026-12345",
|
|
"InstalledVersion": "1.2.3",
|
|
"FixedVersion": "1.2.4",
|
|
"Severity": "CRITICAL",
|
|
}
|
|
]
|
|
path.unlink()
|
|
h.write_input(path.name, report)
|
|
with pytest.raises(h.controller.DeployError, match="fixable"):
|
|
h.utility.capture_candidate(h.controller.MCP_HUB_REPOSITORY, h.revision)
|
|
assert not (h.candidate_dir / "candidate.json").exists()
|
|
|
|
|
|
def test_capture_never_overwrites_existing_records(harness):
|
|
h = harness
|
|
h.utility.capture_candidate(h.controller.MCP_HUB_REPOSITORY, h.revision)
|
|
original = (h.candidate_dir / "candidate.json").read_bytes()
|
|
with pytest.raises(h.controller.DeployError):
|
|
h.utility.capture_candidate(h.controller.MCP_HUB_REPOSITORY, h.revision)
|
|
assert (h.candidate_dir / "candidate.json").read_bytes() == original
|
|
assert h.state.mutations == []
|
|
|
|
|
|
def test_capture_preserves_full_supported_trivy_database_metadata(harness):
|
|
h = harness
|
|
path = h.candidate_dir / "trivy-db-metadata.json"
|
|
path.unlink()
|
|
metadata = {
|
|
"UpdatedAt": h.now,
|
|
"DownloadedAt": h.now,
|
|
"Version": 2,
|
|
"NextUpdate": h.now,
|
|
}
|
|
h.write_input(path.name, metadata)
|
|
h.utility.capture_candidate(h.controller.MCP_HUB_REPOSITORY, h.revision)
|
|
candidate = json.loads((h.candidate_dir / "candidate.json").read_text())
|
|
assert (
|
|
candidate["databaseMetadataSha256"]
|
|
== hashlib.sha256(path.read_bytes()).hexdigest()
|
|
)
|
|
assert json.loads(path.read_text()) == metadata
|
|
|
|
|
|
def test_cli_requires_explicit_fixed_target_and_subcommand(modules, monkeypatch):
|
|
_controller, utility = modules
|
|
monkeypatch.setattr(utility, "_is_host_root", lambda: False)
|
|
with pytest.raises(SystemExit):
|
|
utility.main([])
|
|
monkeypatch.setattr(utility, "_is_host_root", lambda: True)
|
|
assert (
|
|
utility.main(["plan", "--repository", "other/repo", "--revision", "a" * 40])
|
|
== 2
|
|
)
|