diff --git a/src/engineering_platform/execution_models.py b/src/engineering_platform/execution_models.py index e6faca9..e394f03 100644 --- a/src/engineering_platform/execution_models.py +++ b/src/engineering_platform/execution_models.py @@ -25,6 +25,7 @@ class PullRequestEvidence: head_branch: str | None = None base_branch: str | None = None merge_state_status: str | None = None + head_sha: str | None = None @dataclass(frozen=True) diff --git a/src/engineering_platform/execution_repository.py b/src/engineering_platform/execution_repository.py index 43435dc..81ba68e 100644 --- a/src/engineering_platform/execution_repository.py +++ b/src/engineering_platform/execution_repository.py @@ -44,6 +44,9 @@ def pull_request_for_head_branch(self, branch: str) -> PullRequestEvidence | Non def ready(self, number: int) -> None: ... def normalize_markdown_body(self, number: int) -> bool: ... def merge(self, number: int) -> None: ... + def create_or_recover_pull_request(self, branch: str, base: str, title: str, body: str) -> PullRequestEvidence: ... + def qualification_for_exact_head(self, number: int, head_sha: str) -> dict[str, object]: ... + def version_preparation_writer(self) -> dict[str, object]: ... class SubprocessRepositoryClient: @@ -192,8 +195,28 @@ def _github(self, *args: str) -> str: scoped = (*args, "--repo", self.repository) if self.repository else args return self.provider.github(*scoped) + def version_preparation_writer(self) -> dict[str, object]: + """Read the configured writer's safe identity and repository scope. + + This is a preflight observation, never a credential issuer or branch + protection bypass. GitHub does not expose a general branch-write + guarantee here, so a protected merge remains GitHub's authority. + """ + if not self.repository or not re.fullmatch(r"[A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+", self.repository): + raise RunnerError("Version preparation writer requires one exact GitHub repository scope.") + try: + identity = json.loads(self.provider.github("api", "user")) + repository = json.loads(self._github("api", f"repos/{self.repository}")) + except (RuntimeError, json.JSONDecodeError) as error: + raise RunnerError("Version preparation writer identity could not be read.") from error + actor = identity.get("login") if isinstance(identity, dict) else None + permissions = repository.get("permissions") if isinstance(repository, dict) else None + if not isinstance(actor, str) or not actor or not isinstance(permissions, dict): + raise RunnerError("Version preparation writer identity is incomplete.") + return {"actor": actor, "repository_id": self.repository, "can_push": permissions.get("push") is True} + def pull_request(self, number: int) -> PullRequestEvidence: - try: raw = json.loads(self._github("pr", "view", str(number), "--json", "number,state,isDraft,mergeCommit,statusCheckRollup,headRefName,baseRefName,mergeStateStatus")) + try: raw = json.loads(self._github("pr", "view", str(number), "--json", "number,state,isDraft,mergeCommit,statusCheckRollup,headRefName,headRefOid,baseRefName,mergeStateStatus")) except RuntimeError as error: raise RunnerError(str(error)) from error # GitHub can append an empty rollup entry to an otherwise completed # merged PR. It is not a check and must not keep terminal evidence in @@ -209,7 +232,7 @@ def pull_request(self, number: int) -> PullRequestEvidence: merge = raw.get("mergeCommit") or {} return PullRequestEvidence( raw["number"], raw["state"], terminal, passed, merge.get("oid"), raw["isDraft"], failed, - raw.get("headRefName"), raw.get("baseRefName"), raw.get("mergeStateStatus"), + raw.get("headRefName"), raw.get("baseRefName"), raw.get("mergeStateStatus"), raw.get("headRefOid"), ) def pull_request_for_head_branch(self, branch: str) -> PullRequestEvidence | None: @@ -226,6 +249,72 @@ def pull_request_for_head_branch(self, branch: str) -> PullRequestEvidence | Non if len(numbers) != 1: raise RunnerError("Finalization recovery found more than one pull request for its checkpointed branch.") return self.pull_request(numbers[0]) + + def create_or_recover_pull_request(self, branch: str, base: str, title: str, body: str) -> PullRequestEvidence: + """Create one bounded PR or recover the sole existing branch identity.""" + existing = self.pull_request_for_head_branch(branch) + if existing is not None: + if existing.base_branch != base: + raise RunnerError("Version preparation branch already has a pull request for another base.") + return existing + try: + raw = self._github("pr", "create", "--head", branch, "--base", base, "--title", title, "--body", body) + except RuntimeError as error: + # A successful create may lose its acknowledgement. Only recover + # the deterministic branch identity; never create a second PR. + recovered = self.pull_request_for_head_branch(branch) + if recovered is None: + raise RunnerError(str(error)) from error + if recovered.base_branch != base: + raise RunnerError("Version preparation PR recovery found wrong base.") from error + return recovered + match = re.search(r"/pull/(\d+)(?:\s|$)", raw) + if match is None: + recovered = self.pull_request_for_head_branch(branch) + if recovered is None: + raise RunnerError("Version preparation PR create acknowledgement is ambiguous.") + return recovered + return self.pull_request(int(match.group(1))) + + def qualification_for_exact_head(self, number: int, head_sha: str) -> dict[str, object]: + """Read the base branch's required checks for this exact candidate.""" + if not re.fullmatch(r"[0-9a-f]{40}", head_sha): + raise RunnerError("Qualification requires an exact candidate SHA.") + try: + raw = json.loads(self._github("pr", "view", str(number), "--json", "headRefOid,baseRefOid,baseRefName,statusCheckRollup")) + except (RuntimeError, json.JSONDecodeError) as error: + raise RunnerError("Version preparation qualification could not be read.") from error + if raw.get("headRefOid") != head_sha: + raise RunnerError("Qualification evidence belongs to a different pull request head.") + base = raw.get("baseRefName") + if not self.repository or not isinstance(base, str) or not base: + raise RunnerError("Version preparation qualification lacks an exact protected base branch.") + try: + required = json.loads(self._github( + "api", f"repos/{self.repository}/branches/{base}/protection/required_status_checks", + )) + except (RuntimeError, json.JSONDecodeError) as error: + raise RunnerError("Version preparation required-check policy could not be read.") from error + contexts = required.get("contexts") if isinstance(required, dict) else None + protected_checks = required.get("checks") if isinstance(required, dict) else None + names = set() + if isinstance(contexts, list): + names.update(item for item in contexts if isinstance(item, str) and item) + if isinstance(protected_checks, list): + names.update(item.get("context") for item in protected_checks if isinstance(item, dict) and isinstance(item.get("context"), str) and item["context"]) + if not names: + raise RunnerError("Version preparation qualification has no required checks configured.") + checks = [item for item in (raw.get("statusCheckRollup") or []) if isinstance(item, dict) and isinstance(item.get("status"), str)] + if not checks or any(item.get("status") != "COMPLETED" for item in checks): + raise RunnerError("Version preparation qualification is incomplete.") + failed = [str(item.get("name") or "unnamed check") for item in checks if item.get("conclusion") not in {"SUCCESS", "NEUTRAL", "SKIPPED"}] + if failed: + raise RunnerError("Version preparation qualification failed: " + ", ".join(failed)) + observed = {item.get("name") for item in checks if isinstance(item.get("name"), str) and item["name"]} + missing = sorted(names - observed) + if missing: + raise RunnerError("Version preparation qualification is missing required checks: " + ", ".join(missing)) + return {"pull_request_id": number, "exact_qualified_sha": head_sha, "base_revision": raw.get("baseRefOid"), "required_checks": sorted(names), "checks": checks, "conclusion": "PASS"} def ready(self, number: int) -> None: try: self._github("pr", "ready", str(number)) except RuntimeError as error: diff --git a/src/engineering_platform/version_preparation_delivery.py b/src/engineering_platform/version_preparation_delivery.py new file mode 100644 index 0000000..476e0e7 --- /dev/null +++ b/src/engineering_platform/version_preparation_delivery.py @@ -0,0 +1,437 @@ +"""Bounded EP execution of a product-owned version preparation operation. + +The product helper owns version rules; this module owns only admission, +isolated candidate construction and allow-listed diff verification. It never +infers a release, runs arbitrary commands, or grants merge authority. +""" +from __future__ import annotations + +from dataclasses import dataclass +import hashlib +import json +import os +from pathlib import Path +import re +from typing import Mapping, Protocol + +from .execution_errors import RunnerError +from .providers import GitProvider +from .execution_repository import GitHubClient + +_SHA = re.compile(r"^[0-9a-f]{40}$") +_SHA256 = re.compile(r"^sha256:[0-9a-f]{64}$") +_SEMVER = re.compile(r"^(?:0|[1-9][0-9]*)\.(?:0|[1-9][0-9]*)\.(?:0|[1-9][0-9]*)$") +_OPERATION = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{7,127}$") +_PATH = re.compile(r"^(?:[A-Za-z0-9][A-Za-z0-9._-]*/)*[A-Za-z0-9][A-Za-z0-9._-]*$") +_REPOSITORY_PATH = re.compile(r"^(?:[A-Za-z0-9.][A-Za-z0-9._-]*/)*[A-Za-z0-9.][A-Za-z0-9._-]*$") +_REQUEST_KEYS = frozenset({"contract_version", "operation_id", "product_id", "component_id", "repository_id", "policy_revision", "policy_digest", "source_event_set", "source_event_policy", "release_class", "release_rationale", "expected_source_revision", "expected_target_branch_revision", "expected_version", "requested_change", "determined_target_version", "allowed_projection_paths", "prepared_operation_digest", "authorization_reference", "delivery_mode"}) +_HELPER_KEYS = frozenset({"contract_version", "product_id", "repository_id", "helper_path", "receipt_directory", "allowed_projection_paths", "policy_revision"}) + + +class VersionPreparationError(RunnerError): + pass + + +@dataclass(frozen=True) +class ProductHelperDeclaration: + product_id: str + repository_id: str + helper_path: str + receipt_directory: str + allowed_projection_paths: tuple[str, ...] + policy_revision: str + + @classmethod + def load(cls, worktree: Path) -> "ProductHelperDeclaration": + try: value = json.loads((worktree / ".version-preparation.json").read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError) as error: raise VersionPreparationError("product version helper declaration is unreadable") from error + if not isinstance(value, dict) or set(value) != _HELPER_KEYS or value.get("contract_version") != "1": + raise VersionPreparationError("product version helper declaration has unknown fields or schema") + fields = ("product_id", "repository_id", "helper_path", "policy_revision") + receipt_directory = value.get("receipt_directory") + if (not all(isinstance(value.get(key), str) and _PATH.fullmatch(value[key]) for key in fields) + or not isinstance(receipt_directory, str) or not _REPOSITORY_PATH.fullmatch(receipt_directory) + or any(part in {".", ".."} for part in receipt_directory.split("/"))): + raise VersionPreparationError("product version helper declaration has invalid paths or identities") + paths = value.get("allowed_projection_paths") + if not isinstance(paths, list) or not paths or not all(isinstance(path, str) and _PATH.fullmatch(path) for path in paths): + raise VersionPreparationError("product version helper declaration has invalid projection paths") + return cls(value["product_id"], value["repository_id"], value["helper_path"], receipt_directory, tuple(paths), value["policy_revision"]) + + +@dataclass(frozen=True) +class VersionPreparationRequest: + contract_version: str + operation_id: str + product_id: str + component_id: str | None + repository_id: str + policy_revision: str + policy_digest: str + source_event_set: tuple[str, ...] + source_event_policy: str + release_class: str + release_rationale: str + expected_source_revision: str + expected_target_branch_revision: str | None + expected_version: str + requested_change: str + determined_target_version: str + allowed_projection_paths: tuple[str, ...] + prepared_operation_digest: str + authorization_reference: str + delivery_mode: str + + @classmethod + def parse(cls, value: object) -> "VersionPreparationRequest": + if not isinstance(value, dict) or set(value) != _REQUEST_KEYS: + raise VersionPreparationError("version preparation request has unknown or missing fields") + def text(key: str, optional: bool = False) -> str | None: + item = value[key] + if optional and item is None: return None + if not isinstance(item, str) or not item or len(item) > 512: raise VersionPreparationError(f"invalid {key}") + return item + events = value["source_event_set"] + paths = value["allowed_projection_paths"] + if (not isinstance(events, list) or not events or len(set(events)) != len(events) + or not all(isinstance(item, str) and item for item in events)): + raise VersionPreparationError("source event set is invalid") + if (not isinstance(paths, list) or not paths or len(set(paths)) != len(paths) + or not all(isinstance(item, str) and _PATH.fullmatch(item) for item in paths)): + raise VersionPreparationError("allowed projection paths are invalid") + request = cls(*(text(key, key in {"component_id", "expected_target_branch_revision"}) for key in ( + "contract_version", "operation_id", "product_id", "component_id", "repository_id", "policy_revision", "policy_digest")), + tuple(events), text("source_event_policy"), text("release_class"), text("release_rationale"), text("expected_source_revision"), + text("expected_target_branch_revision", True), text("expected_version"), text("requested_change"), + text("determined_target_version"), tuple(paths), text("prepared_operation_digest"), + text("authorization_reference"), text("delivery_mode")) + if request.contract_version != "1" or not _OPERATION.fullmatch(request.operation_id): + raise VersionPreparationError("unsupported contract version or operation ID") + if not _SHA.fullmatch(request.expected_source_revision): + raise VersionPreparationError("expected source revision must be an exact SHA") + if request.expected_target_branch_revision is not None and not _SHA.fullmatch(request.expected_target_branch_revision): + raise VersionPreparationError("expected target branch revision must be an exact SHA") + if not _SEMVER.fullmatch(request.expected_version) or not _SEMVER.fullmatch(request.determined_target_version): + raise VersionPreparationError("expected and determined versions must be stable SemVer") + if not _SHA256.fullmatch(request.policy_digest) or not _SHA256.fullmatch(request.prepared_operation_digest): + raise VersionPreparationError("policy and prepared operation digests must be SHA-256 identities") + if request.delivery_mode not in {"EXISTING_FEATURE_CANDIDATE", "PROTECTED_VERSION_PREPARATION_CANDIDATE"}: + raise VersionPreparationError("unsupported delivery mode") + if request.requested_change not in {"patch", "minor", "exact-version"}: + raise VersionPreparationError("unsupported requested change") + if request.release_class not in {"PATCH", "MINOR", "MAJOR", "EXACT", "NO_BUMP"}: + raise VersionPreparationError("unsupported release classification") + expected_class = {"patch": "PATCH", "minor": "MINOR", "exact-version": "EXACT"}[request.requested_change] + if request.release_class != expected_class: + raise VersionPreparationError("release classification does not bind the requested change") + return request + + +class ProductHelper(Protocol): + def apply(self, worktree: Path, request: VersionPreparationRequest) -> None: ... + + +class VersionPreparationDelivery: + """Deterministic candidate builder; publication/merge remain provider authority.""" + def __init__(self, git: GitProvider, helper: ProductHelper) -> None: + self.git, self.helper = git, helper + + def _changed_paths(self, worktree: Path) -> tuple[str, ...]: + """Return every changed path, including the helper's new receipt. + + ``git diff --name-only`` alone omits untracked receipts, which would + let a candidate commit omit its operation evidence. NUL-delimited + Git path inventories avoid whitespace/quote interpretation entirely. + """ + tracked = self.git.command(worktree, "git", "diff", "--no-renames", "--name-only", "-z", "HEAD") + untracked = self.git.command(worktree, "git", "ls-files", "--others", "--exclude-standard", "-z") + paths = [path for path in (tracked + untracked).split("\0") if path] + for path in paths: + if not _REPOSITORY_PATH.fullmatch(path) or any(part in {".", ".."} for part in path.split("/")): + raise VersionPreparationError("version preparation has an invalid changed path") + if len(set(paths)) != len(paths): + raise VersionPreparationError("version preparation has duplicate changed paths") + return tuple(sorted(paths)) + + @staticmethod + def _prepared_candidate_digest( + worktree: Path, changed: tuple[str, ...], request: VersionPreparationRequest, + ) -> str: + """Return a canonical identity for the complete, resulting candidate. + + The identity deliberately covers resulting bytes rather than just a + path inventory. It also repeats the operation facts that select the + product-owned mutation. This is evidence for one prepared candidate, + not a substitute for a Git tree identity after it is committed. + """ + entries: list[dict[str, str]] = [] + for relative in changed: + path = worktree / relative + if path.is_symlink() or not path.is_file(): + raise VersionPreparationError("prepared candidate contains a non-regular projection") + try: + content = path.read_bytes() + except OSError as error: + raise VersionPreparationError("prepared candidate projection is unreadable") from error + entries.append({ + "path": relative, + "resulting_sha256": "sha256:" + hashlib.sha256(content).hexdigest(), + }) + operation = { + "operation_id": request.operation_id, + "product_id": request.product_id, + "component_id": request.component_id, + "repository_id": request.repository_id, + "policy_revision": request.policy_revision, + "policy_digest": request.policy_digest, + "source_event_set": list(request.source_event_set), + "source_event_policy": request.source_event_policy, + "release_class": request.release_class, + "release_rationale": request.release_rationale, + "expected_source_revision": request.expected_source_revision, + "expected_target_branch_revision": request.expected_target_branch_revision, + "expected_version": request.expected_version, + "requested_change": request.requested_change, + "determined_target_version": request.determined_target_version, + "allowed_projection_paths": list(request.allowed_projection_paths), + "authorization_reference": request.authorization_reference, + "delivery_mode": request.delivery_mode, + } + payload = json.dumps( + {"schema_version": 1, "operation": operation, "resulting_paths": entries}, + sort_keys=True, separators=(",", ":"), ensure_ascii=True, + ).encode("ascii") + return "sha256:" + hashlib.sha256(payload).hexdigest() + + @staticmethod + def _validate_prepared_receipt(path: Path, declaration: ProductHelperDeclaration, request: VersionPreparationRequest) -> None: + """Bind the product-owned receipt without imposing one product schema. + + Product helpers deliberately retain their own detailed receipt forms. + EP validates only the shared operation facts required to publish their + output as one bounded candidate. + """ + try: + receipt = json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError, json.JSONDecodeError) as error: + raise VersionPreparationError("prepared operation receipt is unreadable") from error + if not isinstance(receipt, dict): + raise VersionPreparationError("prepared operation receipt must be an object") + if receipt.get("schema_version") not in {1, "1"} or isinstance(receipt.get("schema_version"), bool): + raise VersionPreparationError("prepared operation receipt has an unsupported schema") + expected = { + "operation_id": request.operation_id, + "product": request.product_id, + "component_id": request.component_id, + "repository_id": request.repository_id, + "policy_revision": declaration.policy_revision, + "policy_digest": request.policy_digest, + "source_event_set": list(request.source_event_set), + "source_event_policy": request.source_event_policy, + "release_class": request.release_class, + "release_rationale": request.release_rationale, + "expected_source_revision": request.expected_source_revision, + "expected_target_branch_revision": request.expected_target_branch_revision, + "expected_version": request.expected_version, + "requested_change": request.requested_change, + "determined_target_version": request.determined_target_version, + "allowed_projection_paths": list(request.allowed_projection_paths), + "authorization_reference": request.authorization_reference, + "delivery_mode": request.delivery_mode, + } + if any(receipt.get(key) != value for key, value in expected.items()): + raise VersionPreparationError("prepared operation receipt does not bind the admitted operation") + + def _verify_prepared_candidate( + self, worktree: Path, declaration: ProductHelperDeclaration, request: VersionPreparationRequest, + *, expected_paths: tuple[str, ...] | None = None, + ) -> tuple[tuple[str, ...], str]: + """Re-read every candidate input; never publish a stale preparation.""" + changed = self._changed_paths(worktree) + if expected_paths is not None and changed != expected_paths: + raise VersionPreparationError("prepared candidate changed after verification") + receipt = f"{declaration.receipt_directory}/{request.operation_id}.json" + allowed = set(request.allowed_projection_paths) | {receipt} + if receipt not in changed or set(changed) - allowed: + raise VersionPreparationError("version preparation changed a path outside its declared operation") + self._validate_prepared_receipt(worktree / receipt, declaration, request) + digest = self._prepared_candidate_digest(worktree, changed, request) + if digest != request.prepared_operation_digest: + raise VersionPreparationError("prepared operation digest does not bind the candidate content") + return changed, digest + + def prepare(self, repository: Path, worktree: Path, request: VersionPreparationRequest) -> dict[str, object]: + head = self.git.command(repository, "git", "rev-parse", "HEAD") + if head != request.expected_source_revision: + raise VersionPreparationError("stale expected source revision") + status = self.git.command(repository, "git", "status", "--porcelain", "--untracked-files=all") + if status: + raise VersionPreparationError("repository checkout is not clean") + if self._changed_paths(worktree): + raise VersionPreparationError("isolated version preparation worktree is not clean") + declaration = ProductHelperDeclaration.load(worktree) + if (declaration.product_id != request.product_id or declaration.repository_id != request.repository_id + or declaration.policy_revision != request.policy_revision + or tuple(request.allowed_projection_paths) != declaration.allowed_projection_paths): + raise VersionPreparationError("product helper declaration does not bind the admitted operation") + self.helper.apply(worktree, request) + changed, digest = self._verify_prepared_candidate(worktree, declaration, request) + return {"operation_id": request.operation_id, "changed_paths": changed, "prepared_operation_digest": digest} + + @staticmethod + def branch_name(request: VersionPreparationRequest) -> str: + return f"ep/version-preparation/{request.operation_id}" + + def create_isolated_worktree(self, repository: Path, worktree: Path, request: VersionPreparationRequest) -> None: + """Create the one deterministic candidate worktree from the pinned source.""" + if worktree.exists(): + raise VersionPreparationError("version preparation worktree path already exists") + if self.git.command(repository, "git", "rev-parse", request.expected_source_revision) != request.expected_source_revision: + raise VersionPreparationError("expected source revision is unavailable") + self.git.command(repository, "git", "worktree", "add", "-b", self.branch_name(request), str(worktree), request.expected_source_revision) + if self.git.command(worktree, "git", "rev-parse", "HEAD") != request.expected_source_revision: + raise VersionPreparationError("isolated worktree did not start at the expected source revision") + if self.git.command(worktree, "git", "status", "--porcelain", "--untracked-files=all"): + raise VersionPreparationError("isolated version preparation worktree is not clean") + + def prepare_in_isolated_worktree( + self, repository: Path, worktree: Path, request: VersionPreparationRequest, + ) -> dict[str, object]: + """Create and prepare one candidate without publishing it. + + The worktree is deliberately retained on an apply failure. It has no + commit, push, PR or publication side effect, and retaining it prevents + a retry from silently deriving a new operation from partial files. + """ + self.create_isolated_worktree(repository, worktree, request) + return self.prepare(repository, worktree, request) + + def publish_candidate( + self, worktree: Path, request: VersionPreparationRequest, prepared: Mapping[str, object], github: GitHubClient, + *, base_branch: str, + ) -> dict[str, object]: + """Commit only the verified candidate diff and create/recover one PR.""" + branch = self.branch_name(request) + paths = prepared.get("changed_paths") + if not isinstance(paths, tuple) or not paths or not all(isinstance(path, str) for path in paths): + raise VersionPreparationError("prepared candidate has no bounded changed paths") + if self.git.command(worktree, "git", "branch", "--show-current") != branch: + raise VersionPreparationError("isolated worktree branch does not bind the operation ID") + writer = github.version_preparation_writer() + if (writer.get("repository_id") != request.repository_id + or not isinstance(writer.get("actor"), str) or not writer["actor"] + or writer.get("can_push") is not True): + raise VersionPreparationError("configured GitHub writer is not authorized for this version candidate") + if request.expected_target_branch_revision is not None: + target = self.git.command(worktree, "git", "rev-parse", f"origin/{base_branch}") + if target != request.expected_target_branch_revision: + raise VersionPreparationError("target branch revision changed before candidate publication") + # This is deliberately the final read before staging. A preparation + # is not authority to commit bytes that changed after it was checked. + declared = ProductHelperDeclaration.load(worktree) + if (declared.product_id != request.product_id or declared.repository_id != request.repository_id + or declared.policy_revision != request.policy_revision + or declared.allowed_projection_paths != request.allowed_projection_paths): + raise VersionPreparationError("product helper declaration does not bind the admitted operation") + actual_paths, actual_digest = self._verify_prepared_candidate( + worktree, declared, request, expected_paths=paths, + ) + if actual_digest != prepared.get("prepared_operation_digest"): + raise VersionPreparationError("prepared candidate digest changed after verification") + self.git.command(worktree, "git", "add", "--", *actual_paths) + self.git.command(worktree, "git", "commit", "-m", f"build: prepare version operation {request.operation_id}") + candidate_sha = self.git.command(worktree, "git", "rev-parse", "HEAD") + candidate_tree_sha = self.git.command(worktree, "git", "rev-parse", "HEAD^{tree}") + if not _SHA.fullmatch(candidate_tree_sha): + raise VersionPreparationError("candidate tree identity is unavailable") + self.git.command(worktree, "git", "push", "origin", f"HEAD:{branch}") + body = "\n".join(( + "Bounded EP version-preparation candidate.", + f"operation_id: `{request.operation_id}`", + f"prepared_operation_digest: `{request.prepared_operation_digest}`", + f"expected_source_revision: `{request.expected_source_revision}`", + )) + pr = github.create_or_recover_pull_request(branch, base_branch, f"build: prepare version {request.determined_target_version}", body) + if pr.head_branch != branch: + raise VersionPreparationError("recovered pull request does not bind the candidate branch") + if pr.head_sha != candidate_sha: + raise VersionPreparationError("pull request head changed after version candidate publication") + return { + **prepared, + "candidate_commit_sha": candidate_sha, + "candidate_tree_sha": candidate_tree_sha, + "branch": branch, + "pull_request_id": pr.number, + "pull_request_head_sha": pr.head_sha, + "authorization_reference": request.authorization_reference, + "delivery_mode": request.delivery_mode, + } + + @staticmethod + def qualify_candidate(candidate: Mapping[str, object], github: GitHubClient) -> dict[str, object]: + """Bind read-only repository qualification to the candidate's exact SHA.""" + sha, number = candidate.get("candidate_commit_sha"), candidate.get("pull_request_id") + if not isinstance(sha, str) or not _SHA.fullmatch(sha) or not isinstance(number, int): + raise VersionPreparationError("candidate lacks exact SHA/PR binding") + evidence = github.qualification_for_exact_head(number, sha) + if evidence.get("exact_qualified_sha") != sha or evidence.get("conclusion") != "PASS": + raise VersionPreparationError("candidate qualification is not bound to the exact candidate SHA") + return dict(evidence) + + @staticmethod + def record_delivery_evidence(evidence_root: Path, candidate: Mapping[str, object], qualification: Mapping[str, object]) -> Path: + """Append immutable delivery evidence outside the tracked prepared receipt.""" + operation, sha, tree, branch, pr, digest, authorization = ( + candidate.get(key) for key in ( + "operation_id", "candidate_commit_sha", "candidate_tree_sha", "branch", "pull_request_id", + "prepared_operation_digest", "authorization_reference", + ) + ) + if (not all(isinstance(value, str) and value for value in (operation, sha, tree, branch, digest, authorization)) + or not _SHA.fullmatch(sha) or not _SHA.fullmatch(tree) or not isinstance(pr, int)): + raise VersionPreparationError("candidate is incomplete for delivery evidence") + if qualification.get("exact_qualified_sha") != sha or qualification.get("conclusion") != "PASS": + raise VersionPreparationError("delivery evidence requires exact successful qualification") + payload = { + "schema_version": 1, + "operation_id": operation, + "prepared_operation_digest": digest, + "candidate_commit_sha": sha, + "candidate_tree_sha": tree, + "branch": branch, + "pull_request_id": pr, + "pull_request_head_sha": candidate.get("pull_request_head_sha"), + "qualification": dict(qualification), + "authorization_reference": authorization, + # This adapter never merges. A later authorized delivery route + # may append separate merge evidence; it cannot relabel this as a + # completed protected delivery. + "delivery": {"state": "PENDING_PROTECTED_MERGE"}, + } + encoded = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode() + directory = evidence_root / "version-preparation-delivery" + directory.mkdir(parents=True, exist_ok=True) + target = directory / f"{operation}-{sha}.json" + if target.exists(): + if target.read_bytes() != encoded: + raise VersionPreparationError("delivery evidence identity conflicts with existing bytes") + return target + temporary = target.with_suffix(".tmp") + try: + with temporary.open("xb") as handle: + handle.write(encoded); handle.flush(); os.fsync(handle.fileno()) + os.replace(temporary, target) + except FileExistsError as error: + raise VersionPreparationError("delivery evidence write collided") from error + return target + + def execute( + self, repository: Path, worktree: Path, request: VersionPreparationRequest, github: GitHubClient, + *, base_branch: str, evidence_root: Path, + ) -> dict[str, object]: + """Run one bounded preparation transaction; never merge or publish a release.""" + prepared = self.prepare(repository, worktree, request) + candidate = self.publish_candidate(worktree, request, prepared, github, base_branch=base_branch) + qualification = self.qualify_candidate(candidate, github) + evidence = self.record_delivery_evidence(evidence_root, candidate, qualification) + return {**candidate, "qualification": qualification, "delivery_evidence_path": str(evidence)} diff --git a/tests/engineering/test_execution_host.py b/tests/engineering/test_execution_host.py index bdc2338..ce83411 100644 --- a/tests/engineering/test_execution_host.py +++ b/tests/engineering/test_execution_host.py @@ -1199,6 +1199,119 @@ def github(self, *_: str) -> str: self.assertTrue(evidence.checks_passed) self.assertEqual(evidence.failed_checks, ()) + def test_github_writer_preflight_reads_only_identity_and_scoped_permission(self) -> None: + class Provider: + def __init__(self) -> None: + self.calls: list[tuple[str, ...]] = [] + def github(self, *args: str) -> str: + self.calls.append(args) + if args[:2] == ("api", "user"): + return json.dumps({"login": "ep-delivery-app"}) + if args[:2] == ("api", "repos/pcvantol/forge"): + return json.dumps({"permissions": {"push": True}}) + raise AssertionError(args) + + provider = Provider() + self.assertEqual( + GhCliClient(provider, "pcvantol/forge").version_preparation_writer(), + {"actor": "ep-delivery-app", "repository_id": "pcvantol/forge", "can_push": True}, + ) + self.assertEqual(provider.calls, [("api", "user"), ("api", "repos/pcvantol/forge", "--repo", "pcvantol/forge")]) + + def test_github_writer_preflight_rejects_missing_scope_or_identity(self) -> None: + with self.assertRaisesRegex(RunnerError, "exact GitHub repository"): + GhCliClient(object()).version_preparation_writer() + + class Provider: + def github(self, *args: str) -> str: + return json.dumps({"login": "ep-delivery-app"}) if args[:2] == ("api", "user") else json.dumps({}) + + with self.assertRaisesRegex(RunnerError, "identity is incomplete"): + GhCliClient(Provider(), "pcvantol/forge").version_preparation_writer() + + def test_github_client_recovers_only_one_matching_version_preparation_pr(self) -> None: + evidence = { + "number": 12, "state": "OPEN", "isDraft": False, + "mergeCommit": None, "headRefName": "codex/version-prepare/op", + "headRefOid": "a" * 40, "baseRefName": "main", "mergeStateStatus": "CLEAN", + "statusCheckRollup": [{"name": "validate", "status": "COMPLETED", "conclusion": "SUCCESS"}], + } + + class Provider: + def __init__(self, listing: list[object]) -> None: self.listing, self.calls = listing, [] + def github(self, *args: str) -> str: + self.calls.append(args) + if args[:2] == ("pr", "list"): return json.dumps(self.listing) + if args[:2] == ("pr", "view"): return json.dumps(evidence) + raise AssertionError(args) + + client = GhCliClient(Provider([{"number": 12}])) + recovered = client.create_or_recover_pull_request("codex/version-prepare/op", "main", "title", "body") + self.assertEqual(recovered.number, 12) + self.assertIsNone(GhCliClient(Provider([])).pull_request_for_head_branch("missing")) + with self.assertRaisesRegex(RunnerError, "more than one"): + GhCliClient(Provider([{"number": 12}, {"number": 13}])).pull_request_for_head_branch("ambiguous") + + def test_github_client_create_acknowledgement_and_qualification_are_exact_head_bound(self) -> None: + head = "b" * 40 + evidence = { + "number": 17, "state": "OPEN", "isDraft": False, "mergeCommit": None, + "headRefName": "codex/version-prepare/op", "headRefOid": head, + "baseRefName": "main", "baseRefOid": "c" * 40, "mergeStateStatus": "CLEAN", + "statusCheckRollup": [{"name": "validate", "status": "COMPLETED", "conclusion": "SUCCESS"}], + } + + class Provider: + def github(self, *args: str) -> str: + if args[:2] == ("pr", "list"): return "[]" + if args[:2] == ("pr", "create"): return "https://github.com/pcvantol/forge/pull/17\n" + if args[:2] == ("pr", "view"): return json.dumps(evidence) + if args[:2] == ("api", "repos/pcvantol/forge/branches/main/protection/required_status_checks"): + return json.dumps({"contexts": ["validate"], "checks": []}) + raise AssertionError(args) + + client = GhCliClient(Provider(), "pcvantol/forge") + self.assertEqual(client.create_or_recover_pull_request("codex/version-prepare/op", "main", "title", "body").number, 17) + qualification = client.qualification_for_exact_head(17, head) + self.assertEqual(qualification["exact_qualified_sha"], head) + with self.assertRaisesRegex(RunnerError, "exact candidate SHA"): + client.qualification_for_exact_head(17, "not-a-sha") + evidence["headRefOid"] = "d" * 40 + with self.assertRaisesRegex(RunnerError, "different pull request head"): + client.qualification_for_exact_head(17, head) + + def test_github_client_rejects_incomplete_or_failed_version_preparation_qualification(self) -> None: + head = "e" * 40 + class Provider: + def __init__(self, checks: list[dict[str, str]]) -> None: self.checks = checks + def github(self, *args: str) -> str: + if args[:2] == ("pr", "view"): + return json.dumps({"headRefOid": head, "baseRefOid": "f" * 40, "baseRefName": "main", "statusCheckRollup": self.checks}) + if args[:2] == ("api", "repos/pcvantol/forge/branches/main/protection/required_status_checks"): + return json.dumps({"contexts": ["lint"], "checks": []}) + raise AssertionError(args) + + with self.assertRaisesRegex(RunnerError, "incomplete"): + GhCliClient(Provider([]), "pcvantol/forge").qualification_for_exact_head(1, head) + with self.assertRaisesRegex(RunnerError, "failed: lint"): + GhCliClient(Provider([{"name": "lint", "status": "COMPLETED", "conclusion": "FAILURE"}]), "pcvantol/forge").qualification_for_exact_head(1, head) + + def test_github_client_requires_the_protected_branch_check_set(self) -> None: + head = "e" * 40 + class Provider: + def __init__(self, required: dict[str, object]) -> None: self.required = required + def github(self, *args: str) -> str: + if args[:2] == ("pr", "view"): + return json.dumps({"headRefOid": head, "baseRefOid": "f" * 40, "baseRefName": "main", "statusCheckRollup": [{"name": "unit", "status": "COMPLETED", "conclusion": "SUCCESS"}]}) + if args[:2] == ("api", "repos/pcvantol/forge/branches/main/protection/required_status_checks"): + return json.dumps(self.required) + raise AssertionError(args) + + with self.assertRaisesRegex(RunnerError, "no required checks"): + GhCliClient(Provider({"contexts": [], "checks": []}), "pcvantol/forge").qualification_for_exact_head(1, head) + with self.assertRaisesRegex(RunnerError, "missing required checks: integration"): + GhCliClient(Provider({"contexts": ["unit", "integration"], "checks": []}), "pcvantol/forge").qualification_for_exact_head(1, head) + @patch("engineering_platform.execution_host.subprocess.run") def test_codex_client_handles_valid_review_and_invoke_results(self, run: object) -> None: review_message = json.dumps( diff --git a/tests/engineering/test_version_preparation_delivery.py b/tests/engineering/test_version_preparation_delivery.py new file mode 100644 index 0000000..32dd58e --- /dev/null +++ b/tests/engineering/test_version_preparation_delivery.py @@ -0,0 +1,382 @@ +from __future__ import annotations + +from pathlib import Path +import unittest +import tempfile +import hashlib +import subprocess +import shutil +import sys +from dataclasses import asdict + +from engineering_platform.version_preparation_delivery import ProductHelperDeclaration, VersionPreparationDelivery, VersionPreparationError, VersionPreparationRequest +from engineering_platform.execution_models import PullRequestEvidence + + +def request(**overrides: object) -> dict[str, object]: + value: dict[str, object] = { + "contract_version": "1", "operation_id": "operation-0001", "product_id": "forge", + "component_id": "product", "repository_id": "pcvantol/forge", "policy_revision": "v1", + "policy_digest": "sha256:" + "a" * 64, "source_event_set": ["increment:I-123"], "source_event_policy": "engineering-increment", "release_class": "MINOR", "release_rationale": "capability boundary", + "expected_source_revision": "a" * 40, "expected_target_branch_revision": None, + "expected_version": "2.3.0", "requested_change": "minor", "determined_target_version": "2.4.0", + "allowed_projection_paths": ["product-version.json"], "prepared_operation_digest": "sha256:" + "b" * 64, + "authorization_reference": "grant:bounded", "delivery_mode": "PROTECTED_VERSION_PREPARATION_CANDIDATE", + } + value.update(overrides) + return value + + +def receipt_for(operation: VersionPreparationRequest) -> dict[str, object]: + """The cross-repository receipt fields EP admits and re-checks.""" + return { + "schema_version": 1, + "operation_id": operation.operation_id, + "product": operation.product_id, + "component_id": operation.component_id, + "repository_id": operation.repository_id, + "policy_revision": operation.policy_revision, + "policy_digest": operation.policy_digest, + "source_event_set": list(operation.source_event_set), + "source_event_policy": operation.source_event_policy, + "release_class": operation.release_class, + "release_rationale": operation.release_rationale, + "expected_source_revision": operation.expected_source_revision, + "expected_target_branch_revision": operation.expected_target_branch_revision, + "expected_version": operation.expected_version, + "requested_change": operation.requested_change, + "determined_target_version": operation.determined_target_version, + "allowed_projection_paths": list(operation.allowed_projection_paths), + "authorization_reference": operation.authorization_reference, + "delivery_mode": operation.delivery_mode, + } + + +class VersionPreparationRequestTest(unittest.TestCase): + def test_accepts_exact_bounded_contract(self) -> None: + parsed = VersionPreparationRequest.parse(request()) + self.assertEqual(parsed.operation_id, "operation-0001") + self.assertEqual(parsed.allowed_projection_paths, ("product-version.json",)) + + def test_rejects_unknown_and_untrusted_paths(self) -> None: + with self.assertRaisesRegex(VersionPreparationError, "unknown"): + VersionPreparationRequest.parse({**request(), "shell": "rm"}) + with self.assertRaisesRegex(VersionPreparationError, "paths"): + VersionPreparationRequest.parse(request(allowed_projection_paths=["../outside"])) + + def test_rejects_duplicate_events_and_wrong_source_sha(self) -> None: + with self.assertRaisesRegex(VersionPreparationError, "event"): + VersionPreparationRequest.parse(request(source_event_set=["merge:1", "merge:1"])) + with self.assertRaisesRegex(VersionPreparationError, "exact SHA"): + VersionPreparationRequest.parse(request(expected_source_revision="main")) + with self.assertRaisesRegex(VersionPreparationError, "target branch revision"): + VersionPreparationRequest.parse(request(expected_target_branch_revision="main")) + + def test_rejects_noncanonical_versions_and_digests(self) -> None: + with self.assertRaisesRegex(VersionPreparationError, "stable SemVer"): + VersionPreparationRequest.parse(request(expected_version="02.3.0")) + with self.assertRaisesRegex(VersionPreparationError, "stable SemVer"): + VersionPreparationRequest.parse(request(determined_target_version="2.3.00")) + with self.assertRaisesRegex(VersionPreparationError, "SHA-256"): + VersionPreparationRequest.parse(request(policy_digest="sha256:policy")) + with self.assertRaisesRegex(VersionPreparationError, "SHA-256"): + VersionPreparationRequest.parse(request(prepared_operation_digest="sha256:diff")) + + def test_rejects_a_release_classification_that_disagrees_with_the_operation(self) -> None: + with self.assertRaisesRegex(VersionPreparationError, "classification"): + VersionPreparationRequest.parse(request(release_class="PATCH")) + + def test_rejects_no_bump_from_the_mutating_prepare_adapter(self) -> None: + with self.assertRaisesRegex(VersionPreparationError, "requested change"): + VersionPreparationRequest.parse(request(requested_change="none", release_class="NO_BUMP")) + + def test_candidate_branch_is_deterministically_bound_to_operation(self) -> None: + parsed = VersionPreparationRequest.parse(request()) + self.assertEqual(VersionPreparationDelivery.branch_name(parsed), "ep/version-preparation/operation-0001") + + def test_qualification_cannot_substitute_an_old_head(self) -> None: + class GitHub: + def qualification_for_exact_head(self, number: int, sha: str) -> dict[str, object]: + return {"pull_request_id": number, "exact_qualified_sha": "b" * 40, "conclusion": "PASS"} + with self.assertRaisesRegex(VersionPreparationError, "exact candidate SHA"): + VersionPreparationDelivery.qualify_candidate({"candidate_commit_sha": "a" * 40, "pull_request_id": 9}, GitHub()) + + def test_delivery_evidence_is_idempotent_and_exact_head_bound(self) -> None: + candidate = {"operation_id": "operation-0001", "prepared_operation_digest": "sha256:" + "b" * 64, "candidate_commit_sha": "a" * 40, "candidate_tree_sha": "c" * 40, "branch": "ep/version-preparation/operation-0001", "pull_request_id": 9, "pull_request_head_sha": "a" * 40, "authorization_reference": "grant:bounded"} + qualification = {"exact_qualified_sha": "a" * 40, "conclusion": "PASS", "checks": []} + with tempfile.TemporaryDirectory() as directory: + first = VersionPreparationDelivery.record_delivery_evidence(Path(directory), candidate, qualification) + self.assertEqual(first, VersionPreparationDelivery.record_delivery_evidence(Path(directory), candidate, qualification)) + recorded = __import__("json").loads(first.read_text(encoding="utf-8")) + self.assertEqual(recorded["candidate_tree_sha"], "c" * 40) + self.assertEqual(recorded["delivery"]["state"], "PENDING_PROTECTED_MERGE") + with self.assertRaisesRegex(VersionPreparationError, "exact successful"): + VersionPreparationDelivery.record_delivery_evidence(Path(directory), candidate, {**qualification, "exact_qualified_sha": "b" * 40}) + + def test_execute_binds_prepare_candidate_qualification_and_evidence(self) -> None: + class Git: + def __init__(self) -> None: self.status_calls = 0 + def command(self, _root: Path, *args: str) -> str: + if args[-2:] == ("-z", "HEAD"): + self.status_calls += 1 + return "" if self.status_calls == 1 else "product-version.json\0" + if args[-1] == "-z": return "" if self.status_calls == 1 else ".version-operations/operation-0001.json\0" + if args[-1] == "HEAD^{tree}": return "c" * 40 + if args[-1] == "HEAD": return "a" * 40 + if args[-1] == "--untracked-files=all": return "" + if args[-2:] == ("branch", "--show-current"): return "ep/version-preparation/operation-0001" + return "" + class Helper: + def apply(self, worktree: Path, operation: VersionPreparationRequest) -> None: + (worktree / "product-version.json").write_text('{"version":"2.4.0"}\n', encoding="utf-8") + receipt = worktree / ".version-operations" + receipt.mkdir(exist_ok=True) + (receipt / "operation-0001.json").write_text(__import__("json").dumps(receipt_for(operation)), encoding="utf-8") + class GitHub: + def version_preparation_writer(self) -> dict[str, object]: + return {"actor": "ep-writer", "repository_id": "pcvantol/forge", "can_push": True} + def create_or_recover_pull_request(self, branch: str, base: str, title: str, body: str) -> PullRequestEvidence: + return PullRequestEvidence(9, "OPEN", True, True, head_branch=branch, base_branch=base, head_sha="a" * 40) + def qualification_for_exact_head(self, number: int, sha: str) -> dict[str, object]: + return {"pull_request_id": number, "exact_qualified_sha": sha, "conclusion": "PASS", "checks": []} + with tempfile.TemporaryDirectory() as directory: + Path(directory, ".version-preparation.json").write_text(__import__("json").dumps({"contract_version": "1", "product_id": "forge", "repository_id": "pcvantol/forge", "helper_path": "scripts/advance_product_version.py", "receipt_directory": ".version-operations", "allowed_projection_paths": ["product-version.json"], "policy_revision": "v1"}), encoding="utf-8") + root = Path(directory) + provisional = VersionPreparationRequest.parse(request()) + (root / "product-version.json").write_text('{"version":"2.4.0"}\n', encoding="utf-8") + (root / ".version-operations").mkdir() + (root / ".version-operations" / "operation-0001.json").write_text(__import__("json").dumps(receipt_for(provisional)), encoding="utf-8") + digest = VersionPreparationDelivery(Git(), Helper())._prepared_candidate_digest(root, (".version-operations/operation-0001.json", "product-version.json"), provisional) + (root / "product-version.json").unlink() + (root / ".version-operations" / "operation-0001.json").unlink() + (root / ".version-operations").rmdir() + parsed = VersionPreparationRequest.parse(request(prepared_operation_digest=digest)) + result = VersionPreparationDelivery(Git(), Helper()).execute(root, root, parsed, GitHub(), base_branch="main", evidence_root=root) + self.assertEqual(result["pull_request_id"], 9) + self.assertTrue(Path(str(result["delivery_evidence_path"])).is_file()) + + def test_publish_rejects_a_pull_request_head_that_raced_the_candidate_push(self) -> None: + class Git: + def command(self, _root: Path, *args: str) -> str: + if args[-2:] == ("branch", "--show-current"): return "ep/version-preparation/operation-0001" + if args[-2:] == ("-z", "HEAD"): return "product-version.json\0" + if args[-1] == "-z": return ".version-operations/operation-0001.json\0" + if args[-1] == "HEAD^{tree}": return "c" * 40 + if args[-1] == "HEAD": return "a" * 40 + return "" + class GitHub: + def version_preparation_writer(self) -> dict[str, object]: + return {"actor": "ep-writer", "repository_id": "pcvantol/forge", "can_push": True} + def create_or_recover_pull_request(self, branch: str, base: str, title: str, body: str) -> PullRequestEvidence: + return PullRequestEvidence(9, "OPEN", True, True, head_branch=branch, base_branch=base, head_sha="b" * 40) + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / ".version-preparation.json").write_text(__import__("json").dumps({"contract_version": "1", "product_id": "forge", "repository_id": "pcvantol/forge", "helper_path": "scripts/advance_product_version.py", "receipt_directory": ".version-operations", "allowed_projection_paths": ["product-version.json"], "policy_revision": "v1"}), encoding="utf-8") + provisional = VersionPreparationRequest.parse(request()) + (root / "product-version.json").write_text('{"version":"2.4.0"}\n', encoding="utf-8") + (root / ".version-operations").mkdir() + (root / ".version-operations" / "operation-0001.json").write_text(__import__("json").dumps(receipt_for(provisional)), encoding="utf-8") + digest = VersionPreparationDelivery(Git(), object())._prepared_candidate_digest(root, (".version-operations/operation-0001.json", "product-version.json"), provisional) + parsed = VersionPreparationRequest.parse(request(prepared_operation_digest=digest)) + prepared = {"operation_id": parsed.operation_id, "changed_paths": (".version-operations/operation-0001.json", "product-version.json"), "prepared_operation_digest": digest} + with self.assertRaisesRegex(VersionPreparationError, "head changed"): + VersionPreparationDelivery(Git(), object()).publish_candidate(root, parsed, prepared, GitHub(), base_branch="main") + + def test_publish_rejects_candidate_bytes_changed_after_prepare(self) -> None: + class Git: + def command(self, _root: Path, *args: str) -> str: + if args[-2:] == ("branch", "--show-current"): return "ep/version-preparation/operation-0001" + if args[-2:] == ("-z", "HEAD"): return "product-version.json\0" + if args[-1] == "-z": return ".version-operations/operation-0001.json\0" + if args[1:3] == ("add", "--"): + raise AssertionError("must not stage mutated candidate bytes") + return "" + class GitHub: + def version_preparation_writer(self) -> dict[str, object]: + return {"actor": "ep-writer", "repository_id": "pcvantol/forge", "can_push": True} + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / ".version-preparation.json").write_text(__import__("json").dumps({"contract_version": "1", "product_id": "forge", "repository_id": "pcvantol/forge", "helper_path": "scripts/advance_product_version.py", "receipt_directory": ".version-operations", "allowed_projection_paths": ["product-version.json"], "policy_revision": "v1"}), encoding="utf-8") + provisional = VersionPreparationRequest.parse(request()) + (root / "product-version.json").write_text('{"version":"2.4.0"}\n', encoding="utf-8") + (root / ".version-operations").mkdir() + receipt = root / ".version-operations" / "operation-0001.json" + receipt.write_text(__import__("json").dumps(receipt_for(provisional)), encoding="utf-8") + delivery = VersionPreparationDelivery(Git(), object()) + digest = delivery._prepared_candidate_digest(root, (".version-operations/operation-0001.json", "product-version.json"), provisional) + parsed = VersionPreparationRequest.parse(request(prepared_operation_digest=digest)) + prepared = {"operation_id": parsed.operation_id, "changed_paths": (".version-operations/operation-0001.json", "product-version.json"), "prepared_operation_digest": digest} + (root / "product-version.json").write_text('{"version":"9.9.9"}\n', encoding="utf-8") + with self.assertRaisesRegex(VersionPreparationError, "candidate content"): + delivery.publish_candidate(root, parsed, prepared, GitHub(), base_branch="main") + + def test_publish_rejects_a_writer_without_exact_repository_push_scope(self) -> None: + class Git: + def command(self, _root: Path, *args: str) -> str: + if args[-2:] == ("branch", "--show-current"): return "ep/version-preparation/operation-0001" + raise AssertionError(f"candidate must stop before {args!r}") + class GitHub: + def version_preparation_writer(self) -> dict[str, object]: + return {"actor": "ep-writer", "repository_id": "pcvantol/other", "can_push": True} + parsed = VersionPreparationRequest.parse(request()) + prepared = {"operation_id": parsed.operation_id, "changed_paths": ("product-version.json",), "prepared_operation_digest": parsed.prepared_operation_digest} + with self.assertRaisesRegex(VersionPreparationError, "not authorized"): + VersionPreparationDelivery(Git(), object()).publish_candidate(Path("."), parsed, prepared, GitHub(), base_branch="main") + + def test_publish_rejects_a_target_branch_that_moved_after_admission(self) -> None: + class Git: + def command(self, _root: Path, *args: str) -> str: + if args[-2:] == ("branch", "--show-current"): return "ep/version-preparation/operation-0001" + if args[-1] == "origin/main": return "b" * 40 + raise AssertionError(f"candidate must stop before {args!r}") + class GitHub: + def version_preparation_writer(self) -> dict[str, object]: + return {"actor": "ep-writer", "repository_id": "pcvantol/forge", "can_push": True} + parsed = VersionPreparationRequest.parse(request(expected_target_branch_revision="a" * 40)) + prepared = {"operation_id": parsed.operation_id, "changed_paths": ("product-version.json",), "prepared_operation_digest": parsed.prepared_operation_digest} + with self.assertRaisesRegex(VersionPreparationError, "target branch revision changed"): + VersionPreparationDelivery(Git(), object()).publish_candidate(Path("."), parsed, prepared, GitHub(), base_branch="main") + + def test_existing_worktree_path_is_rejected_before_git_write(self) -> None: + class Git: + def command(self, *_args: str) -> str: raise AssertionError("must not invoke Git") + with tempfile.TemporaryDirectory() as directory: + parsed = VersionPreparationRequest.parse(request()) + with self.assertRaisesRegex(VersionPreparationError, "already exists"): + VersionPreparationDelivery(Git(), object()).create_isolated_worktree(Path(directory), Path(directory), parsed) + + def test_receipt_must_be_in_the_declared_directory(self) -> None: + class Git: + def __init__(self) -> None: self.status_calls = 0 + def command(self, _root: Path, *args: str) -> str: + if args[-2:] == ("-z", "HEAD"): + self.status_calls += 1 + return "" if self.status_calls == 1 else "product-version.json\0" + if args[-1] == "-z": return "" if self.status_calls == 1 else "other/operation-0001.json\0" + if args[-1] == "HEAD": return "a" * 40 + if args[-1] == "--untracked-files=all": return "" + return "" + class Helper: + def apply(self, _worktree: Path, _request: object) -> None: pass + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / ".version-preparation.json").write_text(__import__("json").dumps({"contract_version": "1", "product_id": "forge", "repository_id": "pcvantol/forge", "helper_path": "scripts/advance_product_version.py", "receipt_directory": ".version-operations", "allowed_projection_paths": ["product-version.json"], "policy_revision": "v1"}), encoding="utf-8") + digest = "sha256:" + hashlib.sha256(b"other/operation-0001.json\nproduct-version.json").hexdigest() + with self.assertRaisesRegex(VersionPreparationError, "declared operation"): + VersionPreparationDelivery(Git(), Helper()).prepare(root, root, VersionPreparationRequest.parse(request(prepared_operation_digest=digest))) + + def test_prepared_receipt_must_bind_the_admitted_product_operation(self) -> None: + declaration = ProductHelperDeclaration("forge", "pcvantol/forge", "scripts/advance_product_version.py", ".version-operations", ("product-version.json",), "v1") + with tempfile.TemporaryDirectory() as directory: + receipt = Path(directory, "operation-0001.json") + receipt.write_text(__import__("json").dumps({"schema_version": 1, "operation_id": "operation-0001", "product": "workspace", "policy_revision": "v1", "expected_source_revision": "a" * 40, "allowed_projection_paths": ["product-version.json"]}), encoding="utf-8") + with self.assertRaisesRegex(VersionPreparationError, "does not bind"): + VersionPreparationDelivery._validate_prepared_receipt(receipt, declaration, VersionPreparationRequest.parse(request())) + + def test_prepared_receipt_rejects_changed_version_or_classification(self) -> None: + declaration = ProductHelperDeclaration("forge", "pcvantol/forge", "scripts/advance_product_version.py", ".version-operations", ("product-version.json",), "v1") + parsed = VersionPreparationRequest.parse(request()) + with tempfile.TemporaryDirectory() as directory: + receipt = Path(directory, "operation-0001.json") + receipt.write_text(__import__("json").dumps({"schema_version": 1, "operation_id": parsed.operation_id, "product": parsed.product_id, "policy_revision": parsed.policy_revision, "expected_source_revision": parsed.expected_source_revision, "expected_version": "8.0.0", "requested_change": "exact-version", "release_class": "EXACT", "release_rationale": parsed.release_rationale, "determined_target_version": "9.0.0", "allowed_projection_paths": list(parsed.allowed_projection_paths)}), encoding="utf-8") + with self.assertRaisesRegex(VersionPreparationError, "does not bind"): + VersionPreparationDelivery._validate_prepared_receipt(receipt, declaration, parsed) + + def test_isolated_git_worktree_includes_an_untracked_receipt_in_candidate_scope(self) -> None: + """A real temporary Git checkout proves receipts cannot be omitted by diff.""" + from engineering_platform.providers import GitProvider + + class Helper: + def apply(self, worktree: Path, operation: VersionPreparationRequest) -> None: + (worktree / "product-version.json").write_text('{"version":"2.4.0"}\n', encoding="utf-8") + receipt = worktree / ".version-operations" + receipt.mkdir() + (receipt / "operation-0001.json").write_text(__import__("json").dumps(receipt_for(operation)), encoding="utf-8") + + def git(root: Path, *args: str) -> None: + subprocess.run(("git", *args), cwd=root, check=True, text=True, capture_output=True) + + declaration = {"contract_version": "1", "product_id": "forge", "repository_id": "pcvantol/forge", "helper_path": "scripts/advance_product_version.py", "receipt_directory": ".version-operations", "allowed_projection_paths": ["product-version.json"], "policy_revision": "v1"} + with tempfile.TemporaryDirectory() as directory: + root, candidate = Path(directory, "source"), Path(directory, "candidate") + root.mkdir() + git(root, "init", "-q") + git(root, "config", "user.email", "test@example.invalid") + git(root, "config", "user.name", "Version Preparation Test") + (root / "product-version.json").write_text('{"version":"2.3.0"}\n', encoding="utf-8") + (root / ".version-preparation.json").write_text(__import__("json").dumps(declaration), encoding="utf-8") + git(root, "add", "product-version.json", ".version-preparation.json") + git(root, "commit", "-qm", "baseline") + sha = subprocess.run(("git", "rev-parse", "HEAD"), cwd=root, check=True, text=True, capture_output=True).stdout.strip() + delivery = VersionPreparationDelivery(GitProvider(), Helper()) + provisional = VersionPreparationRequest.parse(request(expected_source_revision=sha)) + planned = Path(directory, "planned") + planned.mkdir() + (planned / "product-version.json").write_text('{"version":"2.4.0"}\n', encoding="utf-8") + (planned / ".version-operations").mkdir() + (planned / ".version-operations" / "operation-0001.json").write_text(__import__("json").dumps(receipt_for(provisional)), encoding="utf-8") + digest = delivery._prepared_candidate_digest(planned, (".version-operations/operation-0001.json", "product-version.json"), provisional) + parsed = VersionPreparationRequest.parse(request(expected_source_revision=sha, prepared_operation_digest=digest)) + prepared = delivery.prepare_in_isolated_worktree(root, candidate, parsed) + self.assertEqual(prepared["changed_paths"], (".version-operations/operation-0001.json", "product-version.json")) + self.assertEqual(GitProvider().command(candidate, "git", "rev-parse", "HEAD"), sha) + + class FailingHelper: + def apply(self, worktree: Path, operation: VersionPreparationRequest) -> None: + (worktree / "product-version.json").write_text('{"version":"2.4.0"}\n', encoding="utf-8") + (worktree / ".version-operations").mkdir() + (worktree / ".version-operations" / f"{operation.operation_id}.json").write_text("not-json", encoding="utf-8") + + failed = Path(directory, "failed-candidate") + failed_request = VersionPreparationRequest.parse(request(operation_id="operation-0002", expected_source_revision=sha, prepared_operation_digest=digest)) + with self.assertRaisesRegex(VersionPreparationError, "receipt is unreadable"): + VersionPreparationDelivery(GitProvider(), FailingHelper()).prepare_in_isolated_worktree(root, failed, failed_request) + self.assertEqual(GitProvider().command(root, "git", "rev-parse", "HEAD"), sha) + self.assertEqual(GitProvider().command(failed, "git", "rev-parse", "HEAD"), sha) + self.assertTrue(GitProvider().command(failed, "git", "status", "--porcelain")) + + def test_external_fixture_product_proves_the_declared_receipt_contract(self) -> None: + """Exercise a real product helper process, not an in-memory fake.""" + from engineering_platform.providers import GitProvider + + class ExternalFixtureHelper: + def apply(self, worktree: Path, operation: VersionPreparationRequest) -> None: + payload = asdict(operation) + payload["source_event_set"] = list(operation.source_event_set) + payload["allowed_projection_paths"] = list(operation.allowed_projection_paths) + subprocess.run( + (sys.executable, str(worktree / "scripts" / "apply_version.py"), str(worktree)), + input=__import__("json").dumps(payload), text=True, check=True, capture_output=True, + ) + + def git(root: Path, *args: str) -> None: + subprocess.run(("git", *args), cwd=root, check=True, text=True, capture_output=True) + + fixture = Path(__file__).parents[1] / "fixtures" / "version_preparation_product" + with tempfile.TemporaryDirectory() as directory: + root, candidate, planned = Path(directory, "source"), Path(directory, "candidate"), Path(directory, "planned") + shutil.copytree(fixture, root) + git(root, "init", "-q") + git(root, "config", "user.email", "test@example.invalid") + git(root, "config", "user.name", "Version Preparation Test") + git(root, "add", ".") + git(root, "commit", "-qm", "fixture baseline") + sha = subprocess.run(("git", "rev-parse", "HEAD"), cwd=root, check=True, text=True, capture_output=True).stdout.strip() + raw = request( + product_id="fixture-product", repository_id="pcvantol/fixture-product", component_id="fixture-product", + policy_revision="fixture-policy-v1", expected_source_revision=sha, + ) + provisional = VersionPreparationRequest.parse(raw) + shutil.copytree(fixture, planned) + ExternalFixtureHelper().apply(planned, provisional) + delivery = VersionPreparationDelivery(GitProvider(), ExternalFixtureHelper()) + digest = delivery._prepared_candidate_digest(planned, (".version-operations/operation-0001.json", "product-version.json"), provisional) + prepared = delivery.prepare_in_isolated_worktree( + root, candidate, VersionPreparationRequest.parse({**raw, "prepared_operation_digest": digest}), + ) + self.assertEqual(prepared["prepared_operation_digest"], digest) + self.assertEqual(__import__("json").loads((candidate / "product-version.json").read_text())["version"], "2.4.0") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/fixtures/version_preparation_product/.version-preparation.json b/tests/fixtures/version_preparation_product/.version-preparation.json new file mode 100644 index 0000000..9273fee --- /dev/null +++ b/tests/fixtures/version_preparation_product/.version-preparation.json @@ -0,0 +1,9 @@ +{ + "contract_version": "1", + "product_id": "fixture-product", + "repository_id": "pcvantol/fixture-product", + "helper_path": "scripts/apply_version.py", + "receipt_directory": ".version-operations", + "allowed_projection_paths": ["product-version.json"], + "policy_revision": "fixture-policy-v1" +} diff --git a/tests/fixtures/version_preparation_product/product-version.json b/tests/fixtures/version_preparation_product/product-version.json new file mode 100644 index 0000000..f94f961 --- /dev/null +++ b/tests/fixtures/version_preparation_product/product-version.json @@ -0,0 +1,4 @@ +{ + "product": "fixture-product", + "version": "2.3.0" +} diff --git a/tests/fixtures/version_preparation_product/scripts/apply_version.py b/tests/fixtures/version_preparation_product/scripts/apply_version.py new file mode 100644 index 0000000..f38b40e --- /dev/null +++ b/tests/fixtures/version_preparation_product/scripts/apply_version.py @@ -0,0 +1,38 @@ +"""Minimal external product helper fixture for the EP delivery contract.""" +from __future__ import annotations + +import json +from pathlib import Path +import sys + + +root = Path(sys.argv[1]) +request = json.loads(sys.stdin.read()) +(root / "product-version.json").write_text( + json.dumps({"product": request["product_id"], "version": request["determined_target_version"]}) + "\n", + encoding="utf-8", +) +receipt = { + "schema_version": 1, + "operation_id": request["operation_id"], + "product": request["product_id"], + "component_id": request["component_id"], + "repository_id": request["repository_id"], + "policy_revision": request["policy_revision"], + "policy_digest": request["policy_digest"], + "source_event_set": request["source_event_set"], + "source_event_policy": request["source_event_policy"], + "release_class": request["release_class"], + "release_rationale": request["release_rationale"], + "expected_source_revision": request["expected_source_revision"], + "expected_target_branch_revision": request["expected_target_branch_revision"], + "expected_version": request["expected_version"], + "requested_change": request["requested_change"], + "determined_target_version": request["determined_target_version"], + "allowed_projection_paths": request["allowed_projection_paths"], + "authorization_reference": request["authorization_reference"], + "delivery_mode": request["delivery_mode"], +} +directory = root / ".version-operations" +directory.mkdir(exist_ok=True) +(directory / f"{request['operation_id']}.json").write_text(json.dumps(receipt) + "\n", encoding="utf-8")