Archived
225 lines
15 KiB
Python
225 lines
15 KiB
Python
"""Callback controls staged for root's affected-check run; not executed by the author."""
|
|
|
|
import base64
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from collections.abc import Mapping
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
from typing import Any, cast, override
|
|
from unittest.mock import MagicMock, Mock, patch
|
|
|
|
import yaml
|
|
|
|
from platform_ci.delivery import publisher
|
|
from platform_ci.delivery.errors import DeliveryError
|
|
from platform_ci.delivery.finalize_version import _callback_request, _canonical_build_revision, finalize_version # pyright: ignore[reportPrivateUsage]
|
|
from platform_ci.delivery.forge import CurrentRun, ForgeReader, RepositoryScope, current_workflow_trust
|
|
from platform_ci.delivery.scoped_identity import ScopedWorkflow
|
|
from platform_ci.delivery.version_callback import callback_dispatcher
|
|
from platform_ci.event_selection import push_facts
|
|
from platform_ci.runtimes import RuntimeCatalog
|
|
from tests.synthetic_services import REPOSITORY, REPOSITORY_ID
|
|
from tests.test_finalize_version import _artifact, _configuration, _environment # pyright: ignore[reportPrivateUsage]
|
|
from tests.test_scoped_identity import HEAD, RUN_ID, ListingForge, named
|
|
|
|
MODULE = "platform_ci.delivery.finalize_version"
|
|
DISPATCH = "platform_ci.delivery.version_callback"
|
|
SHA, PLATFORM, WORKFLOW = "a" * 40, "b" * 40, "c" * 40
|
|
|
|
|
|
def environment(**values: str) -> dict[str, str]:
|
|
return _environment(CI_EVENT_NAME="workflow_dispatch", CI_SCOPED_SOURCE_ID="167", CI_ORIGINAL_ATTEMPT="1", **values)
|
|
|
|
|
|
class CallbackAuthorizationTest(unittest.TestCase):
|
|
def check(self, *, env: dict[str, str] | None = None, original: Mapping[str, object] | None = None, facts: Mapping[str, object] | None = None, callback: object | None = None) -> tuple[Any, Any, Any]:
|
|
env = environment() if env is None else env
|
|
source = SimpleNamespace(scoped_workflows=frozenset({ScopedWorkflow(167, "finalize-version.yaml", WORKFLOW)}))
|
|
forge = Mock()
|
|
forge.get.return_value = {"head_sha": SHA, "event": "push", "path": "ci.yaml@refs/heads/release/1.6"} | dict(original or {})
|
|
outcome = SimpleNamespace(status="completed", conclusion="success", head_sha=SHA, event="push", attempt="1")
|
|
for key, value in (facts or {}).items():
|
|
setattr(outcome, key, value)
|
|
with patch(f"{MODULE}.require_repository"), patch(f"{MODULE}.current_workflow_trust", side_effect=[source if callback is None else callback, object()]), patch(f"{MODULE}.inspect_run", return_value=outcome), patch(f"{MODULE}._canonical_build_revision", return_value=PLATFORM):
|
|
return _callback_request(env, 49, cast("ForgeReader", forge))
|
|
|
|
def test_authenticated_original_run_retains_callback_staging_and_default_reuse(self) -> None:
|
|
request, _, _ = self.check()
|
|
self.assertEqual((request.run_id, request.attempt, request.sha), (50, 1, SHA))
|
|
self.assertIsNone(request.release_run, "bundle-less merge must not force its own artifact run")
|
|
|
|
def test_wrong_callback_source_or_workflow_is_denied(self) -> None:
|
|
for callback in (SimpleNamespace(scoped_workflows=frozenset[ScopedWorkflow]()), SimpleNamespace(scoped_workflows=frozenset({ScopedWorkflow(167, "ci.yaml", WORKFLOW)}))):
|
|
with self.subTest(callback=callback), self.assertRaises(DeliveryError):
|
|
self.check(callback=callback)
|
|
|
|
def test_original_sha_ref_event_pending_failure_and_attempt_mismatches_deny(self) -> None:
|
|
for original in ({"head_sha": "d" * 40}, {"event": "pull_request"}, {"path": "ci.yaml@refs/heads/main"}):
|
|
with self.subTest(original=original), self.assertRaises(DeliveryError):
|
|
self.check(original=original)
|
|
for facts in ({"status": "in_progress"}, {"conclusion": "failure"}, {"attempt": "2"}):
|
|
with self.subTest(facts=facts), self.assertRaises(DeliveryError):
|
|
self.check(facts=facts)
|
|
|
|
def test_callback_requires_actual_dispatch_event_and_trusted_forge(self) -> None:
|
|
for env in (environment() | {"CI_EVENT_NAME": "push"}, environment() | {"GITHUB_SERVER_URL": "https://other.invalid"}):
|
|
with self.subTest(env=env["CI_EVENT_NAME"]), self.assertRaises(DeliveryError):
|
|
self.check(env=env)
|
|
|
|
def test_bundleless_merge_uses_prior_pr_artifact_without_rebuilding(self) -> None:
|
|
artifact = _artifact()
|
|
artifact.run = SimpleNamespace(run_id="42", event="pull_request")
|
|
request, forge, trusted = self.check()
|
|
with (
|
|
patch(f"{MODULE}._callback_request", return_value=(request, forge, trusted)),
|
|
patch(f"{MODULE}.require_repository"),
|
|
patch(f"{MODULE}.read_file", return_value=yaml.safe_dump(_configuration(write_back={})).encode()),
|
|
patch(f"{MODULE}.select_artifacts", return_value=({"library-linux-amd64": artifact}, [])) as select,
|
|
patch(f"{MODULE}.selected_artifact_files", return_value=("file",)),
|
|
patch(f"{MODULE}.postpublish", return_value={"status": "skipped", "reason": "unchanged", "pushed_sha": ""}) as write,
|
|
patch(f"{MODULE}.SourceWriter"),
|
|
patch(f"{MODULE}.VersionGit"),
|
|
patch(f"{MODULE}.legacy.package_client"),
|
|
):
|
|
finalize_version(environment(), original_run=49, runtimes=cast("RuntimeCatalog", object()))
|
|
assert select.call_args is not None
|
|
assert write.call_args is not None
|
|
self.assertIsNone(select.call_args.args[7])
|
|
self.assertIs(write.call_args.kwargs["artifact"], artifact)
|
|
self.assertEqual(artifact.run.run_id, "42")
|
|
|
|
|
|
class CallbackScopedIdentityTest(unittest.TestCase):
|
|
def test_real_nested_identity_parser_accepts_finalizer_filename_and_org_source(self) -> None:
|
|
outer, inner = "1" * 40, "2" * 40
|
|
run: dict[str, object] = {"id": RUN_ID, "head_sha": HEAD, "event": "workflow_dispatch", "path": "finalize-version.yaml@refs/heads/main"}
|
|
|
|
class CallbackForge(ListingForge):
|
|
@override
|
|
def get(self, path: str) -> object:
|
|
if path == f"/repos/{REPOSITORY}/actions/runs/{RUN_ID}":
|
|
return run
|
|
return super().get(path)
|
|
|
|
forge = CallbackForge({167: [run]}, [named(f"Scoped workflow [workflow {outer}]"), named(f"Prepare plan [workflow {inner}]")])
|
|
scope = RepositoryScope(REPOSITORY, REPOSITORY_ID, 1000, 10, frozenset({PLATFORM}))
|
|
trusted = current_workflow_trust(forge, scope, CurrentRun(RUN_ID, HEAD, 167, inner))
|
|
self.assertIsNotNone(trusted)
|
|
assert trusted is not None
|
|
self.assertEqual(trusted.scoped_workflows, frozenset({ScopedWorkflow(167, "finalize-version.yaml", outer, inner)}))
|
|
self.assertIsNone(current_workflow_trust(forge, scope, CurrentRun(RUN_ID, HEAD, 3, inner)))
|
|
|
|
|
|
class BuildHelperIdentityTest(unittest.TestCase):
|
|
def test_public_immutable_pipeline_proves_old_and_new_build_pins_without_credentials(self) -> None:
|
|
workflow = ScopedWorkflow(167, "ci.yaml", "1" * 40, "2" * 40)
|
|
scope = RepositoryScope(REPOSITORY, REPOSITORY_ID, 1000, 10, frozenset({PLATFORM}), scoped_workflows=frozenset({workflow}))
|
|
for build in ("530" + "a" * 37, "43c" + "b" * 37):
|
|
content = f"env:\n PLATFORM_CI_REVISION: '{build}'\n".encode()
|
|
metadata = {"type": "file", "encoding": "base64", "size": len(content), "content": base64.b64encode(content).decode()}
|
|
opener = MagicMock()
|
|
opener.open.return_value.__enter__.return_value.read.return_value = json.dumps(metadata).encode()
|
|
with self.subTest(build=build), patch(f"{MODULE}.urllib.request.build_opener", return_value=opener):
|
|
self.assertEqual(_canonical_build_revision(scope), build)
|
|
assert opener.open.call_args is not None
|
|
request = opener.open.call_args.args[0]
|
|
self.assertEqual(request.full_url, "https://git.kof22.com/api/v1/repos/KofTwentyTwo/platform-ci/contents/.gitea/scoped_workflows/pipeline.yaml?ref=" + "2" * 40)
|
|
self.assertIsNone(request.get_header("Authorization"))
|
|
|
|
def test_mutable_or_missing_canonical_build_pin_is_refused(self) -> None:
|
|
workflow = ScopedWorkflow(167, "ci.yaml", "1" * 40, "2" * 40)
|
|
scope = RepositoryScope(REPOSITORY, REPOSITORY_ID, 1000, 10, frozenset({PLATFORM}), scoped_workflows=frozenset({workflow}))
|
|
for content in (b"env: {PLATFORM_CI_REVISION: main}", b"env: {}", None):
|
|
with self.subTest(content=content), patch(f"{MODULE}.read_file", return_value=content), self.assertRaises(DeliveryError):
|
|
_canonical_build_revision(scope)
|
|
|
|
def test_manual_helper_hint_cannot_override_the_canonical_pin(self) -> None:
|
|
with self.assertRaisesRegex(DeliveryError, "immutable build pin"):
|
|
CallbackAuthorizationTest().check(env=environment() | {"CI_PLATFORM_REVISION": "d" * 40})
|
|
|
|
def test_unenrolled_build_hint_never_dispatches(self) -> None:
|
|
with tempfile.TemporaryDirectory() as directory, patch(f"{DISPATCH}.urllib.request.build_opener") as opener, self.assertRaises(DeliveryError):
|
|
enrollment = SimpleNamespace(platform_revisions=frozenset({PLATFORM}))
|
|
callback_dispatcher("synthetic-own-token", Mock())(cast("Any", enrollment), "refs/heads/main", SHA, Path(directory), "d" * 40)
|
|
opener.return_value.open.assert_not_called()
|
|
|
|
|
|
class DispatchTest(unittest.TestCase):
|
|
def test_accepted_marker_prevents_second_post_and_uncertain_marker_refuses(self) -> None:
|
|
enrollment = SimpleNamespace(repository="Org/repo", scope=object(), scoped_workflows=frozenset({ScopedWorkflow(167, "ci.yaml", WORKFLOW)}), platform_revisions=frozenset({PLATFORM}))
|
|
run = {"id": 49, "event": "push", "head_sha": SHA, "path": "ci.yaml@refs/heads/release/1.6"}
|
|
facts = SimpleNamespace(status="completed", conclusion="success", head_sha=SHA, attempt="1")
|
|
response = Mock(status=200)
|
|
response.read.return_value = b'{"workflow_run_id":51}'
|
|
opener = MagicMock()
|
|
opener.open.return_value.__enter__.return_value = response
|
|
with tempfile.TemporaryDirectory() as directory, patch(f"{DISPATCH}.urllib.request.build_opener", return_value=opener), patch(f"{DISPATCH}.list_runs", return_value=[run]), patch(f"{DISPATCH}.inspect_run", return_value=facts):
|
|
callback = callback_dispatcher("synthetic-own-token", Mock())
|
|
root = Path(directory)
|
|
first = callback(cast("Any", enrollment), "refs/heads/release/1.6", SHA, root, PLATFORM)
|
|
self.assertEqual(callback(cast("Any", enrollment), "refs/heads/release/1.6", SHA, root, PLATFORM), first)
|
|
self.assertEqual(opener.open.call_count, 1)
|
|
assert opener.open.call_args is not None
|
|
sent = opener.open.call_args.args[0]
|
|
self.assertIn("scoped_workflow_source_repo_id=167", sent.full_url)
|
|
self.assertEqual(json.loads(sent.data)["inputs"], {"original_run": "49", "original_attempt": "1", "build_revision": PLATFORM})
|
|
marker = next(root.rglob("*.version-callback.json"))
|
|
marker.write_text('{"state":"dispatching"}', encoding="utf-8")
|
|
with self.assertRaises(DeliveryError):
|
|
callback(cast("Any", enrollment), "refs/heads/release/1.6", SHA, root, PLATFORM)
|
|
self.assertEqual(opener.open.call_count, 1)
|
|
|
|
def test_pending_or_failed_original_never_dispatches(self) -> None:
|
|
enrollment = SimpleNamespace(repository="Org/repo", scope=object(), platform_revisions=frozenset({PLATFORM}))
|
|
run = {"id": 49, "event": "push", "head_sha": SHA, "path": "ci.yaml@refs/heads/main"}
|
|
with tempfile.TemporaryDirectory() as directory, patch(f"{DISPATCH}.list_runs", return_value=[run]), patch(f"{DISPATCH}.urllib.request.build_opener") as opener:
|
|
for status, conclusion in (("in_progress", ""), ("completed", "failure")):
|
|
with patch(f"{DISPATCH}.inspect_run", return_value=SimpleNamespace(status=status, conclusion=conclusion, head_sha=SHA)), self.assertRaises(DeliveryError):
|
|
callback_dispatcher("synthetic-own-token", Mock())(cast("Any", enrollment), "refs/heads/main", SHA, Path(directory), PLATFORM)
|
|
opener.return_value.open.assert_not_called()
|
|
|
|
|
|
class TerminalPublisherTest(unittest.TestCase):
|
|
def test_dispatch_waits_for_terminal_publication_and_never_changes_artifact_selection(self) -> None:
|
|
@dataclass
|
|
class Receipt:
|
|
name: str = "package"
|
|
|
|
channel = SimpleNamespace(name="candidate", version_write_back=object())
|
|
project = SimpleNamespace(channels=[channel])
|
|
planned = {"target": "library"}
|
|
request = SimpleNamespace(channel="candidate", destination="packages", format="maven")
|
|
enrollment = SimpleNamespace(repository="Org/repo", bindings=(), scope=object())
|
|
services = SimpleNamespace(forge=Mock(), transports=Mock(), signature_verifier=None, version_finalizer=Mock(return_value={"state": "accepted"}))
|
|
ref = publisher._Ref("branch", "release/1.6", push_facts("release/1.6", SHA)) # pyright: ignore[reportPrivateUsage]
|
|
with (
|
|
tempfile.TemporaryDirectory() as directory,
|
|
patch.object(publisher, "_project_at", return_value=project),
|
|
patch.object(publisher, "channel_matches", return_value=True),
|
|
patch.object(publisher, "publication_requests", return_value=[planned]),
|
|
patch.object(publisher, "_requests", return_value=[(planned, request)]),
|
|
patch.object(publisher, "_select_artifacts", return_value=({"library": SimpleNamespace(manifest=SimpleNamespace(platform_revision=PLATFORM))}, [])),
|
|
patch.object(publisher, "_check_versioning"),
|
|
patch.object(publisher, "authorize"),
|
|
patch.object(publisher, "_check_release_tag"),
|
|
patch.object(publisher, "branch_head", return_value=SHA),
|
|
patch.object(publisher, "publish_artifact", return_value=Receipt()),
|
|
patch.object(publisher, "_write_receipt"),
|
|
patch.object(publisher, "_deploy", return_value=[]),
|
|
patch.object(publisher, "_finalize") as terminal,
|
|
):
|
|
options = SimpleNamespace(spool=Path(directory), receipts=Path(directory) / "receipts", publish=True)
|
|
terminal.return_value = [{"status": "awaiting-completion"}]
|
|
pending = publisher._execute_ref(cast("Any", enrollment), cast("Any", services), cast("Any", options), ref, SHA, {}) # pyright: ignore[reportPrivateUsage]
|
|
assert pending is not None
|
|
self.assertEqual(pending["status"], "published-awaiting-completion")
|
|
services.version_finalizer.assert_not_called()
|
|
terminal.return_value = [{"status": "complete"}]
|
|
completed = publisher._execute_ref(cast("Any", enrollment), cast("Any", services), cast("Any", options), ref, SHA, {}) # pyright: ignore[reportPrivateUsage]
|
|
assert completed is not None
|
|
self.assertEqual(completed["version_callback"], {"state": "accepted"})
|
|
services.version_finalizer.assert_called_once_with(enrollment, "refs/heads/release/1.6", SHA, options.receipts, PLATFORM)
|