platform-ci/tests/test_native_capabilities.py

"""Generic native build capabilities: tools, platforms, runtime selection, branch globs, Maven settings and native reporting.

Nothing here starts an application build. Runtime selection is tested against placeholder catalogs that name no real
image, and against the shipped catalog to prove that every unqualified runtime is blocked with a message.
"""

import json
import tempfile
import unittest
from collections.abc import Mapping, Sequence
from pathlib import Path
from typing import cast, override
from xml.etree import ElementTree

import yaml

from platform_ci.__main__ import main
from platform_ci.artifact_service import PRIVATE_FORGE_ORIGIN, ArtifactServiceError, require_private_artifact_service
from platform_ci.build_tools import BUILD_TOOLS, CONTAINER_EXECUTION, HOST_EXECUTION, PLATFORMS, TOOLCHAIN_DISPLAY
from platform_ci.configuration import ConfigurationError, branch_matches, load_project_configuration, parse_project_configuration
from platform_ci.execution import INHERITED_VARIABLES, ExecutionError, run_target_commands
from platform_ci.maven_settings import MavenSettingsError, install_public_central_settings, public_central_settings_document
from platform_ci.plan import EventContext, Plan, Target, build_matrix, has_execution, prepare_plan, to_compact_json
from platform_ci.reporting import render_report
from platform_ci.runtimes import DEFAULT_CATALOG_PATH, STANDARD_CONTAINER_RUNNER, RuntimeCatalogError, load_runtime_catalog, parse_runtime_catalog
from platform_ci.workspace import PipelineError
from tests.support import EXAMPLES_DIRECTORY, JAVA_IMAGE, NATIVE_RUNTIME_CATALOG, PLATFORM_REVISION, SOURCE_REVISION, event, example_configuration, example_mapping, placeholder_image, plan_for

WORKFLOW_PATH = Path(__file__).resolve().parent.parent / ".gitea" / "scoped_workflows" / "pipeline.yaml"
SCRATCH = "/tmp"  # noqa: S108 - the job container's scratch directory, which the tool environment names
EXPECTED_SETTINGS_DOCUMENT = """<?xml version="1.0" encoding="UTF-8"?>
<settings xmlns="http://maven.apache.org/SETTINGS/1.2.0">
  <mirrors>
    <mirror>
      <id>google-maven-central</id>
      <name>Google-hosted Maven Central mirror</name>
      <url>https://maven-central.storage-download.googleapis.com/maven2/</url>
      <mirrorOf>central</mirrorOf>
    </mirror>
  </mirrors>
<profiles>
<profile>
  <id>qrun-public-central-fallback</id>
  <repositories><repository>
  <id>central</id>
  <url>https://repo.maven.apache.org/maven2/</url>
  <releases><enabled>true</enabled></releases>
  <snapshots><enabled>false</enabled></snapshots>
</repository><repository>
  <id>canonical-maven-central</id>
  <url>https://repo.maven.apache.org/maven2/</url>
  <releases><enabled>true</enabled></releases>
  <snapshots><enabled>false</enabled></snapshots>
</repository></repositories>
  <pluginRepositories><pluginRepository>
  <id>central</id>
  <url>https://repo.maven.apache.org/maven2/</url>
  <releases><enabled>true</enabled></releases>
  <snapshots><enabled>false</enabled></snapshots>
</pluginRepository><pluginRepository>
  <id>canonical-maven-central</id>
  <url>https://repo.maven.apache.org/maven2/</url>
  <releases><enabled>true</enabled></releases>
  <snapshots><enabled>false</enabled></snapshots>
</pluginRepository></pluginRepositories>
</profile>
</profiles>
<activeProfiles>
<activeProfile>qrun-public-central-fallback</activeProfile>
</activeProfiles>
</settings>
"""


def mixed_plan(context: EventContext | None = None) -> Plan:
   """Plan the mixed example against the fully attested placeholder catalog."""
   return plan_for(example_configuration("native-mixed.ci.yaml"), context or event(), NATIVE_RUNTIME_CATALOG)


def targets_by_name(plan: Plan) -> dict[str, Target]:
   """Index a plan's targets by name."""
   return {target["name"]: target for target in plan["targets"]}


class ToolAndPlatformPlanTest(unittest.TestCase):
   """Each tool and platform expands to the right native commands, runtime and fixed environment."""

   def test_every_component_and_platform_becomes_one_target(self) -> None:
      """The mixed example yields one target per component and platform, with semantic names."""
      self.assertEqual(
         sorted(targets_by_name(mixed_plan())),
         [
            "android-library-linux-amd64",
            "cli-linux-amd64",
            "cli-macos-arm64",
            "cli-windows-amd64",
            "desktop-windows-amd64",
            "mac-app-macos-arm64",
            "plugin-linux-amd64",
            "service-linux-amd64",
            "tools-linux-amd64",
         ],
      )

   def test_native_commands_use_each_tools_own_executable_without_a_shell(self) -> None:
      """The executable is the tool's, followed by the declared arguments as separate tokens."""
      targets = targets_by_name(mixed_plan())
      self.assertEqual(targets["plugin-linux-amd64"]["commands"], [["./gradlew", "build"]])
      self.assertEqual(targets["service-linux-amd64"]["commands"], [["go", "vet", "./..."], ["go", "test", "-race", "./..."]])
      self.assertEqual(targets["tools-linux-amd64"]["commands"], [["python", "-m", "ruff", "check"], ["python", "-m", "pytest"]])
      self.assertEqual(targets["desktop-windows-amd64"]["commands"], [["dotnet", "test", "App.slnx", "--configuration", "Release"]])
      self.assertEqual(targets["mac-app-macos-arm64"]["commands"][0][:2], ["xcodebuild", "test"])
      self.assertEqual(targets["android-library-linux-amd64"]["commands"], [["./gradlew", "testDebugUnitTest"]])
      self.assertEqual(targets["cli-macos-arm64"]["commands"][0], ["cargo", "fmt", "--check"])

   def test_container_targets_use_catalog_images_on_the_standard_pool_and_host_targets_use_labels(self) -> None:
      """Only a container target has an image; a host target has an empty image and the catalog's runner label."""
      targets = targets_by_name(mixed_plan())
      for name, execution in (("service-linux-amd64", CONTAINER_EXECUTION), ("cli-linux-amd64", CONTAINER_EXECUTION), ("plugin-linux-amd64", CONTAINER_EXECUTION)):
         self.assertEqual((targets[name]["execution"], targets[name]["runner"]), (execution, STANDARD_CONTAINER_RUNNER), name)
         self.assertTrue(targets[name]["image"].startswith("registry.example/"), name)
      self.assertEqual(targets["plugin-linux-amd64"]["image"], JAVA_IMAGE)
      for name, runner in (("cli-macos-arm64", "test-macos-pool"), ("mac-app-macos-arm64", "test-macos-pool"), ("cli-windows-amd64", "test-windows-pool"), ("desktop-windows-amd64", "test-windows-pool")):
         self.assertEqual((targets[name]["execution"], targets[name]["runner"], targets[name]["image"]), (HOST_EXECUTION, runner, ""), name)

   def test_container_only_variables_are_not_applied_on_hosts(self) -> None:
      """Scratch-directory variables belong to the image; a host build keeps its own defaults. Telemetry opt-out applies everywhere."""
      targets = targets_by_name(mixed_plan())
      self.assertEqual(targets["cli-linux-amd64"]["environment"], {"CARGO_HOME": f"{SCRATCH}/cargo"})
      self.assertEqual(targets["cli-macos-arm64"]["environment"], {})
      self.assertEqual(targets["service-linux-amd64"]["environment"], {"GOPATH": f"{SCRATCH}/go", "GOCACHE": f"{SCRATCH}/go-cache"})
      self.assertEqual(targets["plugin-linux-amd64"]["environment"], {"GRADLE_USER_HOME": f"{SCRATCH}/.gradle", "GRADLE_OPTS": "-Xmx1g"})
      self.assertEqual(targets["desktop-windows-amd64"]["environment"], {"DOTNET_CLI_TELEMETRY_OPTOUT": "1", "DOTNET_NOLOGO": "1"})

   def test_display_names_describe_component_platform_and_toolchain(self) -> None:
      """Job names are semantic, never ordinal."""
      targets = targets_by_name(mixed_plan())
      self.assertEqual(targets["cli-windows-amd64"]["display"], "cli / Windows / Rust 1")
      self.assertEqual(targets["mac-app-macos-arm64"]["display"], "mac-app / macOS / Xcode 27")
      self.assertEqual(targets["desktop-windows-amd64"]["display"], "desktop / Windows / .NET 10.0.403")

   def test_matrices_split_by_execution_kind_and_a_kind_with_no_targets_is_empty(self) -> None:
      """Container and host targets are separate matrices; each entry carries its runner and its target as JSON."""
      plan = mixed_plan()
      container = build_matrix(plan)["include"]
      native = build_matrix(plan, HOST_EXECUTION)["include"]
      self.assertEqual(len(container) + len(native), len(plan["targets"]))
      self.assertEqual(sorted(entry["runner"] for entry in native), ["test-macos-pool", "test-macos-pool", "test-windows-pool", "test-windows-pool"])
      self.assertTrue(all(json.loads(entry["target"])["execution"] == HOST_EXECUTION for entry in native))
      linux_only = plan_for(example_configuration("maven.ci.yaml"), event())
      self.assertTrue(has_execution(linux_only, CONTAINER_EXECUTION))
      self.assertFalse(has_execution(linux_only, HOST_EXECUTION))
      self.assertEqual(build_matrix(linux_only, HOST_EXECUTION), {"include": []})

   def test_identical_configuration_gives_identical_plans_for_unrelated_repositories(self) -> None:
      """Two checkouts with unrelated names and the same mixed `.ci.yaml` get the same plan, so identity never selects behavior."""
      text = (EXAMPLES_DIRECTORY / "native-mixed.ci.yaml").read_text(encoding="utf-8")
      plans: list[str] = []
      for name in ("alpha-service", "zeta-unrelated-tool"):
         with tempfile.TemporaryDirectory() as parent:
            checkout = Path(parent) / name
            checkout.mkdir()
            (checkout / ".ci.yaml").write_text(text, encoding="utf-8")
            plans.append(to_compact_json(prepare_plan(load_project_configuration(checkout), event(), NATIVE_RUNTIME_CATALOG)))
      self.assertEqual(plans[0], plans[1])

   def test_default_platform_is_the_first_one_the_tool_supports(self) -> None:
      """A Swift component that names no platform runs on macOS, not on a platform it cannot use."""
      mapping: dict[str, object] = {"schema": 1, "components": {"app": {"toolchains": {"xcode": "27"}, "build": {"tool": "swift", "arguments": ["test"]}}}}
      self.assertEqual(parse_project_configuration(mapping).components[0].platforms, ("macos-arm64",))

   def test_the_tool_registry_is_internally_consistent(self) -> None:
      """No row names a platform or toolchain that does not exist, and a tool with profiles runs only in a catalog image."""
      for name, tool in BUILD_TOOLS.items():
         with self.subTest(name):
            self.assertTrue(set(tool.platforms) <= set(PLATFORMS))
            self.assertIn(tool.toolchain, TOOLCHAIN_DISPLAY)
            if tool.profiles and not tool.toolchain_choices:
               self.assertTrue(all(PLATFORMS[platform].execution == CONTAINER_EXECUTION for platform in tool.platforms))


class ConfigurationCapabilityTest(unittest.TestCase):
   """New fields are validated as strictly as the rest of `.ci.yaml`."""

   def _component(self, **changes: object) -> dict[str, object]:
      """Return a minimal Maven configuration with component fields changed."""
      component: dict[str, object] = {"toolchains": {"java": "21"}, "build": {"tool": "maven", "arguments": ["verify"]}}
      component.update(changes)
      return {"schema": 1, "components": {"library": component}}

   def test_unsupported_platform_and_profile_combinations_are_rejected_with_the_field(self) -> None:
      """A tool cannot be placed on a platform it does not support, and profiles need a catalog image."""
      cases: dict[str, tuple[dict[str, object], str]] = {
         "swift on linux": ({"schema": 1, "components": {"app": {"platforms": ["linux-amd64"], "toolchains": {"xcode": "27"}, "build": {"tool": "swift", "arguments": ["test"]}}}}, "unsupported platform 'linux-amd64' for build tool 'swift'"),
         "maven on windows": (self._component(platforms=["windows-amd64"]), "unsupported platform 'windows-amd64' for build tool 'maven'"),
         "unknown profile": (self._component(requires=["gpu"]), "components.library.requires: unsupported runtime profile 'gpu'"),
         "docker for node": ({"schema": 1, "components": {"web": {"toolchains": {"node": "22.20.0"}, "requires": ["docker"], "build": {"tool": "node", "arguments": ["test"]}}}}, "unsupported runtime profile 'docker' for build tool 'node'"),
         "duplicate profile": (self._component(requires=["browser", "browser"]), "duplicate runtime profile"),
         "profile not a list": (self._component(requires="browser"), "components.library.requires: expected a list"),
         "profile for a tool without profiles": ({"schema": 1, "components": {"core": {"toolchains": {"go": "1"}, "requires": ["browser"], "build": {"tool": "go", "arguments": ["test"]}}}}, "unsupported runtime profile 'browser' for build tool 'go'"),
         "toolchain of another tool": ({"schema": 1, "components": {"core": {"toolchains": {"java": "21"}, "build": {"tool": "go", "arguments": ["test"]}}}}, "components.core.toolchains.java: unsupported field"),
      }
      for description, (mapping, message) in cases.items():
         with self.subTest(description), self.assertRaises(ConfigurationError) as raised:
            parse_project_configuration(mapping)
         self.assertIn(message, str(raised.exception), description)

   def test_profiles_are_sorted_and_empty_means_standard(self) -> None:
      """Declared order does not change the plan."""
      self.assertEqual(parse_project_configuration(self._component(requires=["docker", "browser"])).components[0].requires, ("browser", "docker"))
      self.assertEqual(parse_project_configuration(self._component()).components[0].requires, ())


class BranchGlobTest(unittest.TestCase):
   """Branch patterns follow the trigger semantics the legacy dispatcher used: `*` inside a segment, `**` across segments."""

   def test_matching_table(self) -> None:
      """Exact names, single-segment stars and double stars behave as documented; dots are literal."""
      cases: tuple[tuple[str, str, bool], ...] = (
         ("main", "main", True),
         ("main", "main2", False),
         ("release/**", "release/1.0", True),
         ("release/**", "release/1.0/rc", True),
         ("release/**", "release", False),
         ("release/**", "hotfix/1", False),
         ("release/*", "release/1.0", True),
         ("release/*", "release/1.0/rc", False),
         ("*", "develop", True),
         ("*", "feature/x", False),
         ("qualification/**", "qualification/20261101", True),
         ("v1.0", "v1x0", False),
         ("feature-*", "feature-login", True),
         ("feature-*", "feature/login", False),
      )
      for pattern, branch, expected in cases:
         with self.subTest(pattern=pattern, branch=branch):
            self.assertEqual(branch_matches([pattern], branch), expected)

   def test_any_matching_pattern_selects_the_branch(self) -> None:
      """A list matches when any one pattern does."""
      self.assertTrue(branch_matches(["main", "hotfix/**"], "hotfix/urgent"))
      self.assertFalse(branch_matches([], "main"))

   def _glob_configuration(self) -> dict[str, object]:
      """Return the Maven example with the legacy dispatcher's glob triggers."""
      mapping = example_mapping("maven.ci.yaml")
      mapping["release"] = {"channels": {"candidate": {"trigger": {"push_branches": ["release/**", "hotfix/**"]}}}}
      mapping["verification"] = {"push": {"branches": ["qualification/**"], "tags": []}}
      return mapping

   def test_globs_select_verification_and_release_eligibility_through_the_plan(self) -> None:
      """A release-glob push verifies and is eligible; a verification-glob push verifies but is not; other branches do neither."""
      configuration = parse_project_configuration(self._glob_configuration())
      release = plan_for(configuration, event("push", "refs/heads/release/2.0"))
      self.assertEqual((release["verification"]["required"], release["delivery"]["eligible_channels"]), (True, ["candidate"]))
      qualification = plan_for(configuration, event("push", "refs/heads/qualification/20261101"))
      self.assertEqual((qualification["verification"]["required"], qualification["delivery"]["eligible_channels"]), (True, []))
      other = plan_for(configuration, event("push", "refs/heads/feature/x"))
      self.assertEqual((other["verification"]["required"], other["delivery"]["eligible_channels"]), (False, []))
      self.assertIn("does not select", other["verification"]["reason"])

   def test_the_browser_example_declares_the_legacy_globs(self) -> None:
      """The shipped example validates and lists release, hotfix and qualification globs."""
      configuration = example_configuration("maven-browser.ci.yaml")
      self.assertEqual(configuration.channels[0].push_branches, ("hotfix/**", "release/**"))
      self.assertEqual(configuration.push.branches, ("develop", "qualification/**"))
      self.assertEqual(configuration.components[0].requires, ("browser",))


class RuntimeSelectionTest(unittest.TestCase):
   """The catalog selects the runtime; anything unqualified is blocked by name, never substituted."""

   @override
   def setUp(self) -> None:
      """Load the catalog shipped with the platform revision."""
      self.shipped = load_runtime_catalog(DEFAULT_CATALOG_PATH)

   def test_the_shipped_catalog_still_resolves_the_qualified_maven_and_node_runtimes(self) -> None:
      """The published java and node images keep resolving on the standard pool."""
      for toolchain, version in (("java", "21"), ("node", "22.20.0")):
         runtime = self.shipped.resolve_runtime(toolchain, version, "linux-amd64")
         self.assertEqual((runtime.runner, runtime.image), (STANDARD_CONTAINER_RUNNER, self.shipped.resolve(toolchain, version)))

   def test_every_unqualified_runtime_is_blocked_with_a_message_naming_what_is_missing(self) -> None:
      """Go, Rust, Python, the Node browser image, .NET and Xcode are listed but not qualified, so prepare would stop."""
      cases: tuple[tuple[tuple[str, str, str, tuple[str, ...]], str], ...] = (
         (("go", "1", "linux-amd64", ()), "go 1 runtime is unqualified"),
         (("rust", "1", "linux-amd64", ()), "rust 1 runtime is unqualified"),
         (("python", "3.13", "linux-amd64", ()), "python 3.13 runtime is unqualified"),
         (("node", "22.20.0", "linux-amd64", ("browser",)), "node 22.20.0 browser runtime is unqualified"),
         (("go", "1", "macos-arm64", ()), "go 1 has no qualified macos-arm64 host runtime"),
         (("rust", "1", "windows-amd64", ()), "rust 1 has no qualified windows-amd64 host runtime"),
         (("dotnet", "10.0.403", "windows-amd64", ()), "dotnet 10.0.403 has no qualified windows-amd64 host runtime"),
         (("xcode", "27", "macos-arm64", ()), "xcode 27 has no qualified macos-arm64 host runtime"),
         (("go", "2", "linux-amd64", ()), "go 2 has no runtime in runtime/catalog.yaml; supported go versions: 1"),
      )
      for (toolchain, version, platform, requires), message in cases:
         with self.subTest(toolchain=toolchain, platform=platform, requires=requires), self.assertRaises(RuntimeCatalogError) as raised:
            self.shipped.resolve_runtime(toolchain, version, platform, requires)
         self.assertIn(message, str(raised.exception))

   def test_only_the_published_expanded_images_resolve_and_profiles_use_the_approved_generic_pool(self) -> None:
      """Java 17, Android and the BuildKit tools are recorded from root's receipts; the profile pools carry the one approved generic label."""
      expected = {
         ("java", "17"): "platform-ci-java17@sha256:5c4216201a5aa3307210cfaafa70c31d6886918b434d96cc1359206d60d4fc5d",
         ("android", "2026.10.1"): "platform-ci-android@sha256:76ffae3719a103a9ee76d7a17339baaa3476f031e8e9e5c6953997ddc66fb7a4",
         ("container", "0.34.0"): "platform-ci-container@sha256:3e34f9c4247853c823dc842c907cdc24126d3423a1c9c79ef06855fcdb0a37ba",
         ("control", "1"): "platform-ci-control@sha256:11f8cb328e7fc7aca8e0c39ed979b33cb1db411a59460baae3e0d48281a15e3c",
      }
      for (toolchain, version), reference in expected.items():
         with self.subTest(toolchain):
            runtime = self.shipped.resolve_runtime(toolchain, version, "linux-amd64")
            self.assertEqual(runtime.runner, STANDARD_CONTAINER_RUNNER)
            self.assertEqual(runtime.image, f"git.kof22.com/koftwentytwo/{reference}")
      self.assertEqual(self.shipped.browser_images["java"]["21"], "git.kof22.com/koftwentytwo/platform-ci-browser@sha256:4db3676ef7ecf307913d4869510ae5aff79e6a5f372c3fd01bcd53d170d92688")
      approved = {"browser": "container-build", "docker": "container-build", "browser+docker": "container-build", "image": "container-build", "docker+image": "container-build", "browser+image": None, "browser+docker+image": None}
      self.assertEqual(self.shipped.runners["linux-amd64"], approved)  # one generic label for every owner; draft qualification, not production activation
      for profile, image_kind in ((("browser",), "browser"), (("docker",), "standard"), (("browser", "docker"), "browser")):
         with self.subTest(profile):
            runtime = self.shipped.resolve_runtime("java", "21", "linux-amd64", profile)
            self.assertEqual(runtime.runner, "container-build")
            self.assertEqual(runtime.image, self.shipped.browser_images["java"]["21"] if image_kind == "browser" else self.shipped.resolve("java", "21"))
      with self.assertRaisesRegex(RuntimeCatalogError, "node 22.20.0 browser runtime is unqualified"):
         self.shipped.resolve_runtime("node", "22.20.0", "linux-amd64", ("browser",))

   def test_a_catalog_without_or_with_null_pool_labels_still_refuses_every_profile(self) -> None:
      """Custom catalogs keep the refusal: an unlisted or null profile pool blocks the component naming a profile."""
      for runners in ({}, {"linux-amd64": {"browser": None, "docker": None, "browser+docker": None}}):
         catalog = parse_runtime_catalog({"schema": 1, "runners": runners, "toolchains": {"java": {"21": {"image": JAVA_IMAGE, "browser_image": placeholder_image("browser")}}}})
         for profile in (("browser",), ("docker",), ("browser", "docker")):
            with self.subTest(runners=runners, profile=profile), self.assertRaisesRegex(RuntimeCatalogError, "has no qualified runner pool"):
               catalog.resolve_runtime("java", "21", "linux-amd64", profile)

   def test_unpublished_images_stay_blocked(self) -> None:
      """Go, Rust and Python have no published, clean-scanned digest, so none is recorded."""
      for toolchain, version in (("go", "1"), ("rust", "1"), ("python", "3.11"), ("python", "3.12"), ("python", "3.13"), ("python", "3.14")):
         with self.subTest(toolchain, version=version), self.assertRaisesRegex(RuntimeCatalogError, "is unqualified"):
            self.shipped.resolve_runtime(toolchain, version, "linux-amd64")

   def test_an_attested_host_without_a_runner_label_is_blocked(self) -> None:
      """The Windows label is unrecorded in the shipped catalog, so even an attested toolchain cannot be placed."""
      catalog = parse_runtime_catalog({"schema": 1, "runners": {"windows-amd64": {"standard": None}}, "toolchains": {"dotnet": {"10.0.403": {"image": None, "hosts": ["windows-amd64"]}}}})
      with self.assertRaisesRegex(RuntimeCatalogError, "no runner label is recorded for windows-amd64"):
         catalog.resolve_runtime("dotnet", "10.0.403", "windows-amd64")
      self.assertEqual(self.shipped.runners["windows-amd64"]["standard"], None)
      self.assertEqual(self.shipped.runners["macos-arm64"]["standard"], "kof-macos-arm64")

   def test_a_qualified_catalog_resolves_every_platform_and_the_browser_profile(self) -> None:
      """When the operator records the runtime, selection is by data only: image, label and profile image."""
      self.assertEqual(NATIVE_RUNTIME_CATALOG.browser_images["java"]["21"], placeholder_image("browser"))
      self.assertEqual(NATIVE_RUNTIME_CATALOG.resolve_runtime("dotnet", "10.0.403", "windows-amd64").runner, "test-windows-pool")
      # a recorded browser image is not enough: a catalog that records no profile pool (this test catalog has none) still refuses the profile
      with self.assertRaisesRegex(RuntimeCatalogError, "the browser runtime profile has no qualified runner pool"):
         NATIVE_RUNTIME_CATALOG.resolve_runtime("java", "21", "linux-amd64", ("browser",))

   def test_prepare_for_the_mixed_example_stops_on_the_shipped_catalog_before_publishing_outputs(self) -> None:
      """Planning with the shipped catalog fails on the first unqualified runtime and writes no workflow output."""
      with tempfile.TemporaryDirectory() as directory:
         checkout = Path(directory)
         outputs_file = checkout / "outputs"
         (checkout / ".ci.yaml").write_text((EXAMPLES_DIRECTORY / "native-mixed.ci.yaml").read_text(encoding="utf-8"), encoding="utf-8")
         environment = {"CI_EVENT_NAME": "pull_request", "CI_EVENT_REF": "refs/pull/1/merge", "CI_SOURCE_REVISION": SOURCE_REVISION, "CI_PLATFORM_REVISION": PLATFORM_REVISION, "GITHUB_OUTPUT": str(outputs_file)}
         with self.assertLogs("platform_ci.__main__", level="ERROR") as logged:
            self.assertEqual(main(["prepare", "--source", str(checkout)], environment, self.shipped), 1)
         self.assertIn("is unqualified", logged.output[0])
         self.assertFalse(outputs_file.exists())

   def test_shipped_standard_pool_matches_the_workflow(self) -> None:
      """The container job's pool is written once in the workflow; the code constant must agree."""
      workflow = cast("dict[str, dict[str, dict[str, object]]]", yaml.safe_load(WORKFLOW_PATH.read_text(encoding="utf-8")))
      self.assertEqual(workflow["jobs"]["verify"]["runs-on"], STANDARD_CONTAINER_RUNNER)

   def test_malformed_runner_and_host_entries_are_rejected(self) -> None:
      """Pool labels, host lists and browser images are validated as strictly as images."""
      base: dict[str, object] = {"schema": 1, "toolchains": {"go": {"1": {"image": None}}}}
      cases: dict[str, tuple[dict[str, object], str]] = {
         "unknown platform": ({**base, "runners": {"plan9": {"standard": "x"}}}, r"catalog\.runners\.plan9: unsupported entry"),
         "linux standard label": ({**base, "runners": {"linux-amd64": {"standard": "x"}}}, r"catalog\.runners\.linux-amd64\.standard: unsupported entry"),
         "pool label with a space": ({**base, "runners": {"linux-amd64": {"browser": "a b"}}}, "expected null or a runner label"),
         "host profile": ({**base, "runners": {"macos-arm64": {"browser": None}}}, r"catalog\.runners\.macos-arm64\.browser: unsupported entry"),
         "label with a space": ({**base, "runners": {"macos-arm64": {"standard": "a b"}}}, "expected null or a runner label"),
         "hosts not a list": ({"schema": 1, "toolchains": {"go": {"1": {"image": None, "hosts": "macos-arm64"}}}}, "hosts: expected a list of host platforms"),
         "linux as a host": ({"schema": 1, "toolchains": {"go": {"1": {"image": None, "hosts": ["linux-amd64"]}}}}, "hosts: expected host platforms"),
         "duplicate host": ({"schema": 1, "toolchains": {"go": {"1": {"image": None, "hosts": ["macos-arm64", "macos-arm64"]}}}}, "duplicate platform"),
         "mutable browser image": ({"schema": 1, "toolchains": {"java": {"21": {"image": None, "browser_image": "local/java-browser:298bba0a"}}}}, r"browser_image: expected null or a reference pinned by @sha256"),
      }
      for description, (raw, message) in cases.items():
         with self.subTest(description), self.assertRaisesRegex(RuntimeCatalogError, message):
            parse_runtime_catalog(raw)


class MavenSettingsTest(unittest.TestCase):
   """The public-central settings reproduce the captured adapter's creation mode and never touch credentials."""

   @override
   def setUp(self) -> None:
      """Create an empty home directory."""
      directory = tempfile.TemporaryDirectory()
      self.addCleanup(directory.cleanup)
      self.home = Path(directory.name)

   def test_the_document_matches_the_layout_derived_from_the_captured_adapter(self) -> None:
      """Byte-for-byte equal to the layout derived by reading the adapter's template code (see IMPLEMENTATION.md for the unexecuted-parity note)."""
      self.assertEqual(public_central_settings_document(), EXPECTED_SETTINGS_DOCUMENT)

   def test_the_document_is_well_formed_with_one_central_mirror_and_an_active_fallback_profile(self) -> None:
      """Parsing proves structure: a `mirrorOf central` mirror, snapshot-free public repositories, no servers or credentials."""
      namespace = {"m": "http://maven.apache.org/SETTINGS/1.2.0"}
      root = ElementTree.fromstring(public_central_settings_document())  # noqa: S314 - parses this module's own constant output
      mirrors = root.findall("m:mirrors/m:mirror", namespace)
      self.assertEqual([mirror.findtext("m:mirrorOf", namespaces=namespace) for mirror in mirrors], ["central"])
      self.assertEqual([element.text for element in root.findall("m:activeProfiles/m:activeProfile", namespace)], ["qrun-public-central-fallback"])
      self.assertEqual(root.findall("m:servers", namespace), [])
      for enabled in root.findall(".//m:snapshots/m:enabled", namespace):
         self.assertEqual(enabled.text, "false")
      for forbidden in ("password", "username", "token", "<server"):
         self.assertNotIn(forbidden, public_central_settings_document().lower())

   def test_install_writes_once_and_is_idempotent(self) -> None:
      """The file lands where Maven looks by itself, with LF endings, and a second identical call changes nothing."""
      path = install_public_central_settings(self.home)
      self.assertEqual(path, self.home / ".m2" / "settings.xml")
      self.assertEqual(path.read_bytes(), EXPECTED_SETTINGS_DOCUMENT.encode("utf-8"))
      modified = path.stat().st_mtime_ns
      self.assertEqual(install_public_central_settings(self.home), path)
      self.assertEqual(path.stat().st_mtime_ns, modified)

   def test_a_different_existing_file_is_refused_and_left_untouched(self) -> None:
      """A settings file that could hold credentials is never overwritten, read for values or merged."""
      path = self.home / ".m2" / "settings.xml"
      path.parent.mkdir()
      path.write_text("<settings/>", encoding="utf-8")
      with self.assertRaisesRegex(MavenSettingsError, "already exists with other content"):
         install_public_central_settings(self.home)
      self.assertEqual(path.read_text(encoding="utf-8"), "<settings/>")

   def test_a_directory_in_place_of_the_file_is_refused(self) -> None:
      """Only a regular file or nothing is acceptable."""
      (self.home / ".m2" / "settings.xml").mkdir(parents=True)
      with self.assertRaisesRegex(MavenSettingsError, "not a regular file"):
         install_public_central_settings(self.home)

   def test_a_symbolic_link_is_refused(self) -> None:
      """A link could redirect the write outside the job's home."""
      target = self.home / "elsewhere.xml"
      target.write_text("x", encoding="utf-8")
      link = self.home / ".m2" / "settings.xml"
      link.parent.mkdir()
      try:
         link.symlink_to(target)
      except OSError:
         self.skipTest("this platform does not allow creating symbolic links")
      with self.assertRaisesRegex(MavenSettingsError, "not a regular file"):
         install_public_central_settings(self.home)
      self.assertEqual(target.read_text(encoding="utf-8"), "x")


class RecordingRunner:
   """A command stub that records each invocation's command and environment."""

   def __init__(self) -> None:
      """Start with nothing recorded."""
      self.commands: list[list[str]] = []
      self.environments: list[Mapping[str, str]] = []

   def __call__(self, command: Sequence[str], working_directory: Path, environment: Mapping[str, str]) -> int:
      """Record the invocation and succeed."""
      del working_directory
      self.commands.append(list(command))
      self.environments.append(environment)
      return 0


class ExecutionCapabilityTest(unittest.TestCase):
   """Execution adds only the target's fixed variables and, for Maven, the public settings; never a credential."""

   @override
   def setUp(self) -> None:
      """Create a source checkout and a runner home."""
      directory = tempfile.TemporaryDirectory()
      self.addCleanup(directory.cleanup)
      self.source = Path(directory.name) / "source"
      self.source.mkdir()
      for name in ("cli", "service", "plugin", "tools", "desktop", "mac", "android"):
         (self.source / name).mkdir()
      self.home = Path(directory.name) / "home"
      self.home.mkdir()

   def test_maven_gets_the_public_settings_in_home_and_no_command_change(self) -> None:
      """The Maven command is exactly the declared one; the settings file is where Maven reads it without a flag."""
      target = plan_for(example_configuration("maven.ci.yaml"), event())["targets"][0]
      runner = RecordingRunner()
      self.assertTrue(run_target_commands(target, self.source, runner, {"PATH": "/usr/bin", "HOME": str(self.home)}))
      self.assertEqual(runner.commands, [["mvn", "--batch-mode", "verify"]])
      self.assertEqual((self.home / ".m2" / "settings.xml").read_text(encoding="utf-8"), EXPECTED_SETTINGS_DOCUMENT)
      self.assertEqual(runner.environments[0], {"PATH": "/usr/bin", "HOME": str(self.home)})

   def test_maven_without_a_home_fails_visibly_before_any_command(self) -> None:
      """An unknown settings location is an error, not a build without the mirror."""
      target = plan_for(example_configuration("maven.ci.yaml"), event())["targets"][0]
      runner = RecordingRunner()
      with self.assertRaisesRegex(ExecutionError, "HOME is not set"):
         run_target_commands(target, self.source, runner, {"PATH": "/usr/bin"})
      self.assertEqual(runner.commands, [])

   def test_other_tools_get_no_maven_settings_and_their_fixed_variables(self) -> None:
      """Node, Go, Gradle and .NET write no settings; their own non-secret variables are added to the sandbox."""
      targets = targets_by_name(mixed_plan())
      node = plan_for(example_configuration("node.ci.yaml"), event())["targets"][0]
      for target in (node, targets["service-linux-amd64"], targets["desktop-windows-amd64"]):
         runner = RecordingRunner()
         self.assertTrue(run_target_commands(target, self.source, runner, {"PATH": "/usr/bin", "HOME": str(self.home), "GITEA_TOKEN": "secret"}))
         self.assertEqual(runner.environments[0], {"PATH": "/usr/bin", "HOME": str(self.home)} | target["environment"])
      self.assertFalse((self.home / ".m2").exists())

   def test_host_system_variables_pass_but_runner_credentials_do_not(self) -> None:
      """A Windows or macOS host needs its system variables to start a tool; tokens and release secrets still never pass."""
      target = targets_by_name(mixed_plan())["desktop-windows-amd64"]
      runner = RecordingRunner()
      host_environment = {"PATH": "C:/tools", "SYSTEMROOT": "C:/Windows", "USERPROFILE": "C:/Users/runner", "PATHEXT": ".EXE", "DEVELOPER_DIR": "/Applications/Xcode.app", "ACTIONS_RUNTIME_TOKEN": "s", "GITEA_TOKEN": "s", "NUGET_API_KEY": "s"}
      run_target_commands(target, self.source, runner, host_environment)
      passed = set(runner.environments[0])
      self.assertEqual(passed, {"PATH", "SYSTEMROOT", "USERPROFILE", "PATHEXT", "DEVELOPER_DIR", "DOTNET_CLI_TELEMETRY_OPTOUT", "DOTNET_NOLOGO"})
      self.assertTrue(passed <= set(INHERITED_VARIABLES) | set(target["environment"]))

   def test_a_component_directory_outside_the_checkout_is_still_rejected(self) -> None:
      """New tools reuse the same containment check."""
      target = targets_by_name(mixed_plan())["service-linux-amd64"]
      target["directory"] = "../elsewhere"
      with self.assertRaisesRegex(PipelineError, "not a directory inside the source checkout"):
         run_target_commands(target, self.source, RecordingRunner(), {"PATH": "/usr/bin"})


class NativeReportingTest(unittest.TestCase):
   """The report expects exactly one successful job per execution kind that has targets, and a skipped job for the other."""

   def test_a_plan_with_both_kinds_needs_both_jobs_and_the_scan(self) -> None:
      """Only three successes pass."""
      plan = mixed_plan()
      self.assertEqual(render_report(plan, "success", "success", "success")[1], True)
      for combination in (("failure", "success", "success"), ("success", "success", "failure"), ("success", "skipped", "success"), ("success", "failure", "success"), ("skipped", "skipped", "skipped")):
         with self.subTest(combination):
            self.assertFalse(render_report(plan, combination[0], combination[2], combination[1])[1])

   def test_a_host_only_plan_expects_the_container_job_to_be_skipped(self) -> None:
      """No container target means a container result of success is inconsistent, not extra assurance."""
      mapping: dict[str, object] = {"schema": 1, "components": {"desktop": {"platforms": ["windows-amd64"], "toolchains": {"dotnet": "10.0.403"}, "build": {"tool": "dotnet", "arguments": ["test"]}}}}
      plan = plan_for(parse_project_configuration(mapping), event(), NATIVE_RUNTIME_CATALOG)
      self.assertTrue(render_report(plan, "skipped", "success", "success")[1])
      self.assertFalse(render_report(plan, "success", "success", "success")[1])
      text, passed = render_report(plan, "skipped", "success", "failure")
      self.assertFalse(passed)
      self.assertIn("FAILED (native verification did not succeed)", text)
      self.assertIn("Native host verification result: failure", text)

   def test_a_missing_native_result_defaults_to_skipped_and_fails_when_host_targets_exist(self) -> None:
      """Omitting the fourth result cannot make a host-target plan pass."""
      self.assertFalse(render_report(mixed_plan(), "success", "success")[1])
      container_only = plan_for(example_configuration("maven.ci.yaml"), event())
      self.assertTrue(render_report(container_only, "success", "success")[1])

   def test_unrequested_work_passes_only_when_every_job_skipped(self) -> None:
      """An unselected push passes as not verified and fails if any job ran."""
      plan = mixed_plan(event("push", "refs/heads/feature-x"))
      self.assertFalse(plan["verification"]["required"])
      self.assertTrue(render_report(plan, "skipped", "skipped", "skipped")[1])
      self.assertFalse(render_report(plan, "skipped", "skipped", "success")[1])


class CommandLineCapabilityTest(unittest.TestCase):
   """The command line publishes the split outputs and handles the new subcommand."""

   @override
   def setUp(self) -> None:
      """Create a checkout directory."""
      directory = tempfile.TemporaryDirectory()
      self.addCleanup(directory.cleanup)
      self.checkout = Path(directory.name)
      self.outputs_file = self.checkout / "outputs"

   def _prepare(self, example: str, name: str = "pull_request", ref: str = "refs/pull/1/merge") -> dict[str, str]:
      """Run `prepare` on an example with the attested catalog and return the workflow outputs."""
      (self.checkout / ".ci.yaml").write_text((EXAMPLES_DIRECTORY / example).read_text(encoding="utf-8"), encoding="utf-8")
      environment = {"CI_EVENT_NAME": name, "CI_EVENT_REF": ref, "CI_DEFAULT_BRANCH": "main", "CI_SOURCE_REVISION": SOURCE_REVISION, "CI_PLATFORM_REVISION": PLATFORM_REVISION, "GITHUB_OUTPUT": str(self.outputs_file)}
      self.assertEqual(main(["prepare", "--source", str(self.checkout)], environment, NATIVE_RUNTIME_CATALOG), 0)
      return dict(line.split("=", 1) for line in self.outputs_file.read_text(encoding="utf-8").splitlines())

   def test_a_mixed_plan_publishes_both_matrices_and_both_flags(self) -> None:
      """Each verification job gets its own matrix and a flag that is true only when it has work."""
      outputs = self._prepare("native-mixed.ci.yaml")
      self.assertEqual((outputs["verify"], outputs["container_targets"], outputs["native_targets"]), ("true", "true", "true"))
      self.assertEqual(len(json.loads(outputs["matrix"])["include"]), 5)
      self.assertEqual(len(json.loads(outputs["native_matrix"])["include"]), 4)

   def test_a_container_only_plan_leaves_the_native_job_unrequested(self) -> None:
      """No host target means native_targets is false and its matrix is empty."""
      outputs = self._prepare("maven.ci.yaml")
      self.assertEqual((outputs["container_targets"], outputs["native_targets"]), ("true", "false"))
      self.assertEqual(json.loads(outputs["native_matrix"]), {"include": []})

   def test_unrequested_verification_requests_no_job_of_either_kind(self) -> None:
      """An unselected push reports false for both kinds even though targets exist."""
      outputs = self._prepare("native-mixed.ci.yaml", "push", "refs/heads/feature-x")
      self.assertEqual((outputs["verify"], outputs["container_targets"], outputs["native_targets"]), ("false", "false", "false"))

   def test_cache_metadata_for_a_tool_without_a_cache_reports_unavailable_and_succeeds(self) -> None:
      """Gradle has no qualified download directory, so the step publishes available=false instead of failing."""
      target = targets_by_name(mixed_plan())["plugin-linux-amd64"]
      environment = {"CI_TARGET": to_compact_json(target), "CI_EVENT_NAME": "pull_request", "CI_EVENT_REF": "refs/pull/1/merge", "GITHUB_OUTPUT": str(self.outputs_file), "ACTIONS_CACHE_URL": "http://10.0.0.1:44300/"}
      self.assertEqual(main(["cache-metadata", "--source", str(self.checkout)], environment), 0)
      outputs = dict(line.split("=", 1) for line in self.outputs_file.read_text(encoding="utf-8").splitlines())
      self.assertEqual((outputs["available"], outputs["save"]), ("false", "false"))
      self.assertIn("no dependency cache is defined for build tool 'gradle'", outputs["note"])

   def test_require_artifact_service_command_uses_the_environment(self) -> None:
      """The command exits zero for the private forge and one otherwise."""
      good = {"ACTIONS_RUNTIME_URL": f"{PRIVATE_FORGE_ORIGIN}/api", "ACTIONS_RESULTS_URL": ""}
      self.assertEqual(main(["require-artifact-service"], good), 0)
      with self.assertLogs("platform_ci.__main__", level="ERROR"):
         self.assertEqual(main(["require-artifact-service"], {"ACTIONS_RUNTIME_URL": "https://example.invalid"}), 1)


class ArtifactServiceTest(unittest.TestCase):
   """Only the private forge origin, or a path under it, is accepted."""

   def test_accepts_the_origin_and_paths_under_it(self) -> None:
      """An empty results endpoint is allowed, as in the container job's shell check."""
      for runtime, results in ((PRIVATE_FORGE_ORIGIN, ""), (f"{PRIVATE_FORGE_ORIGIN}/", f"{PRIVATE_FORGE_ORIGIN}/results"), (f"{PRIVATE_FORGE_ORIGIN}/a/b", "")):
         with self.subTest(runtime=runtime, results=results):
            require_private_artifact_service({"ACTIONS_RUNTIME_URL": runtime, "ACTIONS_RESULTS_URL": results})

   def test_rejects_lookalikes_other_schemes_and_a_missing_runtime(self) -> None:
      """A prefix match without a path boundary, plain HTTP and a missing runtime URL all fail."""
      for runtime, results in (("", ""), (f"{PRIVATE_FORGE_ORIGIN}.evil.example/", ""), ("http://git.kof22.com", ""), (PRIVATE_FORGE_ORIGIN, "https://elsewhere.example"), (PRIVATE_FORGE_ORIGIN, f"{PRIVATE_FORGE_ORIGIN}.evil.example")):
         with self.subTest(runtime=runtime, results=results), self.assertRaises(ArtifactServiceError):
            require_private_artifact_service({"ACTIONS_RUNTIME_URL": runtime, "ACTIONS_RESULTS_URL": results})

   def test_the_container_jobs_shell_guard_names_the_same_origin(self) -> None:
      """The shell check in the container job and this module cannot drift apart silently."""
      text = WORKFLOW_PATH.read_text(encoding="utf-8")
      self.assertIn(f'case "$ACTIONS_RUNTIME_URL" in {PRIVATE_FORGE_ORIGIN}|{PRIVATE_FORGE_ORIGIN}/*)', text)


class NativeWorkflowTest(unittest.TestCase):
   """The native job is wired like the container job but starts no container and names no runner."""

   @override
   def setUp(self) -> None:
      """Parse the one workflow."""
      self.text = WORKFLOW_PATH.read_text(encoding="utf-8")
      self.jobs = cast("dict[str, dict[str, object]]", yaml.safe_load(self.text)["jobs"])
      self.native = self.jobs["verify-native"]
      self.steps = cast("list[dict[str, object]]", self.native["steps"])

   def _step(self, prefix: str) -> dict[str, object]:
      """Return the native job's step whose name starts with the prefix."""
      return next(step for step in self.steps if str(step.get("name", "")).startswith(prefix))

   def test_each_verification_job_runs_only_when_its_kind_has_targets(self) -> None:
      """Both conditions are plan outputs; neither is a name or an event shortcut."""
      self.assertEqual(self.jobs["verify"]["if"], "${{ needs.prepare.outputs.container_targets == 'true' }}")
      self.assertEqual(self.native["if"], "${{ needs.prepare.outputs.native_targets == 'true' }}")
      self.assertEqual(self.native["needs"], ["prepare"])

   def test_the_native_job_uses_the_plans_runner_and_starts_no_container(self) -> None:
      """Placement comes from the matrix, so the workflow contains no runner label, and a host job has no container."""
      self.assertEqual(self.native["runs-on"], "${{ matrix.runner }}")
      self.assertNotIn("container", self.native)
      self.assertEqual(cast("dict[str, object]", self.native["strategy"])["matrix"], "${{ fromJSON(needs.prepare.outputs.native_matrix) }}")
      for label in ("kof-windows", "qqq-windows", "kof-macos", "browser-build", "container-build"):
         self.assertNotIn(label, self.text)

   def test_the_native_job_builds_once_then_collects_and_retains_like_the_container_job(self) -> None:
      """One build step, then collect with the same environment, then the same pinned upload action."""
      build = self._step("Build and test")
      self.assertEqual(build["run"], "python3 -P -m platform_ci verify --source .")
      self.assertEqual(sum(1 for step in self.steps if "platform_ci verify" in str(step.get("run", ""))), 1)
      container_package = next(step for step in cast("list[dict[str, object]]", self.jobs["verify"]["steps"]) if str(step.get("name", "")).startswith("Collect"))
      self.assertEqual(self._step("Collect"), container_package)
      container_upload = next(step for step in cast("list[dict[str, object]]", self.jobs["verify"]["steps"]) if str(step.get("name", "")).startswith("Retain"))
      self.assertEqual(self._step("Retain"), container_upload)

   def test_the_native_job_checks_the_artifact_service_with_the_portable_command(self) -> None:
      """A host shell may not be POSIX, so the guard is the CLI command, not a `case` statement."""
      self.assertEqual(self._step("Require private artifact service")["run"], "python3 -P -m platform_ci require-artifact-service")

   def test_the_native_job_has_no_cache_and_no_install_step(self) -> None:
      """No host download directory is qualified, and a host job installs nothing."""
      for step in self.steps:
         self.assertNotIn("cache", str(step.get("uses", "")))
         self.assertNotRegex(str(step.get("run", "")), r"\b(curl|wget|apt|brew|choco|winget|pip|npm install)\b")


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