platform-ci/tests/test_finalize_version.py

"""The `finalize-version` command and the `version_write_back` configuration contract. UNTESTED: written while test execution was suspended; nothing is run, signed, pushed or contacted.

The forge, run trust, artifact selection and the write itself are patched at the module boundary: these tests pin what the command reads, refuses and passes on, not the write (see
`test_source_lifecycle.py`).
"""

import unittest
from types import SimpleNamespace
from typing import Any, cast
from unittest.mock import patch

import yaml

from platform_ci.__main__ import main
from platform_ci.configuration import ConfigurationError, WriteBackPolicy, parse_project_configuration
from platform_ci.delivery.errors import DeliveryError
from platform_ci.delivery.finalize_version import finalize_version
from platform_ci.delivery.publisher import RunInProgressError
from platform_ci.delivery.source_lifecycle import Fatal, NotYetPublished
from tests.support import example_mapping

SHA = "a" * 40
PLATFORM = "b" * 40
MODULE = "platform_ci.delivery.finalize_version"


def _configuration(*, tag_driven: bool = False, write_back: object = None, central: bool = False, publish: bool = True) -> dict[str, object]:
   """The Maven publication example with one channel (branch- or tag-driven) that may declare `version_write_back`."""
   configuration = example_mapping("maven-publication.ci.yaml")
   channel: dict[str, object] = {"trigger": {"push_tags": ["v*.*.*"]} if tag_driven else {"push_branches": ["release/**"]}, "versioning": "maven-release"}
   if tag_driven:
      channel["tag_policy"] = "stable"
   if central:
      channel["central_completion"] = True
   if write_back is not None:
      channel["version_write_back"] = write_back
   publications = {"library-packages": {"component": "library", "artifact": "packages", "destination": "company-packages", "channels": ["candidate"]}} if publish else {}
   configuration["release"] = {"channels": {"candidate": channel}, "publications": publications}
   return configuration


def _environment(ref: str = "refs/heads/release/1.6", **overrides: str) -> dict[str, str]:
   values = {
      "CI_EVENT_NAME": "push",
      "CI_EVENT_REF": ref,
      "CI_SOURCE_REVISION": SHA,
      "CI_REPOSITORY": "Org/repo",
      "CI_REPOSITORY_ID": "7",
      "CI_RUN_ID": "50",
      "CI_RUN_ATTEMPT": "1",
      "CI_PLATFORM_REVISION": PLATFORM,
      "CI_SCOPED_SOURCE_ID": "3",
      "CI_WORKFLOW_SHA": "c" * 40,
      "GITHUB_SERVER_URL": "https://git.kof22.com",
      "FORGE_TOKEN": "job-token",
   }
   return values | overrides


def _artifact(version: str | None = "1.6.0-RC.1") -> Any:
   inputs = None if version is None else SimpleNamespace(version=version)
   return SimpleNamespace(manifest=SimpleNamespace(build_inputs=inputs))


class ConfigurationContractTest(unittest.TestCase):
   def test_a_branch_driven_channel_opts_in_and_defaults_are_closed(self) -> None:
      channel = parse_project_configuration(_configuration(write_back={"require_push": True})).channels[0]
      self.assertEqual(channel.version_write_back, WriteBackPolicy(require_push=True))
      self.assertIsNone(parse_project_configuration(_configuration()).channels[0].version_write_back)

   def test_a_tag_driven_channel_names_the_branch_it_writes_to(self) -> None:
      channel = parse_project_configuration(_configuration(tag_driven=True, write_back={"target_branch": "main"})).channels[0]
      self.assertEqual(channel.version_write_back, WriteBackPolicy(target_branch="main"))

   def test_each_invalid_declaration_is_refused(self) -> None:
      cases: dict[str, dict[str, object]] = {
         "other file": _configuration(write_back={"file": "build.gradle"}),
         "branch target on a branch channel": _configuration(write_back={"target_branch": "main"}),
         "tag channel without a target": _configuration(tag_driven=True, write_back={}),
         "bad require_push": _configuration(write_back={"require_push": "yes"}),
         "unknown field": _configuration(write_back={"token": "x"}),
         "not a mapping": _configuration(write_back="yes"),
         "no maven publication": _configuration(write_back={}, publish=False),
      }
      for name, configuration in cases.items():
         with self.subTest(name), self.assertRaises(ConfigurationError):
            parse_project_configuration(configuration)

   def test_a_channel_with_both_trigger_kinds_cannot_write_back(self) -> None:
      configuration = _configuration(write_back={})
      cast("dict[str, dict[str, dict[str, Any]]]", configuration["release"])["channels"]["candidate"]["trigger"]["push_tags"] = ["v*.*.*"]
      with self.assertRaises(ConfigurationError):
         parse_project_configuration(configuration)


def _run(environment: dict[str, str], configuration: dict[str, object] | None, *, artifacts: dict[str, Any] | None = None, trusted: bool = True, select_error: Exception | None = None, **injected: Any) -> tuple[list[str], dict[str, Any]]:
   """Run the command with the forge boundary patched; return its lines and the keyword arguments `postpublish` received."""
   seen: dict[str, Any] = {}

   def postpublish(ctx: Any, policy: Any, **keywords: Any) -> dict[str, str]:
      seen.update(keywords, ctx=ctx, policy=policy)
      return {"status": "pushed", "reason": "", "pushed_sha": "d" * 40}

   content = None if configuration is None else yaml.safe_dump(configuration).encode()
   selector = patch(f"{MODULE}.select_artifacts", side_effect=select_error) if select_error else patch(f"{MODULE}.select_artifacts", return_value=(artifacts or {"library-linux-amd64": _artifact()}, []))
   with (
      patch(f"{MODULE}.require_repository"),
      patch(f"{MODULE}.current_workflow_trust", return_value=object() if trusted else None),
      patch(f"{MODULE}.read_file", return_value=content),
      selector,
      patch(f"{MODULE}.selected_artifact_files", return_value=("file",)),
      patch(f"{MODULE}.postpublish", side_effect=postpublish),
      patch(f"{MODULE}.SourceWriter"),
      patch(f"{MODULE}.VersionGit"),
      patch(f"{MODULE}.legacy.package_client"),
      patch(f"{MODULE}.load_runtime_catalog"),
   ):
      lines = finalize_version(environment, forge=cast("Any", object()), **injected)
   return lines, seen


class FinalizeVersionTest(unittest.TestCase):
   def test_a_branch_run_passes_the_recorded_version_and_the_released_branch(self) -> None:
      lines, seen = _run(_environment(), _configuration(write_back={"require_push": True}, central=True))
      self.assertEqual((seen["version"], seen["ctx"].branch, seen["ctx"].tag, seen["central_required"], seen["owner"], seen["repository_id"]), ("1.6.0-RC.1", "release/1.6", "", True, "Org", 7))
      self.assertEqual(seen["policy"], WriteBackPolicy(require_push=True))
      self.assertEqual((seen["ctx"].run, seen["ctx"].attempt, seen["ctx"].sha, seen["ctx"].repository), (50, 1, SHA, "Org/repo"))
      self.assertIn("pushed", lines[0])

   def test_a_tag_run_writes_to_the_configured_branch_and_binds_the_tag(self) -> None:
      _, seen = _run(_environment("refs/tags/v1.6.0"), _configuration(tag_driven=True, write_back={"target_branch": "main"}))
      self.assertEqual((seen["ctx"].branch, seen["ctx"].tag, seen["central_required"]), ("main", "v1.6.0", False))

   def test_no_matching_opt_in_writes_nothing(self) -> None:
      for environment, configuration in (
         (_environment(), _configuration()),
         (_environment("refs/heads/develop"), _configuration(write_back={})),
         (_environment(), None),
      ):
         lines, seen = _run(environment, configuration)
         self.assertEqual(seen, {})
         self.assertIn("nothing to write back", lines[0])

   def test_incomplete_identity_and_foreign_forge_are_refused_before_any_read(self) -> None:
      for overrides in ({"CI_EVENT_NAME": "pull_request"}, {"CI_SOURCE_REVISION": "abc"}, {"FORGE_TOKEN": ""}, {"GITHUB_SERVER_URL": "https://example.com"}, {"CI_RUN_ATTEMPT": ""}, {"CI_PLATFORM_REVISION": ""}, {"CI_EVENT_REF": "refs/pull/1/merge"}):
         with self.subTest(overrides), self.assertRaises(DeliveryError):
            finalize_version(_environment(**overrides))

   def test_an_unproven_workflow_is_refused(self) -> None:
      with self.assertRaises(DeliveryError):
         _run(_environment(), _configuration(write_back={}), trusted=False)

   def test_a_bundle_without_build_inputs_is_fatal(self) -> None:
      with self.assertRaises(Fatal):
         _run(_environment(), _configuration(write_back={}), artifacts={"library-linux-amd64": _artifact(None)})

   def test_an_unfinished_release_run_is_not_yet_published(self) -> None:
      with self.assertRaises(NotYetPublished):
         _run(_environment(), _configuration(write_back={}), select_error=RunInProgressError("run 9 has not finished"))


class EntryPointTest(unittest.TestCase):
   def test_the_command_is_registered_and_fails_closed_without_the_workflow_environment(self) -> None:
      self.assertEqual(main(["finalize-version"], environment={}), 1)


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