"""Unit coverage for the fail-closed macOS rootless-Docker spike.""" from __future__ import annotations import unittest from dataclasses import dataclass from pathlib import Path from types import SimpleNamespace from typing import cast from unittest.mock import patch from bot_bottle.backend.macos_container import rootless_docker from bot_bottle.backend.macos_container import launch as launch_mod from bot_bottle.backend.macos_container.bottle_plan import MacosContainerBottlePlan class _Bottle: def __init__(self, results: list[SimpleNamespace]) -> None: self.results = results self.commands: list[str] = [] def exec(self, command: str) -> SimpleNamespace: self.commands.append(command) return self.results.pop(0) def _result(returncode: int, *, stdout: str = "", stderr: str = "") -> SimpleNamespace: return SimpleNamespace(returncode=returncode, stdout=stdout, stderr=stderr) @dataclass(frozen=True) class _AgentProvision: image: str @dataclass(frozen=True) class _Plan: slug: str image: str dockerfile_path: str docker_access: bool agent_provision: _AgentProvision class TestRootlessDockerStart(unittest.TestCase): def test_bootstraps_then_waits_for_guest_local_daemon(self) -> None: bottle = _Bottle([_result(0), _result(1), _result(0)]) with patch.object(rootless_docker.time, "sleep"): rootless_docker.start(bottle) self.assertIn("rootless-docker-init", bottle.commands[0]) self.assertEqual(2, bottle.commands.count("docker info >/dev/null 2>&1")) def test_bootstrap_failure_is_fatal_without_privilege_fallback(self) -> None: bottle = _Bottle([_result(1, stderr="newuidmap missing")]) with patch.object(rootless_docker, "die", side_effect=RuntimeError) as die: with self.assertRaises(RuntimeError): rootless_docker.start(bottle) self.assertIn("newuidmap missing", die.call_args.args[0]) self.assertEqual(1, len(bottle.commands)) def test_timeout_reports_guest_log(self) -> None: bottle = _Bottle( [_result(0)] + [_result(1) for _ in range(rootless_docker.READY_RETRIES)] + [_result(0, stdout="operation not permitted")] ) with patch.object(rootless_docker.time, "sleep"), \ patch.object(rootless_docker, "die", side_effect=RuntimeError) as die: with self.assertRaises(RuntimeError): rootless_docker.start(bottle) self.assertIn("operation not permitted", die.call_args.args[0]) class TestRootlessDockerImage(unittest.TestCase): def test_layers_tooling_without_changing_base_image(self) -> None: calls: list[tuple[str, str, str]] = [] def build(image: str, context: str, *, dockerfile: str) -> None: calls.append((image, context, dockerfile)) text = Path(dockerfile).read_text(encoding="utf-8") self.assertIn("FROM agent:base", text) self.assertIn("docker.io uidmap rootlesskit slirp4netns", text) self.assertIn("sed -i '/^node:/d' /etc/subuid /etc/subgid", text) self.assertEqual(1, text.count("node:100000:65536\\n' >> /etc/subuid")) self.assertEqual(1, text.count("node:100000:65536\\n' >> /etc/subgid")) self.assertIn("USER node", text) self.assertTrue((Path(context) / "rootless-docker-init.sh").is_file()) image = rootless_docker.build_image("agent:base", build) self.assertEqual("agent:base-rootless-docker", image) self.assertEqual("agent:base-rootless-docker", calls[0][0]) def test_launch_builds_base_then_rootless_variant(self) -> None: plan = cast(MacosContainerBottlePlan, cast(object, _Plan( slug="dev-abc", image="agent:base", dockerfile_path="/repo/Dockerfile", docker_access=True, agent_provision=_AgentProvision(image="agent:base"), ))) with patch.object(launch_mod, "read_committed_image", return_value=None), \ patch.object(launch_mod.container_mod, "build_image") as build, \ patch.object( launch_mod.rootless_docker, "build_image", return_value="agent:base-rootless-docker", ) as build_rootless: result = launch_mod._build_images(plan) # pylint: disable=protected-access build.assert_called_once_with( "agent:base", launch_mod._REPO_DIR, # pylint: disable=protected-access dockerfile="/repo/Dockerfile", ) build_rootless.assert_called_once_with("agent:base", build) self.assertEqual("agent:base-rootless-docker", result.agent_provision.image) if __name__ == "__main__": unittest.main()