platform-ci/tests/test_android_maven_adapter.py

"""Maven adapter and the captured Android reactor layout: AAR, jar, POM and `.module` bytes published exactly, with the retained reactor metadata verified.

UNTESTED: written while test execution was suspended. Stdlib only, no network, no process. The artifact is a stand-in exposing what the adapter uses
(`bundle.read_bytes`, `build_sha`, `manifest.build_inputs`); the layouts are the real ones: a Gradle `publishToMavenLocal` capture arranged by
`android-maven-capture.mjs` (`files/<artifact>/<artifact>-<version>[-classifier].<ext>` plus manifest.json, SOURCE_COMMIT, SHA256SUMS) and a plain Maven
`pom.xml` + `target/*.jar` tree. Publication runs through the existing frozen Maven primitive against the in-memory registry of the delivery tests.
"""

import hashlib
import io
import json
import unittest
import zipfile
from collections.abc import Callable
from dataclasses import dataclass, field
from types import SimpleNamespace
from typing import Any, cast

from platform_ci.delivery import legacy
from platform_ci.delivery.adapters import AdapterInput, maven_coordinates, maven_modules, publish_maven_packages, reactor_manifest_sha256
from platform_ci.delivery.authorization import PackageTarget
from platform_ci.delivery.errors import PublicationError
from platform_ci.delivery.manifest import ManifestFile
from platform_ci.delivery.reuse import ReleaseArtifact
from tests.synthetic_services import World

GROUP = "org.example"
VERSION = "1.2.3"
BUILD_SHA = "c" * 40
ROOT = "build/ci-maven-artifact/"
REGISTRY_PREFIX = f"/api/packages/owner/maven/org/example/library/{VERSION}/library-{VERSION}"


def digest(data: bytes) -> str:
   """Return the SHA-256 hex of bytes."""
   return hashlib.sha256(data).hexdigest()


def zip_bytes(name: str = "a.txt") -> bytes:
   """Return a small valid zip (stands for a jar or an aar)."""
   buffer = io.BytesIO()
   with zipfile.ZipFile(buffer, "w") as archive:
      archive.writestr(name, "x")
   return buffer.getvalue()


def pom(artifact_id: str, packaging: str, version: str = VERSION) -> bytes:
   """Return a literal POM."""
   return f'<project xmlns="http://maven.apache.org/POM/4.0.0"><modelVersion>4.0.0</modelVersion><groupId>{GROUP}</groupId><artifactId>{artifact_id}</artifactId><version>{version}</version><packaging>{packaging}</packaging></project>'.encode()


def gradle_module(artifact_id: str, version: str = VERSION) -> bytes:
   """Return Gradle module metadata naming the coordinates."""
   return json.dumps({"formatVersion": "1.1", "component": {"group": GROUP, "module": artifact_id, "version": version}}).encode()


@dataclass(frozen=True)
class FakeBundle:
   """Byte store keyed by bundle path."""

   data: dict[str, bytes]

   def read_bytes(self, entry: ManifestFile) -> bytes:
      """Return the bytes of one retained file."""
      return self.data[entry.path]


@dataclass(frozen=True)
class FakeArtifact:
   """The attributes the Maven adapter reads."""

   bundle: FakeBundle
   build_sha: str = BUILD_SHA
   manifest: Any = field(default_factory=lambda: SimpleNamespace(build_inputs=None))


def entries(files: dict[str, bytes]) -> tuple[ManifestFile, ...]:
   """Return manifest entries (path, size, SHA-256) for the byte store."""
   return tuple(ManifestFile(kind="artifact", name="maven-reactor", path=path, size=len(data), sha256=digest(data)) for path, data in sorted(files.items()))


def artifact_for(files: dict[str, bytes], build_sha: str = BUILD_SHA) -> ReleaseArtifact:
   """Return the stand-in artifact over a byte store."""
   return cast("ReleaseArtifact", FakeArtifact(FakeBundle(files), build_sha))


def group(files: dict[str, bytes], build_sha: str = BUILD_SHA) -> list[dict[str, Any]]:
   """Run the grouping over a byte store."""
   return maven_modules(artifact_for(files, build_sha), entries(files))


@dataclass(frozen=True)
class Capture:
   """The defects and variations a reactor capture is built with; the defaults are the valid capture."""

   commit: str = BUILD_SHA
   aar: bytes | None = None
   library_main: str = "aar"
   module_data: bytes | None = None
   extra: dict[str, bytes] | None = None
   edit: Callable[[dict[str, Any]], None] | None = None
   drop_sums_line: str = ""
   skip_sums: bool = False


def capture(options: Capture | None = None) -> dict[str, bytes]:
   """Build a reactor capture as android-maven-capture.mjs lays it out: a POM-only parent and an aar module with sources and Gradle metadata."""
   chosen = options if options is not None else Capture()
   commit = chosen.commit
   files: dict[str, bytes] = {}

   def add(artifact: str, classifier: str | None, extension: str, data: bytes) -> dict[str, Any]:
      name = f"files/{artifact}/{artifact}-{VERSION}{'-' + classifier if classifier else ''}.{extension}"
      files[ROOT + name] = data
      return {"path": name, "classifier": classifier, "extension": extension, "sha256": digest(data), "size": len(data)}

   parent = [add("parent", None, "pom", pom("parent", "pom"))]
   library = [
      add("library", None, "pom", pom("library", "aar")),
      add("library", None, chosen.library_main, chosen.aar if chosen.aar is not None else zip_bytes("AndroidManifest.xml")),
      add("library", "sources", "jar", zip_bytes("Source.kt")),
      add("library", None, "module", chosen.module_data if chosen.module_data is not None else gradle_module("library")),
   ]
   manifest: dict[str, Any] = {"schema": "qrun-maven-reactor/1", "source_commit": commit, "version": VERSION, "modules": [{"groupId": GROUP, "artifactId": "library", "packaging": "aar", "files": library}, {"groupId": GROUP, "artifactId": "parent", "packaging": "pom", "files": parent}]}
   if chosen.edit is not None:
      chosen.edit(manifest)
   files.update({ROOT + name: data for name, data in (chosen.extra or {}).items()})
   files[ROOT + "manifest.json"] = json.dumps(manifest).encode()
   files[ROOT + "SOURCE_COMMIT"] = f"{commit}\n".encode()
   if not chosen.skip_sums:
      lines = [f"{digest(data)}  {path[len(ROOT) :]}\n" for path, data in sorted(files.items())]
      files[ROOT + "SHA256SUMS"] = "".join(line for line in lines if not (chosen.drop_sums_line and chosen.drop_sums_line in line)).encode()
   return files


class AarReactorTest(unittest.TestCase):
   """The captured Android layout groups into publishable modules with exact bytes; the metadata is verified and never published."""

   def test_the_aar_module_keeps_its_main_file_classifier_jar_gradle_metadata_and_pom_exactly(self) -> None:
      """Coordinates come from the POMs; the main extension is `aar`; classifiers, `.module` and the POM carry their exact bytes."""
      files = capture()
      modules = {module["artifactId"]: module for module in group(files)}
      self.assertEqual({(m["groupId"], m["version"], m["packaging"]) for m in modules.values()}, {(GROUP, VERSION, "aar"), (GROUP, VERSION, "pom")})
      carried = {(item["classifier"], item["extension"]): item["data"] for item in modules["library"]["files"]}
      self.assertEqual(set(carried), {(None, "pom"), (None, "aar"), ("sources", "jar"), (None, "module")})
      self.assertEqual(carried[(None, "aar")], files[ROOT + f"files/library/library-{VERSION}.aar"])
      self.assertEqual(carried[(None, "module")], files[ROOT + f"files/library/library-{VERSION}.module"])
      self.assertEqual([item["extension"] for item in modules["parent"]["files"]], ["pom"])
      self.assertEqual(maven_coordinates(artifact_for(files), entries(files)), (f"{GROUP}:library:{VERSION}", f"{GROUP}:parent:{VERSION}"))

   def test_the_existing_maven_primitive_publishes_the_exact_bytes_and_none_of_the_metadata(self) -> None:
      """Through `publish_maven_packages` and the frozen primitive: aar, module, classifier jar and POM bytes land at the registry paths; manifest, commit and checksums do not."""
      files = capture()
      world = World()
      result = publish_maven_packages(AdapterInput(artifact=artifact_for(files), files=entries(files), request=cast("Any", None), target=PackageTarget(owner="owner"), transports=world.transports()))
      store = world.packages.store
      self.assertEqual(store[f"{REGISTRY_PREFIX}.aar"], files[ROOT + f"files/library/library-{VERSION}.aar"])
      self.assertEqual(store[f"{REGISTRY_PREFIX}.module"], files[ROOT + f"files/library/library-{VERSION}.module"])
      self.assertEqual(store[f"{REGISTRY_PREFIX}-sources.jar"], files[ROOT + f"files/library/library-{VERSION}-sources.jar"])
      self.assertEqual(store[f"{REGISTRY_PREFIX}.pom"], files[ROOT + f"files/library/library-{VERSION}.pom"])
      self.assertFalse([path for path in store if path.endswith(("manifest.json", "SOURCE_COMMIT", "SHA256SUMS"))])
      self.assertEqual({path for path, _digest in result.file_hashes}, set(store))
      self.assertEqual({path for path, _digest in result.metadata_hashes}, {ROOT + "manifest.json", ROOT + "SOURCE_COMMIT", ROOT + "SHA256SUMS"})
      self.assertEqual(dict(result.metadata_hashes)[ROOT + "manifest.json"], digest(files[ROOT + "manifest.json"]))

   def test_the_frozen_plan_orders_the_pom_after_its_binaries(self) -> None:
      """The existing planner writes a module's POM last, so a POM is never visible without its aar."""
      files = capture()
      evidence = {"version": VERSION, "modules": [{key: value for key, value in module.items() if key != "version"} for module in group(files)]}
      plan = [item["path"] for item in legacy.qrun_maven_publication.plan_files(evidence, "owner", None) if "/library/" in item["path"]]
      self.assertTrue(plan[-1].endswith(".pom"))
      self.assertTrue(any(path.endswith(".aar") for path in plan[:-1]))

   def test_the_reactor_manifest_hash_is_the_retained_files_own_hash(self) -> None:
      """What a completion consumer binds to is the bundle's recorded SHA-256 of the exact manifest.json bytes; a layout without one has none."""
      files = capture()
      self.assertEqual(reactor_manifest_sha256(entries(files)), digest(files[ROOT + "manifest.json"]))
      self.assertIsNone(reactor_manifest_sha256(entries({"pom.xml": pom("library", "jar")})))


class ExtensionAndPackagingTest(unittest.TestCase):
   """The main extension follows the declared packaging; old supported attachments stay; missing mains and unknown files are rejected."""

   def test_aar_packaging_without_an_aar_is_refused(self) -> None:
      """A main jar is not substituted for a missing aar."""
      with self.assertRaisesRegex(PublicationError, "no matching aar"):
         group(capture(Capture(library_main="jar")))

   def test_jar_packaging_without_its_jar_is_refused_and_a_stray_aar_is_not_claimed(self) -> None:
      """The legacy missing-main rejection is kept, and an aar beside a jar module is unassigned and rejected."""
      with self.assertRaisesRegex(PublicationError, "no matching jar"):
         group({"pom.xml": pom("library", "jar"), f"target/library-{VERSION}-sources.jar": zip_bytes()})
      with self.assertRaisesRegex(PublicationError, "do not belong to any module"):
         group({"pom.xml": pom("library", "jar"), f"target/library-{VERSION}.jar": zip_bytes(), f"target/library-{VERSION}.aar": zip_bytes()})

   def test_the_plain_jar_layout_is_unchanged(self) -> None:
      """`pom.xml` with `target/*.jar` and a classifier jar still groups, with no reactor metadata involved."""
      files = {"pom.xml": pom("library", "jar"), f"target/library-{VERSION}.jar": zip_bytes(), f"target/library-{VERSION}-sources.jar": zip_bytes("S")}
      (module,) = group(files)
      self.assertEqual({(item["classifier"], item["extension"]) for item in module["files"]}, {(None, "pom"), (None, "jar"), ("sources", "jar")})

   def test_a_pom_module_keeps_the_jar_attachments_it_always_accepted(self) -> None:
      """Old behavior preserved: a pom-packaging module may carry a main jar and classifier jars, and needs none."""
      files = {"pom.xml": pom("parent", "pom"), f"target/parent-{VERSION}.jar": zip_bytes(), f"target/parent-{VERSION}-sources.jar": zip_bytes("S")}
      (module,) = group(files)
      self.assertEqual({(item["classifier"], item["extension"]) for item in module["files"]}, {(None, "pom"), (None, "jar"), ("sources", "jar")})
      self.assertEqual(len(group({"pom.xml": pom("parent", "pom")})[0]["files"]), 1)

   def test_arbitrary_extensions_are_not_publishable(self) -> None:
      """Any other file in or beside a module directory is rejected instead of being published or dropped."""
      for name in (f"files/library/library-{VERSION}.zip", f"files/library/library-{VERSION}-sources.aar", "files/library/notes.txt"):
         with self.subTest(name), self.assertRaises(PublicationError):
            group(capture(Capture(extra={name: b"data"})))

   def test_an_aar_that_is_not_an_archive_is_refused(self) -> None:
      """Bytes that are not a zip never reach the registry as an aar."""
      with self.assertRaisesRegex(PublicationError, "not a valid archive"):
         group(capture(Capture(aar=b"not a zip")))

   def test_gradle_module_metadata_must_name_its_own_coordinates(self) -> None:
      """A `.module` that names another artifact or version is a relabel and is refused; invalid JSON is refused."""
      for data in (gradle_module("other"), gradle_module("library", "9.9.9"), b"{not json"):
         with self.subTest(data=data[:20]), self.assertRaises(PublicationError):
            group(capture(Capture(module_data=data)))


def source_pom(artifact_id: str, *, version: str = "${revision}", group_id: str = GROUP) -> bytes:
   """Return an unresolved source POM as a flatten-plugin project keeps it (`${revision}` version)."""
   return f'<project xmlns="http://maven.apache.org/POM/4.0.0"><modelVersion>4.0.0</modelVersion><groupId>{group_id}</groupId><artifactId>{artifact_id}</artifactId><version>{version}</version></project>'.encode()


class FlattenedPomTest(unittest.TestCase):
   """`.flattened-pom.xml` is the consumer POM of its module; the unresolved source POM beside it is never published, and ambiguity is refused."""

   def layout(self, **overrides: bytes) -> dict[str, bytes]:
      """A `${revision}` project: source `pom.xml`, the flattened POM and the module jar under `target`."""
      files = {"pom.xml": source_pom("library"), ".flattened-pom.xml": pom("library", "jar"), f"target/library-{VERSION}.jar": zip_bytes(), f"target/library-{VERSION}-sources.jar": zip_bytes("S")}
      return {**files, **overrides}

   def test_the_flattened_pom_is_the_published_consumer_pom_and_the_source_pom_is_not(self) -> None:
      files = self.layout()
      (module,) = group(files)
      self.assertEqual((module["groupId"], module["artifactId"], module["version"]), (GROUP, "library", VERSION))
      carried = {(item["classifier"], item["extension"]): item["data"] for item in module["files"]}
      self.assertEqual(set(carried), {(None, "pom"), (None, "jar"), ("sources", "jar")})
      self.assertEqual(carried[(None, "pom")], files[".flattened-pom.xml"])
      self.assertEqual(maven_coordinates(artifact_for(files), entries(files)), (f"{GROUP}:library:{VERSION}",))

   def test_the_registry_receives_the_flattened_bytes_as_the_pom(self) -> None:
      files = self.layout()
      world = World()
      publish_maven_packages(AdapterInput(artifact=artifact_for(files), files=entries(files), request=cast("Any", None), target=PackageTarget(owner="owner"), transports=world.transports()))
      store = world.packages.store
      self.assertEqual(store[f"{REGISTRY_PREFIX}.pom"], files[".flattened-pom.xml"])
      self.assertNotIn(b"${revision}", b"".join(store.values()))

   def test_a_flattened_pom_alone_is_a_module(self) -> None:
      files = {".flattened-pom.xml": pom("parent", "pom")}
      (module,) = group(files)
      self.assertEqual((module["artifactId"], module["version"], [item["extension"] for item in module["files"]]), ("parent", VERSION, ["pom"]))

   def test_the_flattened_pom_must_carry_literal_coordinates(self) -> None:
      with self.assertRaisesRegex(PublicationError, "no literal version"):
         group(self.layout(**{".flattened-pom.xml": source_pom("library")}))

   def test_a_flattened_pom_beside_a_captured_pom_is_ambiguous(self) -> None:
      with self.assertRaisesRegex(PublicationError, "ambiguous"):
         group({".flattened-pom.xml": pom("library", "jar"), f"library-{VERSION}.pom": pom("library", "jar"), f"library-{VERSION}.jar": zip_bytes()})

   def test_a_source_pom_for_another_artifact_or_literal_version_is_refused(self) -> None:
      for name, source in (("artifact", source_pom("other")), ("version", source_pom("library", version="9.9.9")), ("group", source_pom("library", group_id="org.other"))):
         with self.subTest(name), self.assertRaisesRegex(PublicationError, "do not describe the same artifact|does not carry"):
            group(self.layout(**{"pom.xml": source}))

   def test_a_source_pom_with_a_matching_literal_version_is_accepted(self) -> None:
      (module,) = group(self.layout(**{"pom.xml": source_pom("library", version=VERSION)}))
      self.assertEqual(module["version"], VERSION)

   def test_two_flattened_modules_stay_separate_and_the_plain_layout_is_unaffected(self) -> None:
      files = {"a/pom.xml": source_pom("a"), "a/.flattened-pom.xml": pom("a", "pom"), "b/.flattened-pom.xml": pom("b", "pom")}
      self.assertEqual(sorted(module["artifactId"] for module in group(files)), ["a", "b"])
      self.assertEqual(len(group({"pom.xml": pom("parent", "pom")})), 1)

   def test_files_that_belong_to_no_module_are_still_refused_with_a_flattened_pom(self) -> None:
      with self.assertRaisesRegex(PublicationError, "do not belong to any module"):
         group(self.layout(**{"notes.txt": b"x"}))


class ReactorBindingTest(unittest.TestCase):
   """The retained reactor manifest must describe exactly the bytes and coordinates that are published."""

   def refused(self, options: Capture) -> None:
      """Assert the capture with these defects is refused."""
      with self.assertRaises(PublicationError):
         group(capture(options))

   def test_a_source_commit_other_than_the_build_commit_is_refused(self) -> None:
      """Both the SOURCE_COMMIT file and the manifest field must equal the build commit the publisher read from the forge."""
      self.refused(Capture(commit="d" * 40))

   def test_a_relabelled_version_is_refused(self) -> None:
      """The manifest version must equal the versions the POMs declare."""
      self.refused(Capture(edit=lambda manifest: manifest.update(version="9.9.9")))

   def test_a_manifest_entry_with_other_bytes_is_refused(self) -> None:
      """A manifest hash that differs from the retained file is refused."""

      def edit(manifest: dict[str, Any]) -> None:
         manifest["modules"][0]["files"][1]["sha256"] = "0" * 64

      self.refused(Capture(edit=edit))

   def test_a_module_missing_from_or_added_to_the_manifest_is_refused(self) -> None:
      """The manifest's modules must equal the grouped modules, in both directions."""
      self.refused(Capture(edit=lambda manifest: manifest["modules"].pop()))
      self.refused(Capture(edit=lambda manifest: manifest["modules"].append({"groupId": GROUP, "artifactId": "ghost", "packaging": "pom", "files": []})))

   def test_a_non_canonical_manifest_path_or_unknown_field_is_refused(self) -> None:
      """File paths must be the canonical Maven names, and the manifest keeps its exact key set."""

      def rename(manifest: dict[str, Any]) -> None:
         manifest["modules"][1]["files"][0]["path"] = "files/parent/renamed.pom"

      self.refused(Capture(edit=rename))
      self.refused(Capture(edit=lambda manifest: manifest.update(extra=True)))

   def test_checksums_must_cover_exactly_the_retained_files(self) -> None:
      """SHA256SUMS missing a file, missing entirely, or listing a wrong hash is refused."""
      self.refused(Capture(drop_sums_line="library-1.2.3.aar"))
      self.refused(Capture(skip_sums=True))
      files = capture()
      files[ROOT + "SHA256SUMS"] = files[ROOT + "SHA256SUMS"].replace(b"a", b"b", 1)
      with self.assertRaises(PublicationError):
         group(files)

   def test_an_undeclared_file_under_the_capture_root_is_refused(self) -> None:
      """A file the manifest does not describe cannot ride along, even with a plausible Maven name."""
      self.refused(Capture(extra={f"files/library/library-{VERSION}-javadoc.jar": zip_bytes("J")}))


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