"""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)