platform-ci/tests/test_source_lifecycle.py

"""The source-lifecycle write phase: the captured revision rule, the confined writer, the staged version write and the evidence it waits for.

NOT YET RUN: written while test execution was suspended. Forge and Git are fakes of the captured helper's calls; nothing contacts a forge, signs or pushes.
"""

import base64
import hashlib
import io
import json
import unittest
import urllib.error
from dataclasses import dataclass
from email.message import Message
from types import SimpleNamespace
from typing import Any, override
from unittest.mock import patch

from platform_ci.configuration import WriteBackPolicy
from platform_ci.delivery import completion_receipt
from platform_ci.delivery.maven_reactor import build_reactor
from platform_ci.delivery.source_lifecycle import (
   Fatal,
   NotYetPublished,
   PushFailure,
   SourceLifecycleError,
   SourceWriter,
   VersionGit,
   VersionPlan,
   WriteContext,
   build_plan,
   git_blob_sha,
   persist_version,
   postpublish,
   revision_only_change,
   rewrite_revision,
   verify_bump_commit,
   verify_completion,
   verify_publication,
)

SRC = "a" * 40
CHILD = "b" * 40
OLD = "<project><properties><revision>1.2.0-SNAPSHOT</revision></properties></project>\n"
NEW = OLD.replace("1.2.0-SNAPSHOT", "1.2.0-RC.3")
CTX = WriteContext("O/r", SRC, "release/1.2", 41, 1)


class RevisionRuleTest(unittest.TestCase):
   """The captured `revisionOnlyChange`: only the one root `properties/revision` value may differ, byte for byte."""

   def test_the_one_value_change_is_accepted_and_reported(self) -> None:
      self.assertEqual(revision_only_change(OLD, NEW), ("1.2.0-SNAPSHOT", "1.2.0-RC.3"))
      self.assertEqual(rewrite_revision(OLD, "1.2.0-RC.3"), NEW)
      self.assertEqual(revision_only_change(OLD, OLD), ("1.2.0-SNAPSHOT", "1.2.0-SNAPSHOT"))

   def test_everything_else_is_refused(self) -> None:
      two = OLD.replace("</properties>", "<revision>x</revision></properties>")
      managed = "<project><dependencyManagement><revision>1</revision></dependencyManagement><properties></properties></project>"
      cases = {
         "two revisions": (two, two),
         "revision outside root properties": (managed, managed),
         "another edit": (OLD, NEW.replace("<project>", "<project >")),
         "unsafe value": (OLD, NEW.replace("RC.3", "RC 3")),
         "doctype": ("<!DOCTYPE x>" + OLD, "<!DOCTYPE x>" + NEW),
         "none": ("<project/>", "<project/>"),
      }
      for name, (old, new) in cases.items():
         with self.subTest(name), self.assertRaises(SourceLifecycleError):
            revision_only_change(old, new)

   def test_rewrite_changes_only_the_value(self) -> None:
      with self.assertRaises(SourceLifecycleError):
         rewrite_revision(OLD, "bad value")
      self.assertEqual(git_blob_sha(b"x"), hashlib.sha1(b"blob 1\0x").hexdigest())  # noqa: S324


class PlanTest(unittest.TestCase):
   """The plan is bound to the source and to the version the verified bundle recorded."""

   policy = WriteBackPolicy()

   def test_a_changed_version_plans_and_an_unchanged_one_does_not(self) -> None:
      plan = build_plan(CTX, self.policy, old=OLD.encode(), new=NEW.encode(), verified_version="1.2.0-RC.3")
      assert plan is not None
      self.assertEqual((plan.declared, plan.effective, plan.source_blob, plan.source_commit), ("1.2.0-SNAPSHOT", "1.2.0-RC.3", git_blob_sha(OLD.encode()), SRC))
      self.assertIsNone(build_plan(CTX, self.policy, old=OLD.encode(), new=OLD.encode(), verified_version="1.2.0-SNAPSHOT"))

   def test_relabelling_oversize_and_an_unconfigured_tag_run_are_refused(self) -> None:
      with self.assertRaises(Fatal):
         build_plan(CTX, self.policy, old=OLD.encode(), new=NEW.encode(), verified_version="1.2.0-RC.4")
      with self.assertRaises(Fatal):
         build_plan(CTX, self.policy, old=OLD.encode() + b" " * 600000, new=NEW.encode(), verified_version="1.2.0-RC.3")
      with self.assertRaises(Fatal):
         build_plan(WriteContext("O/r", SRC, "main", 41, 1, tag="v1.2.0"), self.policy, old=OLD.encode(), new=NEW.encode(), verified_version="1.2.0-RC.3")


@dataclass(frozen=True)
class Faults:
   """The ways the captured checks can fail; the defaults are a clean forge."""

   verified: bool = True
   extra_file: bool = False
   message: str | None = None
   moved: bool = False
   staged: bytes | None = None
   parents: list[str] | None = None


class FakeForge:
   """The forge calls `persist_version` makes, with switches for each way the captured checks can fail."""

   def __init__(self, faults: Faults | None = None) -> None:
      self.faults = faults or Faults()
      self.branches = {"release/1.2": "c" * 40 if self.faults.moved else SRC}
      self.calls: list[tuple[str, str]] = []

   def __call__(self, method: str, path: str, body: dict[str, Any] | None = None) -> tuple[int, Any]:
      self.calls.append((method, path))
      tail = path.split("/repos/O/r", 1)[1]
      answered = self._branch_or_write(method, tail, body)
      if answered is not None:
         return answered
      return self._read(method, path, tail)

   def _branch_or_write(self, method: str, tail: str, body: dict[str, Any] | None) -> tuple[int, Any] | None:
      """Answer the branch reads, the staging-branch creation, the staged write and the deletion; None for anything else."""
      if method == "GET" and tail.startswith("/branches/"):
         name = tail.removeprefix("/branches/").replace("%2F", "/")
         return (200, {"commit": {"id": self.branches[name]}}) if name in self.branches else (404, None)
      if method == "POST" and tail == "/branches":
         assert body is not None
         self.branches[body["new_branch_name"]] = body["old_ref_name"]
         return 201, {}
      if method == "PUT":
         assert body is not None
         self.branches[body["branch"]] = CHILD
         return 200, {"commit": {"sha": CHILD}}
      if method == "DELETE":
         self.branches.pop(tail.removeprefix("/branches/").replace("%2F", "/"), None)
         return 204, None
      return None

   def _read(self, method: str, path: str, tail: str) -> tuple[int, Any]:
      """Answer the reads of the staged commit, its server verification and its file bytes."""
      if tail == f"/commits/{CHILD}":
         return 200, {"files": [{"filename": "pom.xml"}] + ([{"filename": "other"}] if self.faults.extra_file else []), "parents": [{"sha": p} for p in (self.faults.parents or [SRC])]}
      if tail == f"/git/commits/{CHILD}":
         return 200, {"commit": {"verification": {"verified": self.faults.verified}, "message": self.faults.message or "Bump version to 1.2.0-RC.3 [skip ci]\n"}}
      if tail.startswith("/contents/pom.xml?ref="):
         return 200, {"content": base64.b64encode(self.faults.staged if self.faults.staged is not None else NEW.encode()).decode(), "sha": "x"}
      raise AssertionError((method, path))


class FakeGit(VersionGit):
   """Pushes the verified child to the branch, or fails the way Git can; it never runs Git."""

   def __init__(self, forge: FakeForge, *, reject: bool = False, silent: bool = False) -> None:
      super().__init__("https://forge.invalid", "O/r", "unused-token")
      self.forge, self.reject, self.silent = forge, reject, silent
      self.pushed: list[tuple[str, str]] = []

   @override
   def push_verified(self, stage: str, child: str, source: str, branch: str) -> None:
      if self.reject:
         raise PushFailure("the version push was rejected")
      self.pushed.append((child, branch))
      if not self.silent:
         self.forge.branches[branch] = child


def plan() -> VersionPlan:
   built = build_plan(CTX, WriteBackPolicy(), old=OLD.encode(), new=NEW.encode(), verified_version="1.2.0-RC.3")
   assert built is not None
   return built


class PersistVersionTest(unittest.TestCase):
   """Staged, server-verified, normal fast-forward; any doubt is refused and only the exact owned staging heads are deleted."""

   def test_the_happy_path_pushes_the_verified_child_and_removes_the_staging_branch(self) -> None:
      forge = FakeForge()
      git = FakeGit(forge)
      self.assertEqual(persist_version(CTX, plan(), forge, git), CHILD)
      self.assertEqual((forge.branches, git.pushed), ({"release/1.2": CHILD}, [(CHILD, "release/1.2")]))
      self.assertIn(("DELETE", "/repos/O/r/branches/ci%2Fversion-stage-41-1"), forge.calls)

   def test_untrustworthy_staging_is_fatal_and_still_cleaned_up(self) -> None:
      for name, forge in {
         "not server verified": FakeForge(Faults(verified=False)),
         "more than one file": FakeForge(Faults(extra_file=True)),
         "wrong message": FakeForge(Faults(message="Bump version to 9 [skip ci]")),
         "wrong bytes": FakeForge(Faults(staged=b"other")),
         "wrong parent": FakeForge(Faults(parents=["d" * 40])),
      }.items():
         with self.subTest(name), self.assertRaises(Fatal):
            persist_version(CTX, plan(), forge, FakeGit(forge))
         self.assertEqual(forge.branches, {"release/1.2": SRC}, name)

   def test_races_rejections_and_an_unacknowledged_push_are_push_failures_that_leave_the_branch_alone(self) -> None:
      moved = FakeForge(Faults(moved=True))
      with self.assertRaises(PushFailure):
         persist_version(CTX, plan(), moved, FakeGit(moved))
      self.assertFalse(any(method == "POST" for method, _ in moved.calls), "no staging branch is created for a moved branch")
      rejected = FakeForge()
      with self.assertRaises(PushFailure):
         persist_version(CTX, plan(), rejected, FakeGit(rejected, reject=True))
      self.assertEqual(rejected.branches, {"release/1.2": SRC})
      silent = FakeForge()
      with self.assertRaises(PushFailure):
         persist_version(CTX, plan(), silent, FakeGit(silent, silent=True))

   def test_a_foreign_staging_head_is_never_deleted(self) -> None:
      forge = FakeForge()

      class Interfering(FakeGit):
         @override
         def push_verified(self, stage: str, child: str, source: str, branch: str) -> None:
            forge.branches[stage] = "e" * 40  # someone else moved the staging branch
            raise PushFailure("rejected")

      with self.assertRaises(PushFailure):
         persist_version(CTX, plan(), forge, Interfering(forge))
      self.assertIn("ci/version-stage-41-1", forge.branches)

   def test_a_tag_run_requires_the_tag_to_still_name_the_source(self) -> None:
      forge = FakeForge()
      tagged = WriteContext("O/r", SRC, "release/1.2", 41, 1, tag="v1.2.0")

      def with_tags(method: str, path: str, body: dict[str, Any] | None = None) -> tuple[int, Any]:
         return (200, {"commit": {"sha": "f" * 40}}) if "/tags/" in path else forge(method, path, body)

      with self.assertRaises(Fatal):
         persist_version(tagged, plan(), with_tags, FakeGit(forge))

   def test_verify_bump_commit_refuses_an_unverified_commit(self) -> None:
      verify_bump_commit(CTX, plan(), CHILD, FakeForge())
      with self.assertRaises(Fatal):
         verify_bump_commit(CTX, plan(), CHILD, FakeForge(Faults(verified=False)))


class SourceWriterTest(unittest.TestCase):
   """The client refuses everything but the planned write, before any request."""

   def make(self) -> tuple[SourceWriter, list[tuple[str, str, dict[str, str], bytes | None]]]:
      sent: list[tuple[str, str, dict[str, str], bytes | None]] = []

      def send(method: str, url: str, headers: Any, data: bytes | None) -> tuple[int, bytes]:
         sent.append((method, url, dict(headers), data))
         return 200, b"{}"

      return SourceWriter("tok", CTX, send=send), sent

   def test_planned_requests_go_to_the_trusted_origin_with_the_token_only_in_a_header(self) -> None:
      writer, sent = self.make()
      stage = "ci/version-stage-41-1"
      writer("GET", "/repos/O/r")
      writer("GET", f"/repos/O/r/contents/pom.xml?ref={SRC}")
      writer("POST", "/repos/O/r/branches", {"new_branch_name": stage, "old_ref_name": SRC})
      writer("PUT", "/repos/O/r/contents/pom.xml", {"branch": stage, "sha": "s", "content": "c", "message": "m", "force_push": False})
      writer("DELETE", "/repos/O/r/branches/ci%2Fversion-stage-41-1")
      self.assertTrue(all(url.startswith("https://git.kof22.com/api/v1/repos/O/r") and "tok" not in url and headers["Authorization"] == "token tok" for _, url, headers, _ in sent))
      self.assertEqual(len(sent), 5)

   def test_everything_else_is_refused_before_any_request(self) -> None:
      writer, sent = self.make()
      stage = "ci/version-stage-41-1"
      for method, path, body in (
         ("GET", "/repos/other/r", None),
         ("GET", "/repos/O/r/contents/other.xml?ref=" + SRC, None),
         ("GET", "/repos/O/r/contents/pom.xml?ref=main", None),
         ("POST", "/repos/O/r/branches", {"new_branch_name": "evil", "old_ref_name": SRC}),
         ("PUT", "/repos/O/r/contents/pom.xml", {"branch": "main", "sha": "s", "content": "c", "message": "m", "force_push": False}),
         ("PUT", "/repos/O/r/contents/pom.xml", {"branch": stage, "sha": "s", "content": "c", "message": "m", "force_push": True}),
         ("PUT", "/repos/O/r/contents/other.xml", {"branch": stage, "sha": "s", "content": "c", "message": "m", "force_push": False}),
         ("DELETE", "/repos/O/r/branches/main", None),
         ("POST", "/repos/O/r/merges", {}),
      ):
         with self.subTest(f"{method} {path}"), self.assertRaises(SourceLifecycleError):
            writer(method, path, body)
      self.assertEqual(sent, [])


def artifact() -> Any:
   """The parts of a verified artifact the evidence readers use."""
   return SimpleNamespace(build_sha=SRC, run=SimpleNamespace(run_id="41", created_at="2026-11-05T01:02:03Z"))


MODULES = [{"groupId": "com.example", "artifactId": "lib", "version": "1.2.0-RC.3", "packaging": "jar", "files": [{"classifier": None, "extension": "jar", "data": b"jar"}, {"classifier": None, "extension": "pom", "data": b"<lib/>"}]}]


class Registry:
   """An in-memory registry reader: 404 as the client raises it."""

   def __init__(self, files: dict[str, bytes] | None = None) -> None:
      self.files = files or {}

   def __call__(self, method: str, path: str) -> tuple[int, bytes]:
      if path not in self.files:
         raise urllib.error.HTTPError(path, 404, "nf", Message(), io.BytesIO(b""))
      return 200, self.files[path]


class EvidenceTest(unittest.TestCase):
   """Nothing is written until the registry and Central evidence are complete and exact."""

   def published(self) -> dict[str, bytes]:
      base = "/api/packages/O/maven/com/example/lib/1.2.0-RC.3"
      return {f"{base}/lib-1.2.0-RC.3.jar": b"jar", f"{base}/lib-1.2.0-RC.3.pom": b"<lib/>"}

   def test_publication_readback_is_pending_when_absent_and_fatal_when_different(self) -> None:
      with patch("platform_ci.delivery.source_lifecycle.verified_modules", return_value=MODULES):
         verify_publication(artifact(), (), "1.2.0-RC.3", "O", Registry(self.published()))
         with self.assertRaises(NotYetPublished):
            verify_publication(artifact(), (), "1.2.0-RC.3", "O", Registry({}))
         changed = self.published()
         changed[next(iter(changed))] = b"other"
         with self.assertRaises(Fatal):
            verify_publication(artifact(), (), "1.2.0-RC.3", "O", Registry(changed))

   def test_the_completion_receipt_must_be_terminal_and_describe_this_reactor(self) -> None:
      reactor = build_reactor(SRC, "1.2.0-RC.3", [{key: value for key, value in MODULES[0].items() if key != "version"}])
      receipt = {
         "schema": "qrun-publication-completion/1",
         "repository": "O/r",
         "repository_id": 7,
         "run_id": 41,
         "source_commit": SRC,
         "version": "1.2.0-RC.3",
         "archive_sha256": reactor.archive_sha256,
         "manifest_sha256": reactor.manifest_sha256,
         "gav": ["com.example:lib:1.2.0-RC.3"],
         "target": "central-publisher",
         "state": "PUBLISHED",
         "deployment_id": "dep-1",
         "files": [],
         "purls": ["pkg:maven/com.example/lib@1.2.0-RC.3"],
      }
      location = completion_receipt.location("O", 7, 41, SRC)

      def run(files: dict[str, bytes]) -> dict[str, Any]:
         with patch("platform_ci.delivery.source_lifecycle.verified_modules", return_value=MODULES), patch("platform_ci.delivery.source_lifecycle.reactor_for", return_value=reactor):
            return verify_completion(artifact(), (), "1.2.0-RC.3", repository="O/r", repository_id=7, owner="O", reader=Registry(files))

      self.assertEqual(run({location: completion_receipt.encode("O", receipt)})["deployment_id"], "dep-1")
      with self.assertRaises(NotYetPublished):
         run({})
      for changes in ({"manifest_sha256": "0" * 64}, {"state": "PENDING"}, {"run_id": 42}):
         with self.subTest(changes), self.assertRaises(Fatal):
            run({location: json.dumps(receipt | changes).encode()})


@dataclass(frozen=True)
class Run:
   """What one `postpublish` run varies; the defaults are a Central release with the default policy and one verified artifact."""

   policy: WriteBackPolicy | None = None
   git: FakeGit | None = None
   version: str = "1.2.0-RC.3"
   central: bool = True
   extra: int = 0


def serve_source_pom(*_arguments: object, **_options: object) -> tuple[int, Any]:
   """Answer every read with the source POM at its blob id."""
   return 200, {"content": base64.b64encode(OLD.encode()).decode(), "sha": git_blob_sha(OLD.encode())}


class PostpublishTest(unittest.TestCase):
   """The captured order: source POM, registry, completion, then the write; outcomes as the helper reported them."""

   def run_it(self, forge: FakeForge, options: Run | None = None) -> dict[str, str]:
      chosen = options or Run()

      def forge_with_source(method: str, path: str, body: dict[str, Any] | None = None) -> tuple[int, Any]:
         if method == "GET" and path == f"/repos/O/r/contents/pom.xml?ref={SRC}":
            return 200, {"content": base64.b64encode(OLD.encode()).decode(), "sha": git_blob_sha(OLD.encode())}
         return forge(method, path, body)

      with patch("platform_ci.delivery.source_lifecycle.verify_publication") as publication, patch("platform_ci.delivery.source_lifecycle.verify_completion") as completion:
         self.publication, self.completion = publication, completion
         return postpublish(
            CTX,
            chosen.policy or WriteBackPolicy(),
            artifact=artifact(),
            files=(),
            version=chosen.version,
            owner="O",
            repository_id=7,
            central_required=chosen.central,
            writer=forge_with_source,
            registry_reader=Registry(),
            completion_reader=Registry(),
            git=chosen.git or FakeGit(forge),
            extra_evidence=tuple((artifact(), ()) for _ in range(chosen.extra)),
         )

   def test_a_changed_version_is_written_only_after_the_evidence_was_read(self) -> None:
      forge = FakeForge()
      result = self.run_it(forge)
      self.assertEqual((result["status"], result["pushed_sha"]), ("pushed", CHILD))
      self.assertEqual((self.publication.call_count, self.completion.call_count), (1, 1))

   def test_every_publication_of_the_release_is_verified_before_the_one_write(self) -> None:
      forge = FakeForge()
      self.run_it(forge, Run(extra=2))
      self.assertEqual((self.publication.call_count, self.completion.call_count), (3, 3))

   def test_missing_evidence_writes_nothing(self) -> None:
      forge = FakeForge()
      with patch("platform_ci.delivery.source_lifecycle.verify_publication", side_effect=NotYetPublished("later")), self.assertRaises(NotYetPublished):
         postpublish(CTX, WriteBackPolicy(), artifact=artifact(), files=(), version="1.2.0-RC.3", owner="O", repository_id=7, central_required=False, writer=serve_source_pom, registry_reader=Registry(), completion_reader=Registry(), git=FakeGit(forge))
      self.assertEqual(forge.calls, [])

   def test_an_unchanged_version_still_waits_for_the_evidence_and_ends_skipped(self) -> None:
      forge = FakeForge()
      result = self.run_it(forge, Run(version="1.2.0-SNAPSHOT"))
      self.assertEqual(result["status"], "skipped")
      self.assertEqual(self.publication.call_count, 1)

   def test_a_push_failure_is_a_warning_unless_the_push_is_required(self) -> None:
      forge = FakeForge()
      self.assertEqual(self.run_it(forge, Run(git=FakeGit(forge, reject=True)))["status"], "push-failed-warning")
      with self.assertRaises(Fatal):
         self.run_it(FakeForge(), Run(policy=WriteBackPolicy(require_push=True), git=FakeGit(FakeForge(), reject=True)))


if __name__ == "__main__":
   unittest.main()