DEVX-110: feat: extract docker-login, tofu-ops, check-deps, install-tofu to Python tools

tests/unit/test_check_deps.py

@@ -0,0 +1,108 @@
+"""Unit tests for devx.tools.check_deps."""
+
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+from click.testing import CliRunner
+
+from devx.tools.check_deps import (
+    _check_python_version,
+    _check_tool,
+    cli,
+)
+
+
+class TestCheckTool:
+    @patch("devx.tools.check_deps.shutil.which", return_value="/usr/bin/tofu")
+    def test_found(self, mock_which: MagicMock) -> None:
+        assert _check_tool("tofu") is True
+
+    @patch("devx.tools.check_deps.shutil.which", return_value=None)
+    def test_not_found_required(self, mock_which: MagicMock) -> None:
+        assert _check_tool("tofu") is False
+
+    @patch("devx.tools.check_deps.shutil.which", return_value=None)
+    def test_not_found_optional(self, mock_which: MagicMock) -> None:
+        assert _check_tool("checkmake", optional=True) is False
+
+
+class TestCheckPythonVersion:
+    @patch("devx.tools.check_deps.subprocess.run")
+    def test_valid_version(self, mock_run: MagicMock, tmp_path: Path) -> None:
+        venv_bin = tmp_path / "bin"
+        venv_bin.mkdir()
+        (venv_bin / "python").touch()
+        mock_run.return_value = MagicMock(returncode=0, stdout="Python 3.12.3\n", stderr="")
+        _check_python_version(venv_bin)
+
+    @patch("devx.tools.check_deps.subprocess.run")
+    def test_old_version(self, mock_run: MagicMock, tmp_path: Path) -> None:
+        venv_bin = tmp_path / "bin"
+        venv_bin.mkdir()
+        (venv_bin / "python").touch()
+        mock_run.return_value = MagicMock(returncode=0, stdout="Python 3.11.0\n", stderr="")
+        _check_python_version(venv_bin)
+
+    def test_no_venv(self, tmp_path: Path) -> None:
+        venv_bin = tmp_path / "bin"
+        _check_python_version(venv_bin)
+
+    @patch("devx.tools.check_deps.subprocess.run")
+    def test_command_fails(self, mock_run: MagicMock, tmp_path: Path) -> None:
+        venv_bin = tmp_path / "bin"
+        venv_bin.mkdir()
+        (venv_bin / "python").touch()
+        mock_run.return_value = MagicMock(returncode=1, stdout="", stderr="error")
+        _check_python_version(venv_bin)
+
+    @patch("devx.tools.check_deps.subprocess.run")
+    def test_unparseable_version(self, mock_run: MagicMock, tmp_path: Path) -> None:
+        venv_bin = tmp_path / "bin"
+        venv_bin.mkdir()
+        (venv_bin / "python").touch()
+        mock_run.return_value = MagicMock(returncode=0, stdout="garbage\n", stderr="")
+        _check_python_version(venv_bin)
+
+
+class TestCli:
+    @patch("devx.tools.check_deps._check_python_version")
+    @patch("devx.tools.check_deps._check_tool")
+    def test_all_present(self, mock_check: MagicMock, mock_py: MagicMock) -> None:
+        mock_check.return_value = True
+        runner = CliRunner()
+        result = runner.invoke(cli, [])
+        assert result.exit_code == 0
+        assert "All core tools present" in result.output
+
+    @patch("devx.tools.check_deps._check_python_version")
+    @patch("devx.tools.check_deps._check_tool")
+    def test_missing_required(self, mock_check: MagicMock, mock_py: MagicMock) -> None:
+        mock_check.side_effect = lambda name, optional=False: name != "tofu"
+        runner = CliRunner()
+        result = runner.invoke(cli, [])
+        assert result.exit_code != 0
+
+    @patch("devx.tools.check_deps._check_python_version")
+    @patch("devx.tools.check_deps._check_tool")
+    def test_missing_optional_with_fallback(self, mock_check: MagicMock, mock_py: MagicMock, tmp_path: Path) -> None:
+        checkmake_bin = tmp_path / "checkmake"
+        checkmake_bin.touch()
+
+        def _side(name: str, optional: bool = False) -> bool:
+            return name != "checkmake"
+
+        mock_check.side_effect = _side
+        runner = CliRunner()
+        result = runner.invoke(cli, ["--checkmake-bin", str(checkmake_bin)])
+        assert result.exit_code == 0
+
+    @patch("devx.tools.check_deps._check_python_version")
+    @patch("devx.tools.check_deps._check_tool")
+    def test_missing_optional_no_fallback(self, mock_check: MagicMock, mock_py: MagicMock) -> None:
+        def _side(name: str, optional: bool = False) -> bool:
+            return name != "checkmake"
+
+        mock_check.side_effect = _side
+        runner = CliRunner()
+        result = runner.invoke(cli, [])
+        assert result.exit_code == 0

tests/unit/test_docker_login.py

@@ -0,0 +1,149 @@
+"""Unit tests for devx.tools.docker_login."""
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+from click.testing import CliRunner
+
+from devx.tools.docker_login import (
+    _resolve_credentials,
+    cli,
+    docker_login,
+)
+
+
+class TestDockerLogin:
+    @patch("devx.tools.docker_login.subprocess.run")
+    def test_success(self, mock_run: MagicMock) -> None:
+        mock_run.return_value = MagicMock(returncode=0, stdout="", stderr="")
+        assert docker_login("registry.io", "user", "tok") is True
+
+    @patch("devx.tools.docker_login.subprocess.run")
+    def test_failure_raises(self, mock_run: MagicMock) -> None:
+        mock_run.return_value = MagicMock(returncode=1, stdout="", stderr="auth failed")
+        with pytest.raises(Exception, match="auth failed"):
+            docker_login("registry.io", "user", "tok")
+
+    @patch("devx.tools.docker_login.subprocess.run")
+    def test_failure_suppressed(self, mock_run: MagicMock) -> None:
+        mock_run.return_value = MagicMock(returncode=1, stdout="", stderr="auth failed")
+        assert docker_login("registry.io", "user", "tok", suppress_failure=True) is False
+
+
+class TestResolveCredentials:
+    @patch.dict("os.environ", {"CI_GITEA_TOKEN": "tok", "CI_GITEA_USERNAME": "emil"}, clear=True)
+    def test_both_set(self) -> None:
+        user, token = _resolve_credentials("CI_GITEA_TOKEN", "CI_GITEA_USERNAME", None)
+        assert user == "emil"
+        assert token == "tok"
+
+    @patch.dict("os.environ", {"CI_GITEA_TOKEN": "tok"}, clear=True)
+    def test_token_only_with_default(self) -> None:
+        user, token = _resolve_credentials("CI_GITEA_TOKEN", "CI_GITEA_USERNAME", "emil")
+        assert user == "emil"
+        assert token == "tok"
+
+    @patch.dict("os.environ", {}, clear=True)
+    def test_no_token(self) -> None:
+        user, token = _resolve_credentials("CI_GITEA_TOKEN", "CI_GITEA_USERNAME", "emil")
+        assert user is None
+        assert token is None
+
+    @patch.dict("os.environ", {"CI_GITEA_TOKEN": "tok"}, clear=True)
+    def test_no_username_no_default(self) -> None:
+        user, token = _resolve_credentials("CI_GITEA_TOKEN", "CI_GITEA_USERNAME", None)
+        assert user == ""
+        assert token == "tok"
+
+
+class TestCli:
+    @patch("devx.tools.docker_login.docker_login")
+    @patch.dict("os.environ", {"CI_GITEA_TOKEN": "tok", "CI_GITEA_USERNAME": "emil"}, clear=True)
+    def test_required_login(self, mock_login: MagicMock) -> None:
+        mock_login.return_value = True
+        runner = CliRunner()
+        result = runner.invoke(
+            cli,
+            ["--registry", "reg.io", "--token-env", "CI_GITEA_TOKEN", "--username-env", "CI_GITEA_USERNAME"],
+        )
+        assert result.exit_code == 0
+        mock_login.assert_called_once()
+
+    @patch("devx.tools.docker_login.docker_login")
+    @patch.dict("os.environ", {"CI_GITEA_TOKEN": "tok"}, clear=True)
+    def test_default_username(self, mock_login: MagicMock) -> None:
+        mock_login.return_value = True
+        runner = CliRunner()
+        result = runner.invoke(
+            cli,
+            [
+                "--registry",
+                "reg.io",
+                "--token-env",
+                "CI_GITEA_TOKEN",
+                "--username-env",
+                "CI_GITEA_USERNAME",
+                "--default-username",
+                "emil",
+            ],
+        )
+        assert result.exit_code == 0
+        mock_login.assert_called_once_with("reg.io", "emil", "tok", suppress_failure=False)
+
+    @patch.dict("os.environ", {}, clear=True)
+    def test_required_no_token_raises(self) -> None:
+        runner = CliRunner()
+        result = runner.invoke(
+            cli,
+            ["--registry", "reg.io", "--token-env", "CI_GITEA_TOKEN", "--username-env", "CI_GITEA_USERNAME"],
+        )
+        assert result.exit_code != 0
+
+    @patch.dict("os.environ", {}, clear=True)
+    def test_optional_no_token_skips(self) -> None:
+        runner = CliRunner()
+        result = runner.invoke(
+            cli,
+            [
+                "--registry",
+                "reg.io",
+                "--token-env",
+                "CI_GITEA_TOKEN",
+                "--username-env",
+                "CI_GITEA_USERNAME",
+                "--optional",
+            ],
+        )
+        assert result.exit_code == 0
+        assert "Skipping" in result.output
+
+    @patch("devx.tools.docker_login.docker_login")
+    @patch.dict("os.environ", {"CI_GITEA_TOKEN": "tok"}, clear=True)
+    def test_no_username_raises(self, mock_login: MagicMock) -> None:
+        runner = CliRunner()
+        result = runner.invoke(
+            cli,
+            ["--registry", "reg.io", "--token-env", "CI_GITEA_TOKEN", "--username-env", "CI_GITEA_USERNAME"],
+        )
+        assert result.exit_code != 0
+        mock_login.assert_not_called()
+
+    @patch("devx.tools.docker_login.docker_login")
+    @patch.dict("os.environ", {"CI_GITEA_TOKEN": "tok", "CI_GITEA_USERNAME": "emil"}, clear=True)
+    def test_suppress_failure(self, mock_login: MagicMock) -> None:
+        mock_login.return_value = False
+        runner = CliRunner()
+        result = runner.invoke(
+            cli,
+            [
+                "--registry",
+                "reg.io",
+                "--token-env",
+                "CI_GITEA_TOKEN",
+                "--username-env",
+                "CI_GITEA_USERNAME",
+                "--suppress-failure",
+            ],
+        )
+        assert result.exit_code == 0
+        mock_login.assert_called_once_with("reg.io", "emil", "tok", suppress_failure=True)

tests/unit/test_install_tools.py

@@ -240,6 +240,35 @@ class TestInstallHadolint:
         assert (tmp_path / "hadolint").exists()
 
 
+class TestInstallTofu:
+    def test_already_installed(self) -> None:
+        with patch.object(install_tools, "_is_installed", return_value=True):
+            assert install_tools.install_tofu() is True
+
+    def test_install(self, tmp_path: Path) -> None:
+        import io
+        import tarfile
+
+        tarball_path = tmp_path / "archive.tar.gz"
+        binary_content = b"fake tofu"
+        with tarfile.open(tarball_path, "w:gz") as tar:
+            info = tarfile.TarInfo(name="tofu")
+            info.size = len(binary_content)
+            tar.addfile(info, io.BytesIO(binary_content))
+
+        with patch.object(install_tools, "_is_installed", return_value=False):
+            with patch.object(install_tools, "TARGET_DIR", tmp_path):
+                with patch.object(platform, "machine", return_value="x86_64"):
+                    with patch.object(platform, "system", return_value="Linux"):
+                        with patch.object(
+                            install_tools,
+                            "_download",
+                            side_effect=lambda url, dest: Path(dest).write_bytes(tarball_path.read_bytes()),
+                        ):
+                            assert install_tools.install_tofu() is True
+        assert (tmp_path / "tofu").exists()
+
+
 class TestListTools:
     def test_list(self, tmp_path: Path) -> None:
         with patch.object(install_tools, "TARGET_DIR", tmp_path):
@@ -274,6 +303,11 @@ class TestInstallTool:
             assert install_tools._install_tool("hadolint") is True
         mock.assert_called_once()
 
+    def test_tofu(self) -> None:
+        with patch.object(install_tools, "install_tofu", return_value=True) as mock:
+            assert install_tools._install_tool("tofu") is True
+        mock.assert_called_once()
+
     def test_unknown_tool(self) -> None:
         with pytest.raises(ClickException, match="Unknown tool"):
             install_tools._install_tool("unknown")
@@ -292,7 +326,7 @@ class TestMain:
         with patch.object(install_tools, "_install_tool", return_value=True) as mock_install:
             result = runner.invoke(install_tools.main, [])
         assert result.exit_code == 0
-        assert mock_install.call_count == 5
+        assert mock_install.call_count == 6
 
     def test_install_specific_tool(self) -> None:
         runner = CliRunner()

tests/unit/test_tofu_ops.py

@@ -0,0 +1,115 @@
+"""Unit tests for devx.tools.tofu_ops."""
+
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+from click.testing import CliRunner
+
+from devx.tools.tofu_ops import (
+    _run_tofu,
+    cli,
+    tofu_init,
+    tofu_validate,
+)
+
+
+class TestRunTofu:
+    @patch("devx.tools.tofu_ops.subprocess.run")
+    def test_success(self, mock_run: MagicMock, tmp_path: Path) -> None:
+        mock_run.return_value = MagicMock(returncode=0, stdout="", stderr="")
+        _run_tofu(["tofu", "init"], tmp_path)
+        mock_run.assert_called_once()
+
+    @patch("devx.tools.tofu_ops.subprocess.run")
+    def test_failure_raises(self, mock_run: MagicMock, tmp_path: Path) -> None:
+        mock_run.return_value = MagicMock(returncode=1, stdout="", stderr="error")
+        with pytest.raises(Exception, match="error"):
+            _run_tofu(["tofu", "validate"], tmp_path)
+
+
+class TestTofuInit:
+    @patch("devx.tools.tofu_ops._run_tofu")
+    def test_init_existing_dirs(self, mock_run: MagicMock, tmp_path: Path) -> None:
+        (tmp_path / "tofu/environments/staging").mkdir(parents=True)
+        (tmp_path / "tofu/environments/dns").mkdir(parents=True)
+        tofu_init("staging", root=str(tmp_path))
+        assert mock_run.call_count == 2
+
+    @patch("devx.tools.tofu_ops._run_tofu")
+    def test_init_skips_missing_dirs(self, mock_run: MagicMock, tmp_path: Path) -> None:
+        (tmp_path / "tofu/environments/staging").mkdir(parents=True)
+        # dns dir doesn't exist
+        tofu_init("staging", root=str(tmp_path))
+        assert mock_run.call_count == 1
+
+    @patch("devx.tools.tofu_ops._run_tofu")
+    def test_init_no_dirs_exist(self, mock_run: MagicMock, tmp_path: Path) -> None:
+        tofu_init("staging", root=str(tmp_path))
+        mock_run.assert_not_called()
+
+    @patch("devx.tools.tofu_ops._run_tofu")
+    def test_init_custom_dirs(self, mock_run: MagicMock, tmp_path: Path) -> None:
+        (tmp_path / "custom/dir").mkdir(parents=True)
+        tofu_init("staging", root=str(tmp_path), extra_dirs=["custom/dir"])
+        assert mock_run.call_count == 1
+
+
+class TestTofuValidate:
+    @patch("devx.tools.tofu_ops._run_tofu")
+    def test_validate_all_dirs(self, mock_run: MagicMock, tmp_path: Path) -> None:
+        for d in [
+            "tofu/modules/hetzner-vm",
+            "tofu/modules/hetzner-network",
+            "tofu/environments/staging",
+            "tofu/environments/production",
+            "tofu/environments/dns",
+        ]:
+            (tmp_path / d).mkdir(parents=True)
+        tofu_validate(root=str(tmp_path))
+        assert mock_run.call_count == 5
+
+    @patch("devx.tools.tofu_ops._run_tofu")
+    def test_validate_skips_missing(self, mock_run: MagicMock, tmp_path: Path) -> None:
+        (tmp_path / "tofu/environments/staging").mkdir(parents=True)
+        tofu_validate(root=str(tmp_path))
+        assert mock_run.call_count == 1
+
+    @patch("devx.tools.tofu_ops._run_tofu")
+    def test_validate_ci_mode(self, mock_run: MagicMock, tmp_path: Path) -> None:
+        (tmp_path / "tofu/environments/staging").mkdir(parents=True)
+        tofu_validate(root=str(tmp_path), ci=True)
+        # CI mode runs init + validate = 2 calls per dir
+        assert mock_run.call_count == 2
+        first_call = mock_run.call_args_list[0][0][0]
+        assert "init" in first_call
+        assert "-backend=false" in first_call
+
+    @patch("devx.tools.tofu_ops._run_tofu")
+    def test_validate_custom_dirs(self, mock_run: MagicMock, tmp_path: Path) -> None:
+        (tmp_path / "custom").mkdir()
+        tofu_validate(root=str(tmp_path), dirs=["custom"])
+        assert mock_run.call_count == 1
+
+
+class TestCli:
+    @patch("devx.tools.tofu_ops.tofu_init")
+    def test_init_command(self, mock_init: MagicMock) -> None:
+        runner = CliRunner()
+        result = runner.invoke(cli, ["init", "--env", "staging"])
+        assert result.exit_code == 0
+        mock_init.assert_called_once_with("staging", ".")
+
+    @patch("devx.tools.tofu_ops.tofu_validate")
+    def test_validate_command(self, mock_validate: MagicMock) -> None:
+        runner = CliRunner()
+        result = runner.invoke(cli, ["validate"])
+        assert result.exit_code == 0
+        mock_validate.assert_called_once_with(".", ci=False)
+
+    @patch("devx.tools.tofu_ops.tofu_validate")
+    def test_validate_ci_command(self, mock_validate: MagicMock) -> None:
+        runner = CliRunner()
+        result = runner.invoke(cli, ["validate", "--ci"])
+        assert result.exit_code == 0
+        mock_validate.assert_called_once_with(".", ci=True)