GovOPlaN/govoplan
1
0
Fork 0
Code Issues 10 Pull Requests Actions 3 Packages Projects Releases Wiki Activity

fix(release): harden review gate diagnostics

Browse Source
This commit is contained in:
Albrecht Degering
2026-07-22 16:52:27 +02:00
parent 36f662c56f
commit 72f697c341
4 changed files with 409 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

183
tests/test_capability_fit_review.py
View File

@@ -6,8 +6,10 @@ from datetime import UTC, datetime, timedelta
import hashlib
import importlib.util
import json
import os
from pathlib import Path
import sys
import tempfile
from unittest import mock
import unittest
@@ -17,8 +19,10 @@ from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
META_ROOT = Path(__file__).resolve().parents[1]
ASSESSMENT_TOOLS_ROOT = META_ROOT / "tools" / "assessments"
if str(ASSESSMENT_TOOLS_ROOT) not in sys.path:
sys.path.insert(0, str(ASSESSMENT_TOOLS_ROOT))
RELEASE_TOOLS_ROOT = META_ROOT / "tools" / "release"
for tools_root in (ASSESSMENT_TOOLS_ROOT, RELEASE_TOOLS_ROOT):
if str(tools_root) not in sys.path:
sys.path.insert(0, str(tools_root))
from govoplan_assessment.capability_fit import ( # noqa: E402
local_tag_provenance,
@@ -87,6 +91,156 @@ class CapabilityFitReviewTests(unittest.TestCase):
"composition_version_changed", {item["code"] for item in report["findings"]}
)
def test_contradictory_signed_selected_unit_metadata_blocks_rerun(self) -> None:
core = next(
item
for item in self.assessment["composition"]
if item["module_id"] == "core"
)
catalog, keyring = signed_catalog(
self.assessment,
selected_units=[
{
"repo": core["repository"],
"version": core["manifest_version"],
"tag": "v999.0.0",
"commit_sha": "0" * 40,
"tag_object_sha": "1" * 40,
}
],
)
report = review_capability_fit(
assessment=deepcopy(self.assessment),
schema=self.schema,
catalog=catalog,
keyring=keyring,
)
self.assertEqual("blocked", report["status"])
codes = {item["code"] for item in report["findings"]}
self.assertIn("catalog_release_metadata", codes)
self.assertIn("composition_commit_changed", codes)
self.assertTrue(report["proof_scope"]["catalog_signature_and_keyring"]["valid"])
self.assertFalse(report["proof_scope"]["release_metadata"]["valid"])
def test_signed_selected_unit_commit_drift_requires_review(self) -> None:
core = next(
item
for item in self.assessment["composition"]
if item["module_id"] == "core"
)
catalog, keyring = signed_catalog(
self.assessment,
selected_units=[
{
"repo": core["repository"],
"version": core["manifest_version"],
"tag": f"v{core['manifest_version']}",
"commit_sha": "0" * 40,
"tag_object_sha": "1" * 40,
}
],
)
report = review_capability_fit(
assessment=deepcopy(self.assessment),
schema=self.schema,
catalog=catalog,
keyring=keyring,
)
self.assertEqual("review_required", report["status"])
self.assertIn(
"composition_commit_changed", {item["code"] for item in report["findings"]}
)
self.assertIn(
"composition.core", {item["id"] for item in report["review_targets"]}
)
self.assertFalse(report["proof_scope"]["release_metadata"]["valid"])
def test_missing_duplicate_and_malformed_selected_unit_provenance_blocks(
self,
) -> None:
core = next(
item
for item in self.assessment["composition"]
if item["module_id"] == "core"
)
valid_unit = {
"repo": core["repository"],
"version": core["manifest_version"],
"tag": f"v{core['manifest_version']}",
"commit_sha": (str(core["commit"]) + "0" * 40)[:40],
"tag_object_sha": "1" * 40,
}
cases = (
("missing", None, False, "catalog_source_provenance"),
(
"duplicate",
[valid_unit, deepcopy(valid_unit)],
True,
"catalog_release_metadata",
),
(
"malformed",
[{**valid_unit, "commit_sha": "not-a-git-object"}],
True,
"catalog_source_provenance",
),
)
for label, selected_units, include_selected_units, expected_code in cases:
with self.subTest(label=label):
catalog, keyring = signed_catalog(
self.assessment,
selected_units=selected_units,
include_selected_units=include_selected_units,
)
report = review_capability_fit(
assessment=deepcopy(self.assessment),
schema=self.schema,
catalog=catalog,
keyring=keyring,
)
self.assertEqual("blocked", report["status"])
self.assertIn(
expected_code, {item["code"] for item in report["findings"]}
)
self.assertFalse(report["proof_scope"]["release_metadata"]["valid"])
def test_catalog_validation_ignores_and_preserves_installer_replay_state(
self,
) -> None:
catalog, keyring = signed_catalog(self.assessment)
with tempfile.TemporaryDirectory() as temp_dir:
state_path = Path(temp_dir) / "catalog-sequence.json"
state_path.write_text(
json.dumps(
{"channels": {"stable": {"last_sequence": catalog["sequence"]}}}
),
encoding="utf-8",
)
replay_environment = {
"GOVOPLAN_MODULE_PACKAGE_CATALOG_SEQUENCE_STATE": str(state_path),
"GOVOPLAN_MODULE_PACKAGE_CATALOG_ENFORCE_SEQUENCE": "true",
}
with mock.patch.dict(os.environ, replay_environment, clear=False):
report = review_capability_fit(
assessment=deepcopy(self.assessment),
schema=self.schema,
catalog=catalog,
keyring=keyring,
)
self.assertEqual(
replay_environment,
{key: os.environ[key] for key in replay_environment},
)
self.assertEqual("current", report["status"])
self.assertTrue(report["proof_scope"]["catalog_signature_and_keyring"]["valid"])
def test_untrusted_or_substituted_keyring_blocks_rerun(self) -> None:
catalog, _ = signed_catalog(self.assessment)
_, other_keyring = signed_catalog(self.assessment)
@@ -186,6 +340,8 @@ def signed_catalog(
*,
versions: dict[str, str] | None = None,
sequence: int = 202607220843,
selected_units: list[dict[str, object]] | None = None,
include_selected_units: bool = True,
) -> tuple[dict[str, object], dict[str, object]]:
versions = versions or {}
private_key = Ed25519PrivateKey.generate()
@@ -220,6 +376,27 @@ def signed_catalog(
else:
modules.append(entry)
assert core_release is not None
if selected_units is None:
selected_units = [
{
"repo": component["repository"],
"version": versions.get(
str(component["module_id"]), str(component["manifest_version"])
),
"tag": "v"
+ versions.get(
str(component["module_id"]), str(component["manifest_version"])
),
"commit_sha": (str(component["commit"]) + "0" * 40)[:40],
"tag_object_sha": hashlib.sha256(
f"{component['repository']}:{component['manifest_version']}:tag".encode()
).hexdigest()[:40],
}
for component in assessment["composition"]
]
release: dict[str, object] = {"keyring_sha256": canonical_hash(keyring)}
if include_selected_units:
release["selected_units"] = selected_units
catalog: dict[str, object] = {
"catalog_version": "1",
"channel": "stable",
@@ -228,7 +405,7 @@ def signed_catalog(
"expires_at": (now + timedelta(days=30)).isoformat().replace("+00:00", "Z"),
"core_release": core_release,
"modules": modules,
"release": {"keyring_sha256": canonical_hash(keyring)},
"release": release,
}
signature_payload = json.dumps(
catalog, sort_keys=True, separators=(",", ":"), ensure_ascii=False

34
tests/test_release_plan_guidance.py
View File

@@ -74,6 +74,40 @@ class ReleasePlanGuidanceTests(unittest.TestCase):
self.assertEqual("preview_source_release", plan.recommended_action.id)
self.assertIn("Preview Tag + Publish", plan.recommended_action.remediation)
def test_malformed_version_metadata_is_a_structured_blocker(self) -> None:
cases = (
("package.json", "{not-json\n", "JSONDecodeError"),
("pyproject.toml", "[project\n", "TOMLDecodeError"),
)
for relative_path, malformed, error_name in cases:
with self.subTest(relative_path=relative_path):
with tempfile.TemporaryDirectory() as temp_dir:
workspace = Path(temp_dir)
repo_path = workspace / "govoplan-files"
repo_path.mkdir()
(repo_path / "pyproject.toml").write_text(
'[project]\nname = "govoplan-files"\nversion = "1.2.4"\n',
encoding="utf-8",
)
(repo_path / relative_path).write_text(malformed, encoding="utf-8")
plan = build_selective_release_plan(
dashboard(workspace=workspace, version="1.2.4"),
selected_repos=("govoplan-files",),
repo_versions={"govoplan-files": "1.2.4"},
)
self.assertEqual("blocked", plan.status)
finding = next(
item
for item in plan.gate_findings
if item.code == "repository_version_metadata_unreadable"
)
self.assertEqual("govoplan-files", finding.repo)
self.assertIn(error_name, finding.message)
self.assertIn("Repair the malformed TOML or JSON", finding.remediation)
self.assertEqual("resolve_release_gate", plan.recommended_action.id)
def test_webui_renders_recommendation_and_remediation(self) -> None:
webui = (RELEASE_ROOT / "webui" / "index.html").read_text(encoding="utf-8")

75
tools/assessments/govoplan_assessment/capability_fit.py
View File

@@ -16,11 +16,14 @@ import re
import subprocess
import tempfile
from typing import Any, Iterable
from unittest import mock
from jsonschema import Draft202012Validator, FormatChecker
from jsonschema.exceptions import SchemaError
from govoplan_core.core.module_package_catalog import validate_module_package_catalog
import govoplan_core.core.module_package_catalog as module_package_catalog
from govoplan_release.source_provenance import catalog_source_selection
from govoplan_release.version_alignment import candidate_catalog_version_issues
RELEASE_REF_PATTERN = re.compile(
@@ -74,6 +77,28 @@ def review_capability_fit(
("assessment.release",),
)
)
findings.extend(
Finding(
"blocker",
"catalog_release_metadata",
(
f"{issue.source}: {issue.message}; expected {issue.expected!r}, "
f"found {issue.actual!r}."
),
("assessment.release",),
)
for issue in candidate_catalog_version_issues(catalog)
)
source_selection = catalog_source_selection(catalog)
findings.extend(
Finding(
"blocker",
"catalog_source_provenance",
issue.message,
("assessment.release",),
)
for issue in source_selection.issues
)
expected_keyring_hash = _catalog_keyring_hash(catalog)
actual_keyring_hash = canonical_hash(keyring)
if expected_keyring_hash is None:
@@ -208,6 +233,37 @@ def review_capability_fit(
)
)
selected_version = source_selection.selected_versions.get(repository)
selected_commit = source_selection.selected_commits.get(repository)
assessed_commit = str(module["commit"]).lower()
if (
selected_version == assessed_version
and selected_commit is not None
and not selected_commit.lower().startswith(assessed_commit)
):
changes.append(
{
"module_id": module_id,
"repository": repository,
"kind": "commit_changed",
"assessed_commit": module["commit"],
"catalog_commit": selected_commit,
}
)
changed_repositories.add(repository)
findings.append(
Finding(
"review",
"composition_commit_changed",
(
f"Signed selected-unit provenance for {repository!r} at "
f"{assessed_version!r} points to commit {selected_commit[:12]}, "
f"not assessed commit {module['commit']}."
),
(f"composition.{module_id}",),
)
)
expected_tag = f"v{catalog_version}" if catalog_version else None
catalog_tags = entry["tags"]
if (
@@ -310,7 +366,22 @@ def validate_catalog(
) as handle:
json.dump(catalog, handle)
handle.flush()
return validate_module_package_catalog(
# Installer replay state answers whether a catalog may be accepted again
# by one runtime. This read-only assessment instead compares explicit
# inputs, so ambient installer state must not affect its result.
with (
mock.patch.object(
module_package_catalog,
"_configured_sequence_state_path",
return_value=None,
),
mock.patch.object(
module_package_catalog,
"_configured_enforce_sequence",
return_value=False,
),
):
return module_package_catalog.validate_module_package_catalog(
Path(handle.name),
require_trusted=True,
approved_channels=(channel,) if channel else (),

127
tools/release/govoplan_release/selective_planner.py
View File

@@ -39,18 +39,29 @@ def build_selective_release_plan(
repositories = selected_repositories(dashboard, selected_repos=selected_repos)
contracts_by_repo = {contract.repo: contract for contract in dashboard.contracts}
units = tuple(
build_unit(repo, target_version=repo_versions.get(repo.spec.name) or target_version, contracts=contracts_by_repo.get(repo.spec.name))
build_unit(
repo,
target_version=repo_versions.get(repo.spec.name) or target_version,
contracts=contracts_by_repo.get(repo.spec.name),
)
for repo in repositories
)
units = apply_repository_version_gate(units, workspace=Path(dashboard.workspace_root))
units = apply_release_webui_bundle_gate(units, workspace=Path(dashboard.workspace_root))
units = apply_repository_version_gate(
units, workspace=Path(dashboard.workspace_root)
)
units = apply_release_webui_bundle_gate(
units, workspace=Path(dashboard.workspace_root)
)
compatibility = compatibility_issues(dashboard)
steps = dry_run_steps(units=units, dashboard=dashboard, channel=channel)
notes = release_notes(dashboard)
status = plan_status(units=units, compatibility=compatibility)
findings = tuple(finding for unit in units for finding in unit.gate_findings)
return SelectiveReleasePlan(
generated_at=datetime.now(tz=UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z"),
generated_at=datetime.now(tz=UTC)
.replace(microsecond=0)
.isoformat()
.replace("+00:00", "Z"),
target_channel=channel,
status=status,
units=units,
@@ -63,7 +74,8 @@ def build_selective_release_plan(
findings=findings,
compatibility=compatibility,
),
source_preflight_ready=bool(units) and all(unit.source_preflight_ready for unit in units),
source_preflight_ready=bool(units)
and all(unit.source_preflight_ready for unit in units),
)
@@ -73,10 +85,33 @@ def apply_repository_version_gate(
workspace: Path,
) -> tuple[ReleasePlanUnit, ...]:
issues_by_repo: dict[str, list[ReleaseGateFinding]] = {}
for issue in selected_repository_version_issues(
repo_versions={unit.repo: unit.target_version for unit in units},
for unit in units:
try:
issues = selected_repository_version_issues(
repo_versions={unit.repo: unit.target_version},
workspace=workspace,
):
)
except Exception as exc: # noqa: BLE001 - source errors become release findings.
issues_by_repo.setdefault(unit.repo, []).append(
ReleaseGateFinding(
code="repository_version_metadata_unreadable",
severity="blocker",
message=(
"Version-bearing source metadata could not be read or parsed "
f"({type(exc).__name__})."
),
remediation=(
"Repair the malformed TOML or JSON version metadata, refresh the "
"repository dashboard, then rebuild this plan."
),
repo=unit.repo,
source="repository version metadata",
expected=unit.target_version,
actual="unreadable",
)
)
continue
for issue in issues:
issues_by_repo.setdefault(issue.repo, []).append(
ReleaseGateFinding(
code="repository_version_alignment",
@@ -158,18 +193,31 @@ def apply_release_webui_bundle_gate(
)
def selected_repositories(dashboard: ReleaseDashboard, *, selected_repos: tuple[str, ...]) -> tuple[RepositorySnapshot, ...]:
def selected_repositories(
dashboard: ReleaseDashboard, *, selected_repos: tuple[str, ...]
) -> tuple[RepositorySnapshot, ...]:
if selected_repos:
wanted = set(selected_repos)
return tuple(repo for repo in dashboard.repositories if repo.spec.name in wanted)
return tuple(
repo for repo in dashboard.repositories if repo.spec.name in wanted
)
return tuple(repo for repo in dashboard.repositories if release_candidate(repo))
def release_candidate(repo: RepositorySnapshot) -> bool:
return repo.dirty or bool(repo.ahead) or (repo.exists and repo.is_git and not repo.has_head)
return (
repo.dirty
or bool(repo.ahead)
or (repo.exists and repo.is_git and not repo.has_head)
)
def build_unit(repo: RepositorySnapshot, *, target_version: str | None, contracts: ModuleContractSnapshot | None) -> ReleasePlanUnit:
def build_unit(
repo: RepositorySnapshot,
*,
target_version: str | None,
contracts: ModuleContractSnapshot | None,
) -> ReleasePlanUnit:
current_version = repo.versions.primary
resolved_target = (target_version or current_version or "0.1.0").removeprefix("v")
target_tag = f"v{resolved_target}"
@@ -238,7 +286,9 @@ def build_unit(repo: RepositorySnapshot, *, target_version: str | None, contract
if value
)
if len(set(local_versions)) > 1:
blockers.append("backend, manifest, and frontend version metadata is not aligned")
blockers.append(
"backend, manifest, and frontend version metadata is not aligned"
)
findings.append(
gate_finding(
repo=repo.spec.name,
@@ -280,7 +330,9 @@ def build_unit(repo: RepositorySnapshot, *, target_version: str | None, contract
)
)
if current_version is None:
warnings.append("no local version metadata found; initial release needs version metadata and catalog entry synthesis")
warnings.append(
"no local version metadata found; initial release needs version metadata and catalog entry synthesis"
)
findings.append(
gate_finding(
repo=repo.spec.name,
@@ -290,7 +342,11 @@ def build_unit(repo: RepositorySnapshot, *, target_version: str | None, contract
remediation=f"Add the required package and manifest version declarations for {resolved_target!r}, then rebuild the plan.",
)
)
if current_version and current_version == resolved_target and repo.local_target_tag_exists:
if (
current_version
and current_version == resolved_target
and repo.local_target_tag_exists
):
warnings.append(f"local tag {target_tag} already exists")
provides: tuple[InterfaceProviderSnapshot, ...] = ()
@@ -398,7 +454,9 @@ def compatibility_issues(dashboard: ReleaseDashboard) -> tuple[CompatibilityIssu
return validate_contracts(dashboard.contracts)
def dry_run_steps(*, units: tuple[ReleasePlanUnit, ...], dashboard: ReleaseDashboard, channel: str) -> tuple[ReleasePlanStep, ...]:
def dry_run_steps(
*, units: tuple[ReleasePlanUnit, ...], dashboard: ReleaseDashboard, channel: str
) -> tuple[ReleasePlanStep, ...]:
steps: list[ReleasePlanStep] = []
snapshots = {repo.spec.name: repo for repo in dashboard.repositories}
for unit in units:
@@ -407,7 +465,8 @@ def dry_run_steps(*, units: tuple[ReleasePlanUnit, ...], dashboard: ReleaseDashb
id=f"{unit.repo}:preflight",
title=f"Inspect {unit.repo}",
detail="Verify local branch, dirty state, current version, and target tag before release.",
command="git status --short --branch && git tag --list " + shlex.quote(unit.target_tag),
command="git status --short --branch && git tag --list "
+ shlex.quote(unit.target_tag),
cwd=f"{dashboard.workspace_root}/{unit.repo}",
repo=unit.repo,
)
@@ -426,7 +485,9 @@ def dry_run_steps(*, units: tuple[ReleasePlanUnit, ...], dashboard: ReleaseDashb
)
)
snapshot = snapshots.get(unit.repo)
if unit.current_version != unit.target_version or (snapshot is not None and snapshot.dirty):
if unit.current_version != unit.target_version or (
snapshot is not None and snapshot.dirty
):
steps.append(
ReleasePlanStep(
id=f"{unit.repo}:commit",
@@ -492,10 +553,10 @@ def dry_run_steps(*, units: tuple[ReleasePlanUnit, ...], dashboard: ReleaseDashb
"and updates only selected release units."
),
command=(
"KEY_DIR=\"$HOME/.config/govoplan/release-keys\" "
'KEY_DIR="$HOME/.config/govoplan/release-keys" '
f"tools/release/release-catalog.py selective --channel {shlex.quote(channel)} "
f"{repo_version_args} "
"--catalog-signing-key \"release-key-1=$KEY_DIR/release-key-1.pem\""
'--catalog-signing-key "release-key-1=$KEY_DIR/release-key-1.pem"'
),
cwd=dashboard.meta_root,
mutating=True,
@@ -512,7 +573,7 @@ def dry_run_steps(*, units: tuple[ReleasePlanUnit, ...], dashboard: ReleaseDashb
),
command=(
"tools/release/release-catalog.py publish-candidate "
f"--candidate-dir \"$CANDIDATE_DIR\" --channel {shlex.quote(channel)}"
f'--candidate-dir "$CANDIDATE_DIR" --channel {shlex.quote(channel)}'
),
cwd=dashboard.meta_root,
status="planned",
@@ -528,12 +589,18 @@ def release_notes(dashboard: ReleaseDashboard) -> tuple[str, ...]:
"Compatibility is checked from currently discovered manifest interface providers and requirements.",
]
if dashboard.catalog.public_checked:
notes.append("Local catalog/keyring hashes are compared with the published channel and keyring.")
notes.append(
"Local catalog/keyring hashes are compared with the published channel and keyring."
)
return tuple(notes)
def plan_status(*, units: tuple[ReleasePlanUnit, ...], compatibility: tuple[CompatibilityIssue, ...]) -> str:
if any(unit.blockers for unit in units) or any(issue.severity == "blocker" for issue in compatibility):
def plan_status(
*, units: tuple[ReleasePlanUnit, ...], compatibility: tuple[CompatibilityIssue, ...]
) -> str:
if any(unit.blockers for unit in units) or any(
issue.severity == "blocker" for issue in compatibility
):
return "blocked"
if units or compatibility:
return "attention"
@@ -544,9 +611,17 @@ def version_satisfies(version: str, requirement: InterfaceRequirementSnapshot) -
parsed = parse_version(version)
if parsed is None:
return False
if requirement.version_min and (minimum := parse_version(requirement.version_min)) is not None and parsed < minimum:
if (
requirement.version_min
and (minimum := parse_version(requirement.version_min)) is not None
and parsed < minimum
):
return False
if requirement.version_max_exclusive and (maximum := parse_version(requirement.version_max_exclusive)) is not None and parsed >= maximum:
if (
requirement.version_max_exclusive
and (maximum := parse_version(requirement.version_max_exclusive)) is not None
and parsed >= maximum
):
return False
return True