334 lines
14 KiB
Python
334 lines
14 KiB
Python
"""Exercise Docker bootstrap with isolated command stubs, never the host daemon."""
|
|
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import shlex
|
|
import shutil
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
MODULE = ROOT / "scripts/lib/docker-bootstrap.zsh"
|
|
|
|
COMMAND_STUB = r"""
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
|
|
state_path = Path(os.environ["DOCKER_TEST_STATE"])
|
|
state = json.loads(state_path.read_text())
|
|
name = Path(sys.argv[0]).name
|
|
args = sys.argv[1:]
|
|
state["calls"].append([name, args])
|
|
|
|
def save() -> None:
|
|
state_path.write_text(json.dumps(state))
|
|
|
|
def finish(code: int = 0, output: str = "") -> None:
|
|
save()
|
|
if output:
|
|
print(output)
|
|
sys.exit(code)
|
|
|
|
def install_command(name: str) -> None:
|
|
target = state_path.parent / "bin" / name
|
|
if not target.exists():
|
|
target.symlink_to("mock-command")
|
|
|
|
if name == "id":
|
|
if args == ["-u"]:
|
|
finish(output=str(state["uid"]))
|
|
if args == ["-un"]:
|
|
finish(output="planet-test")
|
|
finish(output="planet-test" + (" docker" if state["member"] else ""))
|
|
if name == "sudo":
|
|
if state.get("sudo_denied"):
|
|
finish(1)
|
|
if args == ["-v"]:
|
|
finish()
|
|
if "-E" in args:
|
|
finish() # Record re-exec argv without executing planet.sh on this machine.
|
|
save()
|
|
sys.exit(subprocess.call(args[1:] if args[0] == "--" else args))
|
|
if name == "apt-get":
|
|
if state.get("apt_fail"):
|
|
finish(1, "simulated apt failure")
|
|
if args[0] == "install":
|
|
for package in args[2:]:
|
|
if package in ("docker.io", "docker-ce"):
|
|
state["engine"] = True
|
|
install_command("docker")
|
|
install_command("dockerd")
|
|
if package in ("docker-compose-v2", "docker-compose-plugin"):
|
|
state["compose"] = True
|
|
if package in ("docker-buildx", "docker-buildx-plugin"):
|
|
state["buildx"] = "0.20.0"
|
|
if package == "passwd":
|
|
install_command("usermod")
|
|
finish()
|
|
if name == "dpkg-query":
|
|
finish(0 if state.get("ce") else 1, "install ok installed" if state.get("ce") else "")
|
|
if name == "systemctl":
|
|
if args[0] == "show":
|
|
unit_exists = state["engine"] and (args[-1] != "docker.socket" or state["socket_unit"])
|
|
finish(output="loaded" if unit_exists else "not-found")
|
|
if args[0] == "is-active":
|
|
finish(output="active" if state["running"] else "inactive")
|
|
if state.get("service_fail"):
|
|
finish(1, "simulated service failure")
|
|
state["running"] = True
|
|
finish()
|
|
if name == "usermod":
|
|
state["member"] = True
|
|
finish()
|
|
if name == "mock-socket-access":
|
|
finish(0 if state["running"] and state["uid"] != 0 and not state["access"] else 1)
|
|
if name == "docker":
|
|
if args[0] == "info":
|
|
finish(0 if state["running"] and (state["uid"] == 0 or state["access"]) else 1)
|
|
if args[:2] == ["context", "inspect"]:
|
|
finish(output=state["endpoint"])
|
|
if args[:2] == ["compose", "version"]:
|
|
finish(0 if state["compose"] else 1)
|
|
if args[:2] == ["buildx", "version"]:
|
|
finish(0 if state["buildx"] else 1, "github.com/docker/buildx v" + state["buildx"])
|
|
finish(1)
|
|
"""
|
|
|
|
|
|
class DockerBootstrapTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.temporary = tempfile.TemporaryDirectory(prefix="planet-docker-test-")
|
|
self.addCleanup(self.temporary.cleanup)
|
|
self.folder = Path(self.temporary.name)
|
|
self.bin = self.folder / "bin"
|
|
self.bin.mkdir()
|
|
self.state_file = self.folder / "state.json"
|
|
self.state = {
|
|
"calls": [],
|
|
"uid": 1000,
|
|
"member": False,
|
|
"access": False,
|
|
"engine": False,
|
|
"running": False,
|
|
"compose": False,
|
|
"buildx": "",
|
|
"endpoint": "unix:///var/run/docker.sock",
|
|
"socket_unit": True,
|
|
}
|
|
stub = self.bin / "mock-command"
|
|
stub.write_text(f"#!{sys.executable}\n" + COMMAND_STUB)
|
|
stub.chmod(0o755)
|
|
for name in ("id", "sudo", "apt-get", "systemctl", "dpkg-query", "mock-socket-access"):
|
|
(self.bin / name).symlink_to("mock-command")
|
|
for name in ("zsh", "env", "sort", "head", "tail", "readlink", "awk", "sed", "python3",
|
|
"mktemp", "sleep", "cat", "rm"):
|
|
executable = shutil.which(name)
|
|
self.assertIsNotNone(executable, f"test prerequisite missing: {name}")
|
|
(self.bin / name).symlink_to(executable)
|
|
|
|
def existing_engine(self, *, running: bool = True, access: bool = True) -> None:
|
|
for name in ("docker", "dockerd"):
|
|
(self.bin / name).symlink_to("mock-command")
|
|
self.state.update(
|
|
engine=True, running=running, access=access, compose=True, buildx="0.20.0"
|
|
)
|
|
|
|
def run_bootstrap(
|
|
self,
|
|
action: str = "ensure_docker_runtime",
|
|
*,
|
|
extra_env: dict[str, str] | None = None,
|
|
arguments: list[str] | None = None,
|
|
) -> subprocess.CompletedProcess[str]:
|
|
self.state_file.write_text(json.dumps(self.state))
|
|
env = os.environ.copy()
|
|
for name in ("DOCKER_HOST", "DOCKER_CONTEXT", "PLANET_DOCKER_GROUP_REEXEC"):
|
|
env.pop(name, None)
|
|
env.update(
|
|
PATH=str(self.bin),
|
|
DOCKER_TEST_STATE=str(self.state_file),
|
|
TEST_OS="ubuntu",
|
|
TEST_DESKTOP="0",
|
|
PLANET_STATE_DIR=str(self.folder),
|
|
)
|
|
env.update(extra_env or {})
|
|
arguments = arguments or ["init", "--non-motion-agent"]
|
|
prelude = f"""
|
|
set -e
|
|
SCRIPT_DIR={shlex.quote(str(self.folder / "repo with ' quotes"))}
|
|
source {shlex.quote(str(MODULE))}
|
|
log_note() {{ print -r -- "$*"; }}
|
|
log_error() {{ print -r -- "$*"; }}
|
|
log_step() {{ print -r -- "$*"; }}
|
|
docker_bootstrap_os_id() {{ print -r -- "$TEST_OS"; }}
|
|
docker_desktop_present() {{ [[ "$TEST_DESKTOP" == 1 ]]; }}
|
|
docker_socket_needs_group_access() {{ mock-socket-access; }}
|
|
run_command_quiet_unless_verbose() {{ local output="$1"; shift; "$@" > "$output" 2>&1; }}
|
|
ensure_system_command() {{ command -v "$1" >/dev/null || docker_as_root apt-get install -y "$2"; }}
|
|
{action} {shlex.join(arguments)}
|
|
"""
|
|
result = subprocess.run(
|
|
[str(self.bin / "zsh"), "-f", "-c", prelude],
|
|
env=env,
|
|
text=True,
|
|
capture_output=True,
|
|
timeout=20,
|
|
)
|
|
self.state = json.loads(self.state_file.read_text())
|
|
return result
|
|
|
|
def calls(self, name: str) -> list[list[str]]:
|
|
return [args for command, args in self.state["calls"] if command == name]
|
|
|
|
def test_working_environment_is_reused_without_privilege_or_installation(self) -> None:
|
|
self.existing_engine()
|
|
result = self.run_bootstrap()
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
self.assertEqual(self.calls("apt-get"), [])
|
|
self.assertEqual(self.calls("sudo"), [])
|
|
self.assertEqual(self.calls("systemctl"), [])
|
|
|
|
def test_fresh_root_install_starts_and_validates_engine(self) -> None:
|
|
self.state["uid"] = 0
|
|
result = self.run_bootstrap()
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
self.assertEqual(
|
|
self.calls("apt-get"),
|
|
[
|
|
["update", "--error-on=any"],
|
|
["install", "-y", "docker.io", "docker-compose-v2", "docker-buildx"],
|
|
],
|
|
)
|
|
self.assertIn(["enable", "--now", "docker.service"], self.calls("systemctl"))
|
|
self.assertTrue(self.state["running"])
|
|
|
|
def test_fresh_user_installs_missing_usermod_and_reexecutes_without_sg(self) -> None:
|
|
args = ["--verbose", "init", "--non-motion-agent", "a 'quoted' $(argument)"]
|
|
result = self.run_bootstrap(arguments=args)
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
self.assertIn(["install", "-y", "passwd"], self.calls("apt-get"))
|
|
self.assertEqual(self.calls("usermod"), [["-aG", "docker", "planet-test"]])
|
|
reexec = self.calls("sudo")[-1]
|
|
self.assertEqual(reexec[:7], ["-E", "-u", "planet-test", "-g", "docker", "--", "env"])
|
|
self.assertEqual(reexec[7], "PLANET_DOCKER_GROUP_REEXEC=1")
|
|
self.assertEqual(reexec[8], f"PATH={self.bin}")
|
|
self.assertEqual(reexec[10], str(self.folder / "repo with ' quotes/planet.sh"))
|
|
self.assertEqual(reexec[11:], args)
|
|
self.assertFalse((self.bin / "sg").exists())
|
|
|
|
def test_existing_group_membership_refreshes_start_without_installing(self) -> None:
|
|
self.existing_engine(access=False)
|
|
self.state["member"] = True
|
|
result = self.run_bootstrap("refresh_docker_group", arguments=["start", "--allow-lan"])
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
self.assertEqual(self.calls("sudo")[-1][-2:], ["start", "--allow-lan"])
|
|
self.assertEqual(self.calls("apt-get"), [])
|
|
|
|
def test_failed_group_refresh_does_not_loop(self) -> None:
|
|
self.existing_engine(access=False)
|
|
self.state["member"] = True
|
|
result = self.run_bootstrap(
|
|
"refresh_docker_group", extra_env={"PLANET_DOCKER_GROUP_REEXEC": "1"}
|
|
)
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertEqual(self.calls("sudo"), [])
|
|
|
|
def test_stopped_existing_engine_is_started_without_installation(self) -> None:
|
|
self.existing_engine(running=False)
|
|
result = self.run_bootstrap()
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
self.assertEqual(self.calls("apt-get"), [])
|
|
self.assertTrue(self.state["running"])
|
|
|
|
def test_missing_or_outdated_plugins_are_installed_without_replacing_engine(self) -> None:
|
|
self.existing_engine()
|
|
self.state.update(compose=False, buildx="0.16.0")
|
|
result = self.run_bootstrap()
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
self.assertEqual(
|
|
self.calls("apt-get")[-1], ["install", "-y", "docker-compose-v2", "docker-buildx"]
|
|
)
|
|
|
|
def test_existing_docker_ce_keeps_its_package_family(self) -> None:
|
|
self.existing_engine()
|
|
self.state.update(compose=False, buildx="", ce=True)
|
|
result = self.run_bootstrap()
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
self.assertEqual(
|
|
self.calls("apt-get")[-1],
|
|
["install", "-y", "docker-compose-plugin", "docker-buildx-plugin"],
|
|
)
|
|
|
|
def test_install_failure_stops_before_daemon_start(self) -> None:
|
|
self.state["apt_fail"] = True
|
|
result = self.run_bootstrap()
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertEqual(self.calls("systemctl"), [])
|
|
self.assertIn("simulated apt failure", result.stdout)
|
|
|
|
def test_missing_sudo_reports_requirement_without_attempting_install(self) -> None:
|
|
(self.bin / "sudo").unlink()
|
|
result = self.run_bootstrap()
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("缺少 sudo", result.stdout)
|
|
self.assertEqual(self.calls("apt-get"), [])
|
|
|
|
def test_desktop_and_nonlocal_contexts_are_not_replaced(self) -> None:
|
|
result = self.run_bootstrap(extra_env={"TEST_DESKTOP": "1"})
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("Docker Desktop", result.stdout)
|
|
self.assertEqual(self.calls("apt-get"), [])
|
|
result = self.run_bootstrap(extra_env={"DOCKER_HOST": "ssh://remote-docker"})
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertEqual(self.calls("apt-get"), [])
|
|
|
|
def test_working_remote_environment_is_reused(self) -> None:
|
|
self.existing_engine()
|
|
result = self.run_bootstrap(extra_env={"DOCKER_HOST": "ssh://remote-docker"})
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
self.assertEqual(self.calls("sudo"), [])
|
|
|
|
def test_unsupported_os_has_an_explicit_error(self) -> None:
|
|
result = self.run_bootstrap(extra_env={"TEST_OS": "debian"})
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("自动安装目前支持 Ubuntu", result.stdout)
|
|
self.assertEqual(self.calls("apt-get"), [])
|
|
|
|
def test_missing_cli_diagnostic_never_suggests_a_socket_unit(self) -> None:
|
|
result = self.run_bootstrap("log_docker_daemon_diagnostics")
|
|
self.assertEqual(result.returncode, 0, result.stdout + result.stderr)
|
|
self.assertIn("未找到 Docker CLI", result.stdout)
|
|
self.assertNotIn("systemctl", result.stdout)
|
|
self.assertEqual(self.calls("systemctl"), [])
|
|
|
|
def test_missing_service_diagnostic_never_suggests_a_socket_unit(self) -> None:
|
|
(self.bin / "docker").symlink_to("mock-command")
|
|
result = self.run_bootstrap("log_docker_daemon_diagnostics")
|
|
self.assertIn("未检测到可用的 docker.service", result.stdout)
|
|
self.assertNotIn("enable --now docker.socket", result.stdout)
|
|
|
|
def test_service_without_socket_unit_gets_service_only_hint(self) -> None:
|
|
self.existing_engine(running=False)
|
|
self.state["socket_unit"] = False
|
|
result = self.run_bootstrap("log_docker_daemon_diagnostics")
|
|
self.assertIn("enable --now docker.service", result.stdout)
|
|
self.assertNotIn("enable --now docker.socket", result.stdout)
|
|
|
|
def test_daemon_start_failure_is_not_reported_as_database_failure(self) -> None:
|
|
self.existing_engine(running=False)
|
|
self.state["service_fail"] = True
|
|
result = self.run_bootstrap()
|
|
self.assertNotEqual(result.returncode, 0)
|
|
self.assertIn("Docker Engine 启动失败", result.stdout)
|
|
self.assertNotIn("数据库启动失败", result.stdout)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main(verbosity=2)
|