Files
devx/tests/unit/test_install_tools.py
T
emilandDevin <158243242+devin-ai-integration[bot]@users.noreply.github.com> 40896487ff fix: add tenacity retry to install_tools._download for transient network failures
The build-images CI job failed because GitHub releases dropped
connections ("Remote end closed connection without response") for
actionlint, vale, and hadolint simultaneously. The previous code only
fell through to the next fallback URL — there was no retry on the same
URL for transient connection resets.

Now uses tenacity.Retrying with exponential backoff (2-10s, 3 attempts)
on URLError/OSError/ConnectionError, matching the pattern already used
in devx.utils.network. The _sleep kwarg allows tests to skip real
sleeping.

3 new tests cover: retry-then-success, exhaustion-after-max-retries,
and no-retry-on-non-transient-errors.

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
2026-08-12 19:49:20 +02:00

550 lines
23 KiB
Python

from __future__ import annotations
import platform
import urllib.error
from pathlib import Path
from unittest.mock import patch
import pytest
from click import ClickException
from click.testing import CliRunner
import devx.tools.install_tools as install_tools
class TestArch:
def test_amd64(self) -> None:
with patch.object(platform, "machine", return_value="x86_64"):
assert install_tools._arch() == "amd64"
def test_arm64(self) -> None:
with patch.object(platform, "machine", return_value="aarch64"):
assert install_tools._arch() == "arm64"
def test_unsupported(self) -> None:
with patch.object(platform, "machine", return_value="riscv64"):
with pytest.raises(ClickException):
install_tools._arch()
class TestIsInstalled:
def test_on_path(self) -> None:
with patch("shutil.which", return_value="/usr/bin/actionlint"):
assert install_tools._is_installed("actionlint") is True
def test_in_target_dir(self, tmp_path: Path) -> None:
with patch.object(install_tools, "TARGET_DIR", tmp_path):
(tmp_path / "actionlint").touch()
with patch("shutil.which", return_value=None):
assert install_tools._is_installed("actionlint") is True
def test_not_installed(self, tmp_path: Path) -> None:
with patch.object(install_tools, "TARGET_DIR", tmp_path):
with patch("shutil.which", return_value=None):
assert install_tools._is_installed("actionlint") is False
class TestDownload:
def test_download(self, tmp_path: Path) -> None:
dest = tmp_path / "file.bin"
class _FakeResponse:
def __init__(self) -> None:
self._sent = False
def __enter__(self) -> _FakeResponse:
return self
def __exit__(self, *args: object) -> None:
pass
def read(self, n: int = -1) -> bytes:
if self._sent:
return b""
self._sent = True
return b"data"
with patch("urllib.request.urlopen", return_value=_FakeResponse()) as mock_urlopen:
install_tools._download("https://example.com/file", dest)
mock_urlopen.assert_called_once()
assert dest.read_bytes() == b"data"
def test_download_retries_on_transient_error(self, tmp_path: Path) -> None:
"""Download retries on URLError then succeeds."""
dest = tmp_path / "file.bin"
call_count = [0]
class _FakeResponse:
def __init__(self) -> None:
self._sent = False
def __enter__(self) -> _FakeResponse:
return self
def __exit__(self, *args: object) -> None:
pass
def read(self, n: int = -1) -> bytes:
if self._sent:
return b""
self._sent = True
return b"data"
def _flaky_urlopen(url: str, timeout: int = 60):
call_count[0] += 1
if call_count[0] < 2:
raise urllib.error.URLError("Remote end closed connection")
return _FakeResponse()
with patch("urllib.request.urlopen", side_effect=_flaky_urlopen):
install_tools._download("https://example.com/file", dest, _sleep=lambda _: None)
assert call_count[0] == 2
assert dest.read_bytes() == b"data"
def test_download_fails_after_max_retries(self, tmp_path: Path) -> None:
"""Download raises after MAX_DOWNLOAD_RETRIES attempts."""
dest = tmp_path / "file.bin"
call_count = [0]
def _always_fail(url: str, timeout: int = 60):
call_count[0] += 1
raise urllib.error.URLError("Remote end closed connection")
with patch("urllib.request.urlopen", side_effect=_always_fail):
with pytest.raises(urllib.error.URLError):
install_tools._download("https://example.com/file", dest, _sleep=lambda _: None)
assert call_count[0] == install_tools.MAX_DOWNLOAD_RETRIES
assert not dest.exists()
def test_download_no_retry_on_non_transient_error(self, tmp_path: Path) -> None:
"""Download does not retry on non-network errors (e.g. ValueError)."""
dest = tmp_path / "file.bin"
call_count = [0]
def _fail_with_value_error(url: str, timeout: int = 60):
call_count[0] += 1
raise ValueError("not a network error")
with patch("urllib.request.urlopen", side_effect=_fail_with_value_error):
with pytest.raises(ValueError):
install_tools._download("https://example.com/file", dest, _sleep=lambda _: None)
assert call_count[0] == 1
class TestDownloadBinary:
def test_download(self, tmp_path: Path) -> None:
dest = tmp_path / "act_runner"
def _write_file(url: str, path: Path) -> tuple[str, None]:
Path(path).write_bytes(b"binary")
return str(path), None
with patch.object(install_tools, "TARGET_DIR", tmp_path):
with patch.object(install_tools, "_download", side_effect=_write_file):
result = install_tools._download_binary("https://example.com/act_runner", "act_runner")
assert result == dest
assert dest.exists()
assert dest.stat().st_mode & 0o111
class TestDownloadAndExtractTarball:
def test_extract(self, tmp_path: Path) -> None:
import tarfile
# Create a fake tarball with a binary
tarball_path = tmp_path / "archive.tar.gz"
binary_content = b"fake binary"
with tarfile.open(tarball_path, "w:gz") as tar:
import io
info = tarfile.TarInfo(name="actionlint")
info.size = len(binary_content)
tar.addfile(info, io.BytesIO(binary_content))
target_dir = tmp_path / "bin"
target_dir.mkdir()
with patch.object(install_tools, "TARGET_DIR", target_dir):
with patch.object(
install_tools,
"_download",
side_effect=lambda url, dest: Path(dest).write_bytes(tarball_path.read_bytes()),
):
result = install_tools._download_and_extract_tarball(
"https://example.com/actionlint.tar.gz", "actionlint"
)
assert result == target_dir / "actionlint"
assert result.exists()
assert result.read_bytes() == binary_content
def test_binary_not_found(self, tmp_path: Path) -> None:
import tarfile
tarball_path = tmp_path / "archive.tar.gz"
with tarfile.open(tarball_path, "w:gz") as tar:
import io
info = tarfile.TarInfo(name="other_file")
info.size = 0
tar.addfile(info, io.BytesIO(b""))
target_dir = tmp_path / "bin"
target_dir.mkdir()
with patch.object(install_tools, "TARGET_DIR", target_dir):
with patch.object(
install_tools,
"_download",
side_effect=lambda url, dest: Path(dest).write_bytes(tarball_path.read_bytes()),
):
with pytest.raises(ClickException, match="not found in archive"):
install_tools._download_and_extract_tarball("https://example.com/actionlint.tar.gz", "actionlint")
class TestInstallActionlint:
def test_already_installed(self) -> None:
with patch.object(install_tools, "_is_installed", return_value=True):
assert install_tools.install_actionlint() is True
def test_install(self, tmp_path: Path) -> None:
import tarfile
tarball_path = tmp_path / "archive.tar.gz"
binary_content = b"fake actionlint"
with tarfile.open(tarball_path, "w:gz") as tar:
import io
info = tarfile.TarInfo(name="actionlint")
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(
install_tools,
"_download",
side_effect=lambda url, dest: Path(dest).write_bytes(tarball_path.read_bytes()),
):
assert install_tools.install_actionlint() is True
assert (tmp_path / "actionlint").exists()
class TestInstallGitCliff:
def test_already_installed(self) -> None:
with patch.object(install_tools, "_is_installed", return_value=True):
assert install_tools.install_git_cliff() is True
def test_install(self, tmp_path: Path) -> None:
import tarfile
tarball_path = tmp_path / "archive.tar.gz"
binary_content = b"fake git-cliff"
with tarfile.open(tarball_path, "w:gz") as tar:
import io
info = tarfile.TarInfo(name="git-cliff")
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(
install_tools,
"_download",
side_effect=lambda url, dest: Path(dest).write_bytes(tarball_path.read_bytes()),
):
assert install_tools.install_git_cliff() is True
assert (tmp_path / "git-cliff").exists()
class TestInstallActRunner:
def test_already_installed(self) -> None:
with patch.object(install_tools, "_is_installed", return_value=True):
assert install_tools.install_act_runner() is True
def test_install(self, tmp_path: Path) -> None:
def _write_file(url: str, path: Path) -> tuple[str, None]:
Path(path).write_bytes(b"binary")
return str(path), None
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(install_tools, "_download", side_effect=_write_file):
assert install_tools.install_act_runner() is True
assert (tmp_path / "act_runner").exists()
class TestInstallTea:
def test_already_installed(self) -> None:
with patch.object(install_tools, "_is_installed", return_value=True):
assert install_tools.install_tea() is True
def test_install(self, tmp_path: Path) -> None:
def _write_file(url: str, path: Path) -> tuple[str, None]:
Path(path).write_bytes(b"binary")
return str(path), None
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(install_tools, "_download", side_effect=_write_file):
assert install_tools.install_tea() is True
assert (tmp_path / "tea").exists()
def test_install_fallback_to_second_url(self, tmp_path: Path) -> None:
"""First URL fails (403), second URL succeeds."""
call_count = [0]
def _download_side_effect(url: str, dest: Path) -> None:
call_count[0] += 1
if call_count[0] == 1:
raise OSError("HTTP Error 403: Forbidden")
Path(dest).write_bytes(b"binary")
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(install_tools, "_download", side_effect=_download_side_effect):
assert install_tools.install_tea() is True
assert (tmp_path / "tea").exists()
assert call_count[0] == 2
def test_install_all_urls_fail(self, tmp_path: Path) -> None:
"""All URLs fail — should raise ClickException."""
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(install_tools, "_download", side_effect=OSError("403 Forbidden")):
with pytest.raises(ClickException, match="Failed to download tea"):
install_tools.install_tea()
class TestInstallHadolint:
def test_already_installed(self) -> None:
with patch.object(install_tools, "_is_installed", return_value=True):
assert install_tools.install_hadolint() is True
def test_install(self, tmp_path: Path) -> None:
def _write_file(url: str, path: Path) -> tuple[str, None]:
Path(path).write_bytes(b"binary")
return str(path), None
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(install_tools, "_download", side_effect=_write_file):
assert install_tools.install_hadolint() is True
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 TestInstallVale:
def test_already_installed(self) -> None:
with patch.object(install_tools, "_is_installed", return_value=True):
assert install_tools.install_vale() 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 vale"
with tarfile.open(tarball_path, "w:gz") as tar:
info = tarfile.TarInfo(name="vale")
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(
install_tools,
"_download",
side_effect=lambda url, dest: Path(dest).write_bytes(tarball_path.read_bytes()),
):
assert install_tools.install_vale() is True
assert (tmp_path / "vale").exists()
class TestInstallPromtool:
def test_already_installed(self) -> None:
with patch.object(install_tools, "_is_installed", return_value=True):
assert install_tools.install_promtool() 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 promtool"
with tarfile.open(tarball_path, "w:gz") as tar:
info = tarfile.TarInfo(name="promtool")
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(install_tools, "_arch", return_value="amd64"):
with patch.object(
install_tools,
"_download",
side_effect=lambda url, dest: Path(dest).write_bytes(tarball_path.read_bytes()),
):
assert install_tools.install_promtool() is True
assert (tmp_path / "promtool").exists()
def test_url_contains_version(self, tmp_path: Path) -> None:
"""Verify the download URL includes the correct promtool version."""
captured_url = []
def fake_extract(url: str, binary_name: str) -> Path:
captured_url.append(url)
return tmp_path / binary_name
with patch.object(install_tools, "_is_installed", return_value=False):
with patch.object(install_tools, "_download_and_extract_tarball", side_effect=fake_extract):
install_tools.install_promtool()
assert any(f"v{install_tools.PROMTOOL_VERSION}" in url for url in captured_url)
class TestListTools:
def test_list(self, tmp_path: Path) -> None:
with patch.object(install_tools, "TARGET_DIR", tmp_path):
with patch("shutil.which", return_value=None):
with patch.object(install_tools, "TOOL_NAMES", ["actionlint", "git-cliff", "act_runner"]):
install_tools.list_tools()
class TestInstallTool:
def test_actionlint(self) -> None:
with patch.object(install_tools, "install_actionlint", return_value=True) as mock:
assert install_tools._install_tool("actionlint") is True
mock.assert_called_once()
def test_git_cliff(self) -> None:
with patch.object(install_tools, "install_git_cliff", return_value=True) as mock:
assert install_tools._install_tool("git-cliff") is True
mock.assert_called_once()
def test_act_runner(self) -> None:
with patch.object(install_tools, "install_act_runner", return_value=True) as mock:
assert install_tools._install_tool("act_runner") is True
mock.assert_called_once()
def test_tea(self) -> None:
with patch.object(install_tools, "install_tea", return_value=True) as mock:
assert install_tools._install_tool("tea") is True
mock.assert_called_once()
def test_hadolint(self) -> None:
with patch.object(install_tools, "install_hadolint", return_value=True) as mock:
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_vale(self) -> None:
with patch.object(install_tools, "install_vale", return_value=True) as mock:
assert install_tools._install_tool("vale") is True
mock.assert_called_once()
def test_promtool(self) -> None:
with patch.object(install_tools, "install_promtool", return_value=True) as mock:
assert install_tools._install_tool("promtool") is True
mock.assert_called_once()
def test_unknown_tool(self) -> None:
with pytest.raises(ClickException, match="Unknown tool"):
install_tools._install_tool("unknown")
class TestMain:
def test_list_status(self) -> None:
runner = CliRunner()
with patch.object(install_tools, "_is_installed", return_value=True):
result = runner.invoke(install_tools.main, ["--list"])
assert result.exit_code == 0
assert "actionlint" in result.output
def test_install_all(self) -> None:
runner = CliRunner()
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 == 8
def test_install_specific_tool(self) -> None:
runner = CliRunner()
with patch.object(install_tools, "_install_tool", return_value=True) as mock_install:
result = runner.invoke(install_tools.main, ["--tool", "actionlint"])
assert result.exit_code == 0
mock_install.assert_called_once_with("actionlint")
def test_install_multiple_specific_tools(self) -> None:
runner = CliRunner()
with patch.object(install_tools, "_install_tool", return_value=True) as mock_install:
result = runner.invoke(install_tools.main, ["--tool", "git-cliff", "--tool", "tea"])
assert result.exit_code == 0
assert mock_install.call_count == 2
def test_install_failure(self) -> None:
runner = CliRunner()
with patch.object(install_tools, "_install_tool", side_effect=Exception("network error")):
result = runner.invoke(install_tools.main, ["--tool", "actionlint"])
assert result.exit_code != 0
def test_path_reminder(self, monkeypatch: pytest.MonkeyPatch) -> None:
"""When TARGET_DIR is not in PATH, a reminder is printed."""
monkeypatch.setenv("PATH", "/usr/bin:/bin")
runner = CliRunner()
with patch.object(install_tools, "_install_tool", return_value=True):
result = runner.invoke(install_tools.main, [])
assert result.exit_code == 0
assert "Add" in result.output
assert "PATH" in result.output
def test_no_path_reminder_when_in_path(self, monkeypatch: pytest.MonkeyPatch) -> None:
"""When TARGET_DIR is in PATH, no reminder is printed."""
target_dir = str(install_tools.TARGET_DIR)
monkeypatch.setenv("PATH", f"/usr/bin:{target_dir}:/bin")
runner = CliRunner()
with patch.object(install_tools, "_install_tool", return_value=True):
result = runner.invoke(install_tools.main, [])
assert result.exit_code == 0
assert "Add" not in result.output