platform-ci/tests/test_version_callback.py

"""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 ANY, 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 OuterPipelineIdentityTest(unittest.TestCase):
   def resolve(self, outer: str, *, metadata: Mapping[str, object] | None = None) -> tuple[str, MagicMock]:
      workflow = ScopedWorkflow(167, "ci.yaml", WORKFLOW)
      scope = RepositoryScope(REPOSITORY, REPOSITORY_ID, 1000, 10, frozenset({PLATFORM}), scoped_workflows=frozenset({workflow}))
      documents: list[object] = [metadata if metadata is not None else {"id": 167, "full_name": "Org/ci-policy", "private": True, "fork": False}]
      for content in (outer.encode(), f"env: {{PLATFORM_CI_REVISION: '{PLATFORM}'}}".encode()):
         documents.append({"type": "file", "encoding": "base64", "size": len(content), "content": base64.b64encode(content).decode()})
      documents.insert(2, {"id": 3, "full_name": "KofTwentyTwo/platform-ci", "fork": False})
      opener = MagicMock()
      opener.open.return_value.__enter__.return_value.read.side_effect = [json.dumps(document).encode() for document in documents]
      with patch(f"{MODULE}.urllib.request.build_opener", return_value=opener):
         result = _canonical_build_revision(scope, token=PLATFORM)
      return result, opener

   def test_equal_outer_labels_resolve_exact_immutable_definition_with_other_jobs(self) -> None:
      result, opener = self.resolve("jobs:\n  configuration: {runs-on: linux-small, steps: [{run: 'true'}]}\n  shared: {uses: 'self:KofTwentyTwo/platform-ci/.gitea/scoped_workflows/pipeline.yaml@" + SHA + "'}\n")
      self.assertEqual(result, PLATFORM)
      requests = [call.args[0] for call in opener.open.call_args_list]
      self.assertEqual(
         [request.full_url for request in requests],
         [
            "https://git.kof22.com/api/v1/repositories/167",
            "https://git.kof22.com/api/v1/repos/Org/ci-policy/contents/.gitea/scoped_workflows/ci.yaml?ref=" + WORKFLOW,
            "https://git.kof22.com/api/v1/repos/KofTwentyTwo/platform-ci",
            "https://git.kof22.com/api/v1/repos/KofTwentyTwo/platform-ci/contents/.gitea/scoped_workflows/pipeline.yaml?ref=" + SHA,
         ],
      )
      self.assertTrue(all(request.get_header("Authorization") == "token " + PLATFORM and request.get_method() == "GET" for request in requests))

   def test_wrong_or_forked_registered_source_is_refused(self) -> None:
      for delta in ({"id": 166}, {"fork": True}, {"full_name": "https://other.invalid/source"}):
         with self.subTest(delta=delta), self.assertRaises(DeliveryError):
            self.resolve("jobs: {}", metadata={"id": 167, "full_name": "Org/ci-policy", "private": False, "fork": False} | delta)

   def test_mutable_ambiguous_missing_and_expression_pipeline_refs_are_refused(self) -> None:
      prefix = "self:KofTwentyTwo/platform-ci/.gitea/scoped_workflows/pipeline.yaml@"
      for jobs in ({}, {"shared": {"uses": prefix + "main"}}, {"shared": {"uses": prefix + "${{ inputs.sha }}"}}, {"one": {"uses": prefix + SHA}, "two": {"uses": prefix + SHA}}):
         with self.subTest(jobs=jobs), self.assertRaises(DeliveryError):
            self.resolve(yaml.safe_dump({"jobs": jobs}))


class DispatchTest(unittest.TestCase):
   def test_newer_native_run_does_not_replace_the_canonical_release_run(self) -> None:
      enrollment = SimpleNamespace(repository="Org/repo", scope=object(), scoped_workflows=frozenset({ScopedWorkflow(167, "ci.yaml", WORKFLOW)}), platform_revisions=frozenset({PLATFORM}))
      native = {"id": 4638, "event": "push", "head_sha": SHA, "path": "native.yaml@refs/heads/release/1.6"}
      canonical = {"id": 4637, "event": "push", "head_sha": SHA, "path": "ci.yaml@refs/heads/release/1.6"}
      opener = MagicMock()
      response = opener.open.return_value.__enter__.return_value
      response.status = 200
      response.read.return_value = b'{"workflow_run_id":5000}'

      def inspect(_forge: object, _scope: object, identifier: int) -> object:
         if identifier != 4637:
            raise DeliveryError("run 4638 was not executed by a trusted canonical workflow reference")
         return SimpleNamespace(status="completed", conclusion="success", head_sha=SHA, attempt="1")

      with tempfile.TemporaryDirectory() as directory, patch(f"{DISPATCH}.urllib.request.build_opener", return_value=opener), patch(f"{DISPATCH}.list_runs", return_value=[native, canonical]), patch(f"{DISPATCH}.inspect_run", side_effect=inspect) as inspector:
         callback_dispatcher("synthetic-own-token", Mock())(cast("Any", enrollment), "refs/heads/release/1.6", SHA, Path(directory), PLATFORM)
      inspector.assert_called_once_with(ANY, enrollment.scope, 4637)
      assert opener.open.call_args is not None
      self.assertEqual(json.loads(opener.open.call_args.args[0].data)["inputs"]["original_run"], "4637")

   def test_newest_failed_canonical_run_blocks_older_success(self) -> None:
      enrollment = SimpleNamespace(repository="Org/repo", scope=object(), scoped_workflows=frozenset({ScopedWorkflow(167, "ci.yaml", WORKFLOW)}), platform_revisions=frozenset({PLATFORM}))
      rows = [{"id": identifier, "event": "push", "head_sha": SHA, "path": filename + "@refs/heads/release/1.6"} for identifier, filename in ((4640, "native.yaml"), (4639, "ci.yaml"), (4637, "ci.yaml"))]
      with (
         tempfile.TemporaryDirectory() as directory,
         patch(f"{DISPATCH}.urllib.request.build_opener") as opener,
         patch(f"{DISPATCH}.list_runs", return_value=rows),
         patch(f"{DISPATCH}.inspect_run", return_value=SimpleNamespace(status="completed", conclusion="failure", head_sha=SHA, attempt="1")) as inspector,
         self.assertRaises(DeliveryError),
      ):
         callback_dispatcher("synthetic-own-token", Mock())(cast("Any", enrollment), "refs/heads/release/1.6", SHA, Path(directory), PLATFORM)
      inspector.assert_called_once_with(ANY, enrollment.scope, 4639)
      opener.return_value.open.assert_not_called()

   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}), scoped_workflows=frozenset({ScopedWorkflow(167, "ci.yaml", WORKFLOW)}))
      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)