platform-ci/tests/test_docker_environment.py

"""The docker profile's fixed pool contract reaches the build process, and only for targets that require docker."""

import tempfile
import unittest
from collections.abc import Mapping, Sequence
from pathlib import Path
from typing import Any, cast

from platform_ci.build_tools import DOCKER_PROFILE_ENVIRONMENT
from platform_ci.execution import run_target_commands
from tests.test_pool_jobs import DOCKER_VARIABLES, JOBS
from tests.test_pool_jobs import _plan as pool_plan  # pyright: ignore[reportPrivateUsage]

RELEASE_CREDENTIALS = {"GITEA_TOKEN": "x", "ACTIONS_RUNTIME_TOKEN": "x", "ACTIONS_ID_TOKEN_REQUEST_TOKEN": "x", "CI_RELEASE_TOKEN": "x", "MANAGER_PULL_AUTH": "x"}


def _environment_seen_by_build(profiles: Sequence[str], parent_extra: Mapping[str, str] | None = None) -> dict[str, str]:
   """Plan a Maven component with the profiles, run its commands with a recording runner and return the environment the build process received."""
   (target,) = pool_plan(*profiles)["targets"]
   seen: list[dict[str, str]] = []

   def runner(_command: Sequence[str], _directory: Path, environment: Mapping[str, str]) -> int:
      seen.append(dict(environment))
      return 0

   with tempfile.TemporaryDirectory() as directory:
      source = Path(directory)
      (source / "module").mkdir()
      home = source / "home"
      home.mkdir()
      parent = {"HOME": str(home), "PATH": "/usr/bin", **RELEASE_CREDENTIALS, **(parent_extra or {})}
      self_target = cast("Any", dict(target) | {"directory": "."})
      self_target["commands"] = [["true"]]
      self_target["build_tool"] = "maven"
      assert run_target_commands(self_target, source, runner, parent)
   return seen[0]


class DockerEnvironmentTest(unittest.TestCase):
   """Planned environment, allow-list filtering and the shared workflow job agree."""

   def test_docker_targets_plan_and_receive_exactly_the_fixed_contract(self) -> None:
      for profiles in (("docker",), ("browser", "docker")):
         with self.subTest(profiles):
            (target,) = pool_plan(*profiles)["targets"]
            self.assertEqual({key: target["environment"][key] for key in DOCKER_VARIABLES}, dict(DOCKER_PROFILE_ENVIRONMENT))
            received = _environment_seen_by_build(profiles)
            self.assertEqual({key: received[key] for key in DOCKER_VARIABLES}, {"DOCKER_HOST": "unix:///run/test-docker.sock", "TESTCONTAINERS_DOCKER_SOCKET_OVERRIDE": "/run/user/1002/docker.sock", "TESTCONTAINERS_HOST_OVERRIDE": "172.17.0.1"})

   def test_plain_and_browser_only_targets_get_no_docker_variable(self) -> None:
      for profiles in ((), ("browser",)):
         with self.subTest(profiles):
            (target,) = pool_plan(*profiles)["targets"]
            self.assertTrue(set(DOCKER_VARIABLES).isdisjoint(target["environment"]))
            self.assertTrue(set(DOCKER_VARIABLES).isdisjoint(_environment_seen_by_build(profiles)))

   def test_the_runner_cannot_inject_docker_variables_into_a_target_that_does_not_require_docker(self) -> None:
      injected = {"DOCKER_HOST": "unix:///var/run/docker.sock", "TESTCONTAINERS_HOST_OVERRIDE": "10.0.0.1", "DOCKER_CONFIG": "/root/.docker"}
      for profiles in ((), ("browser",)):
         with self.subTest(profiles):
            self.assertTrue(set(injected).isdisjoint(_environment_seen_by_build(profiles, injected)))

   def test_a_runner_value_never_overrides_the_planned_docker_contract(self) -> None:
      received = _environment_seen_by_build(("docker",), {"DOCKER_HOST": "unix:///var/run/docker.sock", "DOCKER_CONFIG": "/root/.docker"})
      self.assertEqual(received["DOCKER_HOST"], "unix:///run/test-docker.sock")
      self.assertNotIn("DOCKER_CONFIG", received)

   def test_release_and_manager_credentials_stay_filtered_for_every_profile(self) -> None:
      for profiles in ((), ("browser",), ("docker",), ("browser", "docker")):
         with self.subTest(profiles):
            self.assertTrue(set(RELEASE_CREDENTIALS).isdisjoint(_environment_seen_by_build(profiles)))

   def test_the_plan_contract_equals_what_the_shared_docker_pool_job_sets(self) -> None:
      job_environment = JOBS["verify-pool-docker"]["env"]
      self.assertEqual({key: job_environment[key] for key in DOCKER_VARIABLES}, dict(DOCKER_PROFILE_ENVIRONMENT))
      self.assertTrue(set(DOCKER_VARIABLES).isdisjoint(JOBS["verify-pool"]["env"]))
      self.assertTrue(set(DOCKER_VARIABLES).isdisjoint(JOBS["verify"].get("env", {})))