diff --git a/tests/test_capability_fit_review.py b/tests/test_capability_fit_review.py index 72089cf..3d57c64 100644 --- a/tests/test_capability_fit_review.py +++ b/tests/test_capability_fit_review.py @@ -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 diff --git a/tests/test_release_plan_guidance.py b/tests/test_release_plan_guidance.py index 3ce6625..1dce283 100644 --- a/tests/test_release_plan_guidance.py +++ b/tests/test_release_plan_guidance.py @@ -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") diff --git a/tools/assessments/govoplan_assessment/capability_fit.py b/tools/assessments/govoplan_assessment/capability_fit.py index 711a5f0..02c2eea 100644 --- a/tools/assessments/govoplan_assessment/capability_fit.py +++ b/tools/assessments/govoplan_assessment/capability_fit.py @@ -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,12 +366,27 @@ def validate_catalog( ) as handle: json.dump(catalog, handle) handle.flush() - return validate_module_package_catalog( - Path(handle.name), - require_trusted=True, - approved_channels=(channel,) if channel else (), - trusted_keys=trusted_keys, - ) + # 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 (), + trusted_keys=trusted_keys, + ) def trusted_keys_from_keyring(payload: dict[str, Any]) -> dict[str, str]: diff --git a/tools/release/govoplan_release/selective_planner.py b/tools/release/govoplan_release/selective_planner.py index c4b7749..ca3044c 100644 --- a/tools/release/govoplan_release/selective_planner.py +++ b/tools/release/govoplan_release/selective_planner.py @@ -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,26 +85,49 @@ 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}, - workspace=workspace, - ): - issues_by_repo.setdefault(issue.repo, []).append( - ReleaseGateFinding( - code="repository_version_alignment", - severity="blocker", - message=issue.message, - remediation=( - f"Align {issue.source} and every other version-bearing source declaration to " - f"{issue.expected!r}. Regenerate lockfiles instead of editing resolved entries by hand, " - "commit the reviewed changes, then rebuild this plan." - ), - repo=issue.repo, - source=issue.source, - expected=issue.expected, - actual=issue.actual, + 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", + severity="blocker", + message=issue.message, + remediation=( + f"Align {issue.source} and every other version-bearing source declaration to " + f"{issue.expected!r}. Regenerate lockfiles instead of editing resolved entries by hand, " + "commit the reviewed changes, then rebuild this plan." + ), + repo=issue.repo, + source=issue.source, + expected=issue.expected, + actual=issue.actual, + ) ) - ) return tuple( replace( unit, @@ -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