Archived
477 lines
25 KiB
Python
477 lines
25 KiB
Python
"""The ordinary development lifecycle: a pull request verifies once, an unchanged-tree merge reuses those bytes, and the publisher selects them.
|
|
|
|
Every step is the real command the workflow runs (`prepare`, `reuse`, `verify`, `package`, `report`) driven through `main`, against a real
|
|
git history, real packaging and a synthetic forge. Only the native build tool itself is replaced by a recorder, so the tests can prove that
|
|
a reused run starts NO compile, test or package, and that any change to the source, platform revision or runtime performs one real verification.
|
|
They prove control flow and refusals, not live forge, registry or Gitea Actions behavior.
|
|
"""
|
|
|
|
import contextlib
|
|
import io
|
|
import json
|
|
import re
|
|
import unittest
|
|
from dataclasses import dataclass, field
|
|
from pathlib import Path
|
|
from typing import cast, override
|
|
from unittest import mock
|
|
|
|
from platform_ci.__main__ import main
|
|
from platform_ci.delivery.errors import ReuseError
|
|
from platform_ci.packaging import collect_outputs
|
|
from platform_ci.plan import Target
|
|
from platform_ci.runtimes import RuntimeCatalog, parse_runtime_catalog
|
|
from tests.support import NODE_IMAGE, PLATFORM_REVISION
|
|
from tests.synthetic_services import CALLER_FILE, CALLER_TEXT, CUSTOM_CALLER_TEXT, REPOSITORY, REPOSITORY_ID
|
|
from tests.test_delivery_publisher import JAR, PublisherTestCase
|
|
|
|
OTHER_PLATFORM_REVISION = "c" * 40
|
|
OTHER_RUNTIME_CATALOG = parse_runtime_catalog({"schema": 1, "toolchains": {"java": {"21": {"image": f"registry.example/other-java@sha256:{'9' * 64}"}}, "node": {"22.20.0": {"image": NODE_IMAGE}}}})
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ScopedRun:
|
|
"""A run of the canonical SCOPED workflow: the server-recorded source id and source commit, and what the job itself is told."""
|
|
|
|
source: int = 3
|
|
workflow_sha: str = "a" * 40
|
|
file: str = "pipeline.yaml"
|
|
env_source_id: str = "3"
|
|
env_workflow_sha: str = ""
|
|
|
|
|
|
@dataclass
|
|
class PipelineResult:
|
|
"""What one run of the shared pipeline did, observed from the commands' own outputs."""
|
|
|
|
plan: dict[str, object]
|
|
outputs: dict[str, str]
|
|
report_code: int
|
|
report_text: str
|
|
builds: int
|
|
packages: int
|
|
evidence: str = ""
|
|
note: str = ""
|
|
built_targets: list[str] = field(default_factory=list[str])
|
|
|
|
|
|
class LifecycleTest(PublisherTestCase):
|
|
"""Drive the real pipeline commands for successive commits and observe what compiled, tested and packaged."""
|
|
|
|
@override
|
|
def setUp(self) -> None:
|
|
"""Create the synthetic world and the recorders that stand in for the native build tool."""
|
|
super().setUp()
|
|
self.builds = 0
|
|
self.packages = 0
|
|
self._checkouts = 0
|
|
self.forge_artifacts: list[tuple[int, str, Path]] = []
|
|
patches = (
|
|
mock.patch("platform_ci.__main__.run_target_commands", self._fake_native_build),
|
|
mock.patch("platform_ci.__main__.collect_outputs", self._counting_package),
|
|
)
|
|
for patch in patches:
|
|
patch.start()
|
|
self.addCleanup(patch.stop)
|
|
|
|
def _fake_native_build(self, target: Target, source: Path, **_keywords: object) -> bool:
|
|
"""Record one compile and test; the build leaves the jar the Maven example declares."""
|
|
self.builds += 1
|
|
(source / "target").mkdir(exist_ok=True)
|
|
(source / "target" / f"widget-{self.version}.jar").write_bytes(JAR)
|
|
return True
|
|
|
|
def _counting_package(self, *arguments: object, **keywords: object) -> dict[str, object]:
|
|
"""Record one packaging and run the real packaging code."""
|
|
self.packages += 1
|
|
return collect_outputs(*arguments, **keywords) # type: ignore[arg-type]
|
|
|
|
def _checkout(self, commit: str) -> Path:
|
|
"""A working tree of the commit, separate from the repository so a build never changes history."""
|
|
self._checkouts += 1
|
|
checkout = self.root / f"checkout-{self._checkouts}"
|
|
for name, content in self.repository.snapshots[commit].items():
|
|
(checkout / name).parent.mkdir(parents=True, exist_ok=True)
|
|
(checkout / name).write_bytes(content)
|
|
return checkout
|
|
|
|
def _command(self, arguments: list[str], environment: dict[str, str], catalog: RuntimeCatalog | None = None, use_forge: bool = True) -> tuple[int, str, dict[str, str]]:
|
|
"""Run one pipeline command; return its exit code, stdout and the workflow outputs it wrote."""
|
|
output_file = self.root / f"output-{len(list(self.root.glob('output-*')))}"
|
|
captured = io.StringIO()
|
|
with contextlib.redirect_stdout(captured), contextlib.suppress(Exception):
|
|
code = main(arguments, {**environment, "GITHUB_OUTPUT": str(output_file)}, catalog or self.runtimes, self.forge if use_forge else None)
|
|
outputs = dict(line.split("=", 1) for line in output_file.read_text(encoding="utf-8").splitlines()) if output_file.exists() else {}
|
|
return code, captured.getvalue(), outputs
|
|
return 99, captured.getvalue(), {}
|
|
|
|
def pipeline( # noqa: PLR0913, PLR0917
|
|
self,
|
|
run_id: int,
|
|
event: str = "push",
|
|
catalog: RuntimeCatalog | None = None,
|
|
platform_revision: str = PLATFORM_REVISION,
|
|
run_head: str | None = None,
|
|
built_commit: str | None = None,
|
|
use_forge: bool = True,
|
|
forge_token: str = "",
|
|
secrets_result: str = "success",
|
|
conclusion: str = "success",
|
|
caller: str = CALLER_FILE,
|
|
scoped: ScopedRun | None = None,
|
|
) -> PipelineResult:
|
|
"""Run prepare, reuse, the matrix's verify and package, and the report for the newest commit, as the workflow would."""
|
|
commit = built_commit or self.repository.head
|
|
checkout = self._checkout(commit)
|
|
ref = "refs/heads/main" if event == "push" else "refs/pull/1/merge"
|
|
base = {"CI_EVENT_NAME": event, "CI_EVENT_REF": ref, "CI_DEFAULT_BRANCH": "main", "CI_SOURCE_REVISION": commit, "CI_SOURCE_TREE": self.repository.trees[commit], "CI_PLATFORM_REVISION": platform_revision}
|
|
builds_before, packages_before = self.builds, self.packages
|
|
# the run is registered while it executes; a scoped run is named by the scoped workflow file and carries the server-recorded source
|
|
source, workflow_sha = (scoped.source, scoped.workflow_sha) if scoped else (None, None)
|
|
self.forge.add_run(run_id, run_head or commit, status="in_progress", conclusion="", event=event, caller=scoped.file if scoped else caller, scoped_source=source, workflow_sha=workflow_sha)
|
|
code, _, prepared = self._command(["prepare", "--source", str(checkout)], base, catalog)
|
|
self.assertEqual(code, 0)
|
|
plan = json.loads(prepared["plan"])
|
|
identity = {
|
|
"CI_RUN_ID": str(run_id),
|
|
"CI_REPOSITORY": REPOSITORY,
|
|
"CI_REPOSITORY_ID": str(REPOSITORY_ID),
|
|
"CI_PLAN": prepared["plan"],
|
|
**({"CI_FORGE_TOKEN": forge_token} if forge_token else {}),
|
|
**({"CI_SCOPED_SOURCE_ID": scoped.env_source_id, "CI_WORKFLOW_SHA": scoped.env_workflow_sha or scoped.workflow_sha} if scoped else {}),
|
|
}
|
|
code, _, decided = self._command(["reuse", "--spool", str(self.spool)], {**base, **identity}, catalog, use_forge)
|
|
self.assertEqual(code, 0)
|
|
built: list[str] = []
|
|
for entry in cast("list[dict[str, str]]", json.loads(decided["matrix"])["include"]):
|
|
target = {"CI_TARGET": entry["target"]}
|
|
self.assertEqual(self._command(["verify", "--source", str(checkout)], {**base, **target}, catalog)[0], 0)
|
|
bundle = self.root / f"bundle-{run_id}-{entry['name']}"
|
|
packaging = {**base, **target, "CI_BUILD_OUTCOME": "success", "CI_RUN_ID": str(run_id), "CI_RUN_ATTEMPT": "1"}
|
|
self.assertEqual(self._command(["package", "--source", str(checkout), "--output", str(bundle)], packaging, catalog)[0], 0)
|
|
self.forge_artifacts.append((run_id, f"verified-{entry['name']}", bundle))
|
|
built.append(entry["name"])
|
|
self.forge.finish_run(run_id, conclusion)
|
|
for artifact_run, name, bundle in self.forge_artifacts:
|
|
if artifact_run == run_id:
|
|
self.forge.add_artifact(run_id, name, bundle)
|
|
verify_result = "success" if built else "skipped"
|
|
report_environment = {
|
|
**base,
|
|
"CI_PREPARE_RESULT": "success",
|
|
"CI_VERIFY_RESULT": verify_result,
|
|
"CI_NATIVE_VERIFY_RESULT": "skipped",
|
|
"CI_SECRETS_RESULT": secrets_result,
|
|
"CI_PLAN": prepared["plan"],
|
|
"CI_REUSE_EVIDENCE": decided["evidence"],
|
|
"CI_REUSE_NOTE": decided["note"],
|
|
}
|
|
report_code, report_text, _ = self._command(["report"], report_environment, catalog)
|
|
return PipelineResult(plan, decided, report_code, report_text, self.builds - builds_before, self.packages - packages_before, decided["evidence"], decided["note"], built)
|
|
|
|
def publish(self) -> dict[str, object]:
|
|
"""Run the trusted publisher against the synthetic forge and registries."""
|
|
return self.execute()
|
|
|
|
|
|
class UnchangedTreeMergeTest(LifecycleTest):
|
|
"""The primary goal: no duplicate build between the pull request that verified and the merge that releases."""
|
|
|
|
@override
|
|
def setUp(self) -> None:
|
|
"""Commit the tested change (the pull request head)."""
|
|
super().setUp()
|
|
self.head = self.commit()
|
|
|
|
def test_pull_request_build_then_unchanged_tree_merge_runs_no_compile_test_or_package_and_the_publisher_selects_the_existing_bytes(self) -> None:
|
|
pull_request = self.pipeline(100, event="pull_request")
|
|
self.assertEqual((pull_request.builds, pull_request.packages, pull_request.report_code, pull_request.built_targets), (1, 1, 0, ["library-linux-amd64"]))
|
|
self.assertEqual(pull_request.evidence, "") # the original verification is never replaced
|
|
merge = self.repository.commit({}, "merge commit with no content change")
|
|
merged = self.pipeline(101)
|
|
self.assertEqual((merged.builds, merged.packages), (0, 0)) # NO compile, test or package invocation
|
|
self.assertEqual(json.loads(merged.outputs["matrix"]), {"include": []})
|
|
self.assertEqual((merged.outputs["container_targets"], merged.outputs["native_targets"]), ("false", "false"))
|
|
self.assertEqual(merged.report_code, 0, merged.report_text)
|
|
self.assertIn("REUSED, not rebuilt", merged.report_text)
|
|
self.assertIn("no compile, test or package ran", merged.report_text)
|
|
evidence = json.loads(merged.evidence)
|
|
entry = evidence["targets"]["library-linux-amd64"]
|
|
self.assertEqual((evidence["release_revision"], entry["build_sha"], entry["source_run"], entry["source_event"]), (merge, self.head, "100", "pull_request"))
|
|
self.assertNotEqual(entry["build_sha"], evidence["release_revision"])
|
|
result = self.publish()
|
|
self.assertEqual(result["status"], "published")
|
|
(published,) = cast("list[dict[str, object]]", result["published"])
|
|
self.assertEqual((published["build_sha"], published["release_sha"], published["reused"], published["source_run_id"]), (self.head, merge, True, "100"))
|
|
self.assertEqual((self.builds, self.packages), (1, 1)) # the whole lifecycle compiled, tested and packaged exactly once
|
|
self.assertIn("/api/packages/KofTwentyTwo/maven/com/example/widget/1.0.0-SNAPSHOT/widget-1.0.0-20261104.102030-100.jar", self.package_paths())
|
|
|
|
def test_a_pull_request_run_built_on_its_merge_commit_is_reused_with_the_built_commit_kept_separate(self) -> None:
|
|
merge_commit = self.repository.commit({}, "synthetic merge ref commit with the same tree")
|
|
self.pipeline(100, event="pull_request", run_head=self.head, built_commit=merge_commit)
|
|
release = self.repository.commit({}, "release commit with the same tree")
|
|
merged = self.pipeline(101)
|
|
self.assertEqual((merged.builds, merged.report_code), (0, 0), merged.report_text)
|
|
entry = json.loads(merged.evidence)["targets"]["library-linux-amd64"]
|
|
self.assertEqual((entry["build_sha"], json.loads(merged.evidence)["release_revision"]), (merge_commit, release))
|
|
result = self.publish()
|
|
(published,) = cast("list[dict[str, object]]", result["published"])
|
|
self.assertEqual((published["build_sha"], published["release_sha"]), (merge_commit, release))
|
|
|
|
def test_a_later_unchanged_merge_reuses_the_original_build_not_the_reuse_run(self) -> None:
|
|
self.pipeline(100, event="pull_request")
|
|
self.repository.commit({}, "first merge")
|
|
self.pipeline(101)
|
|
self.repository.commit({}, "second unchanged merge")
|
|
second = self.pipeline(102)
|
|
self.assertEqual((second.builds, second.report_code), (0, 0))
|
|
self.assertEqual(json.loads(second.evidence)["targets"]["library-linux-amd64"]["source_run"], "100")
|
|
|
|
|
|
class ChangedInputPerformsOneVerificationTest(LifecycleTest):
|
|
"""Anything a rebuild could change performs exactly one real verification, and the run still passes only on success."""
|
|
|
|
@override
|
|
def setUp(self) -> None:
|
|
"""Verify the first commit in a pull request."""
|
|
super().setUp()
|
|
self.commit()
|
|
self.pipeline(100, event="pull_request")
|
|
|
|
def _assert_one_real_verification(self, result: PipelineResult, reason: str) -> None:
|
|
self.assertEqual((result.builds, result.packages, result.report_code, result.evidence), (1, 1, 0, ""), result.report_text)
|
|
self.assertEqual(result.built_targets, ["library-linux-amd64"])
|
|
self.assertIn("reuse not used", result.note)
|
|
self.assertRegex(result.note, reason)
|
|
self.assertIn("required once", result.report_text)
|
|
|
|
def test_changed_source_tree(self) -> None:
|
|
self.repository.commit({"README.md": "a content change"})
|
|
self._assert_one_real_verification(self.pipeline(101), "source_tree differs")
|
|
|
|
def test_changed_declared_version(self) -> None:
|
|
self.version = "2.0.0-SNAPSHOT"
|
|
self.commit(version="2.0.0-SNAPSHOT")
|
|
self._assert_one_real_verification(self.pipeline(101), "source_tree differs")
|
|
|
|
def test_changed_platform_revision(self) -> None:
|
|
self.repository.commit({}, "same tree")
|
|
self._assert_one_real_verification(self.pipeline(101, platform_revision=OTHER_PLATFORM_REVISION), "not trusted")
|
|
|
|
def test_changed_runtime_image_or_configuration(self) -> None:
|
|
self.repository.commit({}, "same tree")
|
|
self._assert_one_real_verification(self.pipeline(101, catalog=OTHER_RUNTIME_CATALOG), "config_digest differs|toolchain differs")
|
|
|
|
def test_a_failed_earlier_run_is_never_reused(self) -> None:
|
|
self.repository.commit({}, "same tree")
|
|
self.forge.runs[0]["conclusion"] = "failure"
|
|
self._assert_one_real_verification(self.pipeline(101), "not a completed success")
|
|
|
|
def test_reuse_is_not_attempted_without_a_read_token_or_forge_and_the_build_runs(self) -> None:
|
|
self.repository.commit({}, "same tree")
|
|
self._assert_one_real_verification(self.pipeline(101, use_forge=False), "no read-only forge token")
|
|
|
|
def test_a_pull_request_does_not_reuse_another_pull_request(self) -> None:
|
|
self.repository.commit({}, "same tree")
|
|
result = self.pipeline(101, event="pull_request")
|
|
self.assertEqual((result.builds, result.packages, result.report_code, result.evidence), (1, 1, 0, ""))
|
|
self.assertIn("no earlier run is proven identical", result.note)
|
|
|
|
|
|
class CompletedPushBeforePullRequestTest(LifecycleTest):
|
|
"""Only a proven completed push replaces the later PR's compile, test and package."""
|
|
|
|
@override
|
|
def setUp(self) -> None:
|
|
"""Retain one successful ordinary push bundle before the PR event."""
|
|
super().setUp()
|
|
self.commit()
|
|
self.pipeline(100)
|
|
|
|
def _assert_builds(self, result: PipelineResult) -> None:
|
|
self.assertEqual((result.builds, result.packages, result.report_code, result.evidence), (1, 1, 0, ""), result.report_text)
|
|
|
|
def _assert_reuses(self, result: PipelineResult) -> None:
|
|
self.assertEqual((result.builds, result.packages, result.report_code), (0, 0, 0), result.report_text)
|
|
entry = json.loads(result.evidence)["targets"]["library-linux-amd64"]
|
|
self.assertEqual((entry["source_run"], entry["source_event"]), ("100", "push"))
|
|
|
|
def test_completed_same_source_push_is_reused_by_pr(self) -> None:
|
|
self._assert_reuses(self.pipeline(101, event="pull_request"))
|
|
|
|
def test_completed_same_source_push_reuse_by_another_push_is_preserved(self) -> None:
|
|
self._assert_reuses(self.pipeline(101))
|
|
|
|
def test_push_association_is_not_confused_with_a_pr_event(self) -> None:
|
|
self.forge.runs[0]["pull_requests"] = [{"number": 29}]
|
|
self._assert_reuses(self.pipeline(101, event="pull_request"))
|
|
|
|
def test_malformed_push_association_metadata_is_refused(self) -> None:
|
|
self.forge.runs[0]["pull_requests"] = {"number": 29}
|
|
self._assert_builds(self.pipeline(101, event="pull_request"))
|
|
|
|
def test_changed_tree_performs_verification(self) -> None:
|
|
self.repository.commit({"README.md": "changed source"})
|
|
self._assert_builds(self.pipeline(101, event="pull_request"))
|
|
|
|
def test_changed_runtime_inputs_perform_verification(self) -> None:
|
|
self._assert_builds(self.pipeline(101, event="pull_request", catalog=OTHER_RUNTIME_CATALOG))
|
|
|
|
def test_changed_helper_revision_performs_verification(self) -> None:
|
|
self._assert_builds(self.pipeline(101, event="pull_request", platform_revision=OTHER_PLATFORM_REVISION))
|
|
|
|
def test_untrusted_source_workflow_performs_verification(self) -> None:
|
|
self.forge.runs[0]["path"] = "not-canonical.yaml@refs/heads/main"
|
|
self._assert_builds(self.pipeline(101, event="pull_request"))
|
|
|
|
def test_fork_source_repository_performs_verification(self) -> None:
|
|
self.forge.runs[0]["repository"] = {"id": REPOSITORY_ID, "full_name": REPOSITORY, "fork": True}
|
|
self._assert_builds(self.pipeline(101, event="pull_request"))
|
|
|
|
def test_unproven_source_commit_performs_verification(self) -> None:
|
|
self.forge.runs[0]["head_sha"] = "f" * 40
|
|
self._assert_builds(self.pipeline(101, event="pull_request"))
|
|
|
|
def test_failed_source_push_performs_verification(self) -> None:
|
|
self.forge.runs[0]["conclusion"] = "failure"
|
|
self._assert_builds(self.pipeline(101, event="pull_request"))
|
|
|
|
def test_in_progress_push_is_never_reused(self) -> None:
|
|
self.forge.runs[0]["status"] = "in_progress"
|
|
self.forge.runs[0]["conclusion"] = ""
|
|
self._assert_builds(self.pipeline(101, event="pull_request"))
|
|
|
|
def test_reuse_does_not_bypass_current_required_secret_scan(self) -> None:
|
|
result = self.pipeline(101, event="pull_request", secrets_result="failure")
|
|
self.assertEqual((result.builds, result.packages, result.report_code), (0, 0, 1))
|
|
self.assertTrue(result.evidence)
|
|
|
|
|
|
class FinalGateTest(LifecycleTest):
|
|
"""A failed mandatory check is never ignored, and reuse needs valid independently checked evidence."""
|
|
|
|
@override
|
|
def setUp(self) -> None:
|
|
"""Verify once in a pull request, then reuse it for an unchanged merge."""
|
|
super().setUp()
|
|
self.commit()
|
|
self.pipeline(100, event="pull_request")
|
|
self.repository.commit({}, "same tree")
|
|
self.reused = self.pipeline(101)
|
|
assert self.reused.evidence
|
|
|
|
def _report(self, **changes: str) -> tuple[int, str]:
|
|
environment = {
|
|
"CI_PREPARE_RESULT": "success",
|
|
"CI_VERIFY_RESULT": "skipped",
|
|
"CI_NATIVE_VERIFY_RESULT": "skipped",
|
|
"CI_SECRETS_RESULT": "success",
|
|
"CI_PLAN": json.dumps(self.reused.plan),
|
|
"CI_REUSE_EVIDENCE": self.reused.evidence,
|
|
"CI_REUSE_NOTE": self.reused.note,
|
|
**changes,
|
|
}
|
|
code, text, _ = self._command(["report"], environment)
|
|
return code, text
|
|
|
|
def test_valid_reuse_passes(self) -> None:
|
|
self.assertEqual(self._report()[0], 0)
|
|
|
|
def test_a_failed_secret_scan_fails_the_run_even_though_nothing_was_rebuilt(self) -> None:
|
|
code, text = self._report(CI_SECRETS_RESULT="failure")
|
|
self.assertEqual(code, 1)
|
|
self.assertIn("the secret scan did not succeed", text)
|
|
|
|
def test_a_failed_preparation_is_never_masked_by_reuse(self) -> None:
|
|
self.assertEqual(self._command(["report"], {"CI_PREPARE_RESULT": "failure", "CI_PLAN": json.dumps(self.reused.plan), "CI_VERIFY_RESULT": "skipped", "CI_SECRETS_RESULT": "success", "CI_REUSE_EVIDENCE": self.reused.evidence})[0], 1)
|
|
|
|
def test_reuse_evidence_with_a_verification_job_that_ran_anyway_is_inconsistent(self) -> None:
|
|
code, text = self._report(CI_VERIFY_RESULT="success")
|
|
self.assertEqual(code, 1)
|
|
self.assertIn("verification was expected to skip", text)
|
|
|
|
def test_a_skipped_verification_without_evidence_is_a_failure_not_a_pass(self) -> None:
|
|
self.assertEqual(self._report(CI_REUSE_EVIDENCE="", CI_REUSE_NOTE="")[0], 1)
|
|
|
|
def test_tampered_or_incomplete_evidence_fails_the_gate(self) -> None:
|
|
evidence = json.loads(self.reused.evidence)
|
|
entry = evidence["targets"]["library-linux-amd64"]
|
|
cases: dict[str, dict[str, object]] = {
|
|
"other revision": {**evidence, "release_revision": "e" * 40},
|
|
"other tree": {**evidence, "source_tree": "e" * 40},
|
|
"no targets": {**evidence, "targets": {}},
|
|
"extra target": {**evidence, "targets": {**evidence["targets"], "ghost-linux-amd64": entry}},
|
|
"bad manifest digest": {**evidence, "targets": {"library-linux-amd64": {**entry, "manifest_digest": "nope"}}},
|
|
"inputs of another build": {**evidence, "targets": {"library-linux-amd64": {**entry, "build_inputs": {**entry["build_inputs"], "config_digest": "0" * 64}}}},
|
|
"unchecked lineage": {**evidence, "targets": {"library-linux-amd64": {**entry, "verified": {**entry["verified"], "lineage": False}}}},
|
|
"untrusted event": {**evidence, "targets": {"library-linux-amd64": {**entry, "source_event": "workflow_dispatch"}}},
|
|
}
|
|
for name, document in cases.items():
|
|
with self.subTest(name):
|
|
self.assertEqual(self._report(CI_REUSE_EVIDENCE=json.dumps(document))[0], 1)
|
|
self.assertEqual(self._report(CI_REUSE_EVIDENCE="not json")[0], 1)
|
|
|
|
def test_evidence_cannot_justify_skipping_a_pull_request(self) -> None:
|
|
plan = {**self.reused.plan, "event": "pull_request", "reuse": {"eligible": False, "reason": "pull_request events always verify"}}
|
|
self.assertEqual(self._report(CI_PLAN=json.dumps(plan))[0], 1)
|
|
|
|
def test_report_names_the_build_sha_and_release_sha_separately(self) -> None:
|
|
evidence = json.loads(self.reused.evidence)
|
|
self.assertNotEqual(evidence["targets"]["library-linux-amd64"]["build_sha"], evidence["release_revision"])
|
|
self.assertTrue(re.search(r"reused tested bytes of run 100", self.reused.note))
|
|
|
|
|
|
class FabricatedWorkflowTest(LifecycleTest):
|
|
"""A custom workflow that uploads valid bundles with the canonical job names is never trusted; the build runs instead."""
|
|
|
|
CUSTOM = ".gitea/workflows/custom.yaml"
|
|
|
|
@override
|
|
def setUp(self) -> None:
|
|
"""Commit a tree holding both the canonical caller and a custom workflow file."""
|
|
super().setUp()
|
|
self.head = self.commit({self.CUSTOM: CUSTOM_CALLER_TEXT})
|
|
|
|
def test_bundles_of_a_custom_workflow_run_are_not_reused_and_the_merge_builds_once(self) -> None:
|
|
fabricated = self.pipeline(100, caller="custom.yaml")
|
|
self.assertEqual((fabricated.builds, fabricated.evidence), (1, ""))
|
|
self.assertIn("does not prove this run was executed by the canonical workflow", fabricated.note)
|
|
self.repository.commit({}, "merge commit with no content change")
|
|
merged = self.pipeline(101)
|
|
self.assertEqual((merged.builds, merged.packages, merged.report_code, merged.evidence), (1, 1, 0, ""))
|
|
self.assertIn("not executed by a trusted canonical workflow reference", merged.note)
|
|
self.assertIn("required once", merged.report_text)
|
|
|
|
def test_the_publisher_refuses_the_custom_workflows_bundles_and_publishes_nothing(self) -> None:
|
|
self.pipeline(100, caller="custom.yaml")
|
|
self.repository.commit({}, "merge commit with no content change")
|
|
with self.assertRaisesRegex(ReuseError, "not executed by a trusted canonical workflow reference"):
|
|
self.publish()
|
|
self.assertFalse(self.world.written_anything())
|
|
|
|
def test_a_canonical_looking_caller_pinned_to_another_implementation_is_not_trusted(self) -> None:
|
|
other = CALLER_TEXT.replace("4" * 40, "5" * 40)
|
|
self.commit({".gitea/workflows/unified-qualification.yaml": other})
|
|
self.pipeline(100) # builds under the other pinned implementation
|
|
self.repository.commit({}, "same tree")
|
|
with self.assertRaisesRegex(ReuseError, "not executed by a trusted canonical workflow reference"):
|
|
self.publish()
|
|
|
|
|
|
class WorkflowSkipsTruthfullyTest(unittest.TestCase):
|
|
"""The workflow's verify jobs follow the reuse step's matrices, so a reuse leaves them skipped rather than green."""
|
|
|
|
def test_verify_jobs_depend_only_on_the_reuse_step_outputs(self) -> None:
|
|
import yaml # noqa: PLC0415 - only this test needs the raw workflow
|
|
|
|
text = (Path(__file__).resolve().parent.parent / ".gitea" / "scoped_workflows" / "pipeline.yaml").read_text(encoding="utf-8")
|
|
jobs = cast("dict[str, dict[str, object]]", yaml.safe_load(text)["jobs"])
|
|
outputs = cast("dict[str, str]", jobs["prepare"]["outputs"])
|
|
for name in ("matrix", "native_matrix", "container_targets", "native_targets", "reuse_evidence", "reuse_note"):
|
|
self.assertIn("steps.reuse.outputs", outputs[name])
|
|
self.assertEqual(jobs["verify"]["if"], "${{ needs.prepare.outputs.container_targets == 'true' }}")
|
|
self.assertEqual(jobs["verify-native"]["if"], "${{ needs.prepare.outputs.native_targets == 'true' }}")
|
|
self.assertEqual(jobs["secrets"]["if"], "${{ needs.prepare.outputs.verify == 'true' }}") # new required validation is never skipped by reuse
|
|
report_env = cast("dict[str, str]", cast("list[dict[str, object]]", jobs["report"]["steps"])[-1]["env"])
|
|
self.assertEqual((report_env["CI_REUSE_EVIDENCE"], report_env["CI_REUSE_NOTE"]), ("${{ needs.prepare.outputs.reuse_evidence }}", "${{ needs.prepare.outputs.reuse_note }}"))
|