Public Access
emil
4d073f3beb
Post-merge / detect-type (push) Successful in 7s
Post-merge / validate-commit-msg (push) Successful in 10s
Post-merge / configure-repo (push) Successful in 19s
Post-merge / release (push) Successful in 42s
Post-merge / vikunja (push) Successful in 8s
Post-merge / sync-wiki (push) Successful in 42s
Post-merge / badges (push) Successful in 41s
279 lines
13 KiB
Python
279 lines
13 KiB
Python
"""Unit tests for devx.molecule.start_docker."""
|
|
|
|
import os
|
|
from unittest.mock import MagicMock, mock_open, patch
|
|
|
|
from click.testing import CliRunner
|
|
|
|
from devx.molecule.start_docker import (
|
|
DOCKER_SOCK,
|
|
_diagnose_socket,
|
|
is_docker_ready,
|
|
main,
|
|
start_docker_daemon,
|
|
)
|
|
|
|
|
|
class TestIsDockerReady:
|
|
@patch("devx.molecule.start_docker.subprocess.run")
|
|
def test_ready(self, mock_run: MagicMock) -> None:
|
|
mock_run.return_value = MagicMock(returncode=0)
|
|
with patch.dict("os.environ", {"DOCKER_HOST": f"unix://{DOCKER_SOCK}"}, clear=False):
|
|
assert is_docker_ready() is True
|
|
call_kwargs = mock_run.call_args
|
|
assert call_kwargs.args[0] == ["docker", "info"]
|
|
assert call_kwargs.kwargs["env"]["DOCKER_HOST"] == f"unix://{DOCKER_SOCK}"
|
|
|
|
@patch("devx.molecule.start_docker.subprocess.run")
|
|
def test_not_ready(self, mock_run: MagicMock) -> None:
|
|
mock_run.return_value = MagicMock(returncode=1)
|
|
assert is_docker_ready() is False
|
|
|
|
@patch("devx.molecule.start_docker.subprocess.run")
|
|
def test_uses_docker_host_env(self, mock_run: MagicMock) -> None:
|
|
"""Should check the socket specified by DOCKER_HOST env var."""
|
|
mock_run.return_value = MagicMock(returncode=0)
|
|
rootless = "unix:///run/user/999/docker.sock"
|
|
with patch.dict("os.environ", {"DOCKER_HOST": rootless}, clear=False):
|
|
assert is_docker_ready() is True
|
|
call_kwargs = mock_run.call_args
|
|
assert call_kwargs.kwargs["env"]["DOCKER_HOST"] == rootless
|
|
|
|
|
|
class TestDiagnoseSocket:
|
|
@patch("devx.molecule.start_docker.os.stat")
|
|
@patch("devx.molecule.start_docker.os.path.exists", return_value=True)
|
|
@patch("devx.molecule.start_docker.subprocess.run")
|
|
def test_socket_exists(self, mock_run: MagicMock, mock_exists: MagicMock, mock_stat: MagicMock) -> None:
|
|
mock_stat.return_value = MagicMock(st_mode=0o660, st_uid=0, st_gid=0)
|
|
mock_run.side_effect = [
|
|
MagicMock(stdout="/dev/sda1 /var/lib/docker ext4\n", returncode=0, text=""),
|
|
MagicMock(stdout="default\n", returncode=0, text=""),
|
|
MagicMock(
|
|
stdout="Server Version: 29.5.2\nStorage Driver: overlay2\nDocker Root Dir: /var/lib/docker\n",
|
|
returncode=0,
|
|
text="",
|
|
),
|
|
]
|
|
_diagnose_socket()
|
|
mock_exists.assert_called_with(DOCKER_SOCK)
|
|
|
|
@patch("devx.molecule.start_docker.os.path.exists", return_value=False)
|
|
@patch("devx.molecule.start_docker.subprocess.run")
|
|
def test_socket_missing(self, mock_run: MagicMock, mock_exists: MagicMock) -> None:
|
|
mock_run.side_effect = [
|
|
MagicMock(stdout="proc on /proc type proc\n", returncode=0, text=""),
|
|
MagicMock(stdout="default\n", returncode=0, text=""),
|
|
MagicMock(stdout="", stderr="Cannot connect", returncode=1, text=""),
|
|
]
|
|
_diagnose_socket()
|
|
mock_exists.assert_called_with(DOCKER_SOCK)
|
|
|
|
|
|
class TestStartDockerDaemon:
|
|
@patch("devx.molecule.start_docker._diagnose_socket")
|
|
@patch("devx.molecule.start_docker.is_docker_ready", return_value=True)
|
|
def test_host_socket_available(self, mock_ready: MagicMock, mock_diag: MagicMock) -> None:
|
|
"""Should return immediately if host Docker is available."""
|
|
assert start_docker_daemon(timeout=5) is True
|
|
mock_ready.assert_called_once()
|
|
mock_diag.assert_called_once()
|
|
|
|
@patch("devx.molecule.start_docker._diagnose_socket")
|
|
@patch("devx.molecule.start_docker.os.path.exists", return_value=True)
|
|
@patch("devx.molecule.start_docker.is_docker_ready")
|
|
def test_rootless_socket_available(
|
|
self, mock_ready: MagicMock, mock_exists: MagicMock, mock_diag: MagicMock
|
|
) -> None:
|
|
"""Should use rootless socket if host socket fails."""
|
|
# First check (host) fails, second check (rootless) succeeds
|
|
mock_ready.side_effect = [False, True]
|
|
with patch("devx.molecule.start_docker.glob.glob", return_value=[]):
|
|
assert start_docker_daemon(timeout=5) is True
|
|
|
|
@patch("devx.molecule.start_docker._diagnose_socket")
|
|
@patch("devx.molecule.start_docker.os.path.exists", return_value=True)
|
|
@patch("devx.molecule.start_docker.is_docker_ready")
|
|
def test_alt_rootless_socket_found(
|
|
self, mock_ready: MagicMock, mock_exists: MagicMock, mock_diag: MagicMock
|
|
) -> None:
|
|
"""Should find rootless socket at a different UID via glob scan."""
|
|
# Host fails, own rootless fails, alt rootless succeeds
|
|
mock_ready.side_effect = [False, False, True]
|
|
alt_sock = "/run/user/999/docker.sock"
|
|
with patch("devx.molecule.start_docker.glob.glob", return_value=[alt_sock]):
|
|
assert start_docker_daemon(timeout=5) is True
|
|
|
|
@patch("devx.molecule.start_docker._diagnose_socket")
|
|
@patch("devx.molecule.start_docker.os.path.exists", return_value=True)
|
|
@patch("devx.molecule.start_docker.is_docker_ready")
|
|
def test_alt_rootless_socket_skips_own(
|
|
self, mock_ready: MagicMock, mock_exists: MagicMock, mock_diag: MagicMock
|
|
) -> None:
|
|
"""Should skip the own rootless socket in glob scan (already tried)."""
|
|
# Host fails, own rootless fails, alt rootless also fails, dockerd fails
|
|
mock_ready.side_effect = [False, False, False, False, False, False]
|
|
own_sock = f"/run/user/{os.getuid()}/docker.sock"
|
|
alt_sock = "/run/user/999/docker.sock"
|
|
with (
|
|
patch("devx.molecule.start_docker.glob.glob", return_value=[own_sock, alt_sock]),
|
|
patch("devx.molecule.start_docker.time.sleep"),
|
|
patch("devx.molecule.start_docker.subprocess.Popen"),
|
|
patch("devx.molecule.start_docker.tempfile.NamedTemporaryFile") as mock_ntf,
|
|
patch("builtins.open", mock_open(read_data="err")),
|
|
):
|
|
mock_ntf.return_value = MagicMock(name="/tmp/dockerd.log")
|
|
assert start_docker_daemon(timeout=2) is False
|
|
|
|
@patch("devx.molecule.start_docker.glob.glob", return_value=[])
|
|
@patch("devx.molecule.start_docker._diagnose_socket")
|
|
@patch("devx.molecule.start_docker.os.path.exists", return_value=False)
|
|
@patch("devx.molecule.start_docker.is_docker_ready", return_value=False)
|
|
@patch("devx.molecule.start_docker.time.sleep")
|
|
@patch("devx.molecule.start_docker.subprocess.Popen")
|
|
@patch("devx.molecule.start_docker.tempfile.NamedTemporaryFile")
|
|
def test_starts_local_daemon(
|
|
self,
|
|
mock_ntf: MagicMock,
|
|
mock_popen: MagicMock,
|
|
mock_sleep: MagicMock,
|
|
mock_ready: MagicMock,
|
|
mock_exists: MagicMock,
|
|
mock_diag: MagicMock,
|
|
mock_glob: MagicMock,
|
|
) -> None:
|
|
mock_ntf.return_value = MagicMock(name="/tmp/dockerd.log")
|
|
# Host fails, rootless doesn't exist, local daemon starts
|
|
mock_ready.side_effect = [False, False, False, False, True]
|
|
assert start_docker_daemon(timeout=5) is True
|
|
mock_popen.assert_called_once()
|
|
popen_args = mock_popen.call_args.args[0]
|
|
assert "dockerd" in popen_args
|
|
assert "--storage-driver" in popen_args
|
|
assert "vfs" in popen_args
|
|
assert "-H" in popen_args
|
|
|
|
@patch("devx.molecule.start_docker.glob.glob", return_value=[])
|
|
@patch("devx.molecule.start_docker._diagnose_socket")
|
|
@patch("devx.molecule.start_docker.os.path.exists", return_value=False)
|
|
@patch("devx.molecule.start_docker.is_docker_ready", return_value=False)
|
|
@patch("devx.molecule.start_docker.time.sleep")
|
|
@patch("devx.molecule.start_docker.subprocess.Popen")
|
|
@patch("devx.molecule.start_docker.tempfile.NamedTemporaryFile")
|
|
def test_fails_after_timeout(
|
|
self,
|
|
mock_ntf: MagicMock,
|
|
mock_popen: MagicMock,
|
|
mock_sleep: MagicMock,
|
|
mock_ready: MagicMock,
|
|
mock_exists: MagicMock,
|
|
mock_diag: MagicMock,
|
|
mock_glob: MagicMock,
|
|
) -> None:
|
|
mock_ntf.return_value = MagicMock(name="/tmp/dockerd.log")
|
|
with patch("builtins.open", mock_open(read_data="dockerd error log")):
|
|
assert start_docker_daemon(timeout=3) is False
|
|
mock_popen.assert_called_once()
|
|
assert mock_sleep.call_count == 3
|
|
|
|
@patch("devx.molecule.start_docker.glob.glob", return_value=[])
|
|
@patch("devx.molecule.start_docker._diagnose_socket")
|
|
@patch("devx.molecule.start_docker.os.path.exists", return_value=False)
|
|
@patch("devx.molecule.start_docker.is_docker_ready", return_value=False)
|
|
@patch("devx.molecule.start_docker.time.sleep")
|
|
@patch("devx.molecule.start_docker.subprocess.Popen")
|
|
@patch("devx.molecule.start_docker.tempfile.NamedTemporaryFile")
|
|
def test_fails_log_read_error(
|
|
self,
|
|
mock_ntf: MagicMock,
|
|
mock_popen: MagicMock,
|
|
mock_sleep: MagicMock,
|
|
mock_ready: MagicMock,
|
|
mock_exists: MagicMock,
|
|
mock_diag: MagicMock,
|
|
mock_glob: MagicMock,
|
|
) -> None:
|
|
"""Should handle log read errors gracefully."""
|
|
mock_ntf.return_value = MagicMock(name="/tmp/dockerd.log")
|
|
with patch("builtins.open", side_effect=OSError("permission denied")):
|
|
assert start_docker_daemon(timeout=2) is False
|
|
|
|
@patch("devx.molecule.start_docker.glob.glob", return_value=[])
|
|
@patch("devx.molecule.start_docker._diagnose_socket")
|
|
@patch("devx.molecule.start_docker.os.path.exists", return_value=False)
|
|
@patch("devx.molecule.start_docker.is_docker_ready")
|
|
@patch("devx.molecule.start_docker.time.sleep")
|
|
@patch("devx.molecule.start_docker.subprocess.Popen")
|
|
@patch("devx.molecule.start_docker.tempfile.NamedTemporaryFile")
|
|
def test_local_daemon_ready_on_first_check(
|
|
self,
|
|
mock_ntf: MagicMock,
|
|
mock_popen: MagicMock,
|
|
mock_sleep: MagicMock,
|
|
mock_ready: MagicMock,
|
|
mock_exists: MagicMock,
|
|
mock_diag: MagicMock,
|
|
mock_glob: MagicMock,
|
|
) -> None:
|
|
mock_ntf.return_value = MagicMock(name="/tmp/dockerd.log")
|
|
# Host fails, rootless doesn't exist, local ready on first loop check
|
|
mock_ready.side_effect = [False, False, True]
|
|
assert start_docker_daemon(timeout=5) is True
|
|
assert mock_popen.call_count == 1
|
|
assert mock_sleep.call_count == 1
|
|
|
|
@patch("devx.molecule.start_docker._diagnose_socket")
|
|
@patch("devx.molecule.start_docker.os.environ")
|
|
@patch("devx.molecule.start_docker.is_docker_ready", return_value=True)
|
|
def test_sets_docker_host(
|
|
self,
|
|
mock_ready: MagicMock,
|
|
mock_environ: MagicMock,
|
|
mock_diag: MagicMock,
|
|
) -> None:
|
|
"""DOCKER_HOST must be set so molecule connects to correct socket."""
|
|
start_docker_daemon(timeout=5)
|
|
mock_environ.__setitem__.assert_called_with("DOCKER_HOST", f"unix://{DOCKER_SOCK}")
|
|
|
|
|
|
class TestMain:
|
|
@patch("devx.molecule.start_docker.start_docker_daemon", return_value=True)
|
|
def test_success(self, mock_start: MagicMock) -> None:
|
|
runner = CliRunner()
|
|
result = runner.invoke(main, [])
|
|
assert result.exit_code == 0
|
|
|
|
@patch("devx.molecule.start_docker.start_docker_daemon", return_value=False)
|
|
def test_failure(self, mock_start: MagicMock) -> None:
|
|
runner = CliRunner()
|
|
result = runner.invoke(main, [])
|
|
assert result.exit_code == 1
|
|
|
|
@patch("devx.molecule.start_docker.start_docker_daemon", return_value=True)
|
|
def test_custom_timeout_flag(self, mock_start: MagicMock) -> None:
|
|
runner = CliRunner()
|
|
result = runner.invoke(main, ["--timeout", "60"])
|
|
assert result.exit_code == 0
|
|
mock_start.assert_called_once_with(60)
|
|
|
|
@patch("devx.molecule.start_docker.os.environ.get")
|
|
@patch("devx.molecule.start_docker.start_docker_daemon", return_value=True)
|
|
def test_exports_github_env(self, mock_start: MagicMock, mock_get: MagicMock) -> None:
|
|
"""Should write DOCKER_HOST to GITHUB_ENV when available."""
|
|
mock_get.side_effect = lambda key, default="": (
|
|
"/tmp/github_env" if key == "GITHUB_ENV" else f"unix://{DOCKER_SOCK}" if key == "DOCKER_HOST" else default
|
|
)
|
|
with patch("builtins.open", mock_open()) as mock_file:
|
|
runner = CliRunner()
|
|
result = runner.invoke(main, [])
|
|
assert result.exit_code == 0
|
|
mock_file.assert_called_with("/tmp/github_env", "a")
|
|
|
|
@patch("devx.molecule.start_docker.os.environ.get", return_value="")
|
|
@patch("devx.molecule.start_docker.start_docker_daemon", return_value=True)
|
|
def test_no_github_env(self, mock_start: MagicMock, mock_get: MagicMock) -> None:
|
|
"""Should not crash when GITHUB_ENV is not set."""
|
|
runner = CliRunner()
|
|
result = runner.invoke(main, [])
|
|
assert result.exit_code == 0
|