GRM-51: refactor: convert shell scripts and inline workflow scripts to Python
This commit is contained in:
@@ -1,11 +1,13 @@
|
||||
"""Unit tests for scripts/ci/classify_changes.py."""
|
||||
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import click
|
||||
import pytest
|
||||
from click.testing import CliRunner
|
||||
|
||||
import scripts.ci.classify_changes as classify_changes_mod
|
||||
from scripts.ci.classify_changes import (
|
||||
classify_changes,
|
||||
get_changed_files,
|
||||
@@ -312,3 +314,60 @@ class TestMain:
|
||||
result = runner.invoke(main, ["--check", "user-facing"])
|
||||
assert result.exit_code == 0
|
||||
assert "User-facing changes detected" in result.output
|
||||
|
||||
|
||||
class TestGithubOutput:
|
||||
def test_writes_outputs(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
gh_file = tmp_path / "output.txt"
|
||||
monkeypatch.setenv("GITHUB_OUTPUT", str(gh_file))
|
||||
with patch.object(
|
||||
classify_changes_mod, "get_changed_files", return_value=["src/cli.py", "ansible/tasks/main.yml"]
|
||||
):
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(main, ["--base", "v1.0", "--head", "HEAD", "--github-output"])
|
||||
assert result.exit_code == 0
|
||||
content = gh_file.read_text()
|
||||
assert "ansible-changed=true" in content
|
||||
assert "user-facing-changed=true" in content
|
||||
|
||||
def test_no_changes(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
gh_file = tmp_path / "output.txt"
|
||||
monkeypatch.setenv("GITHUB_OUTPUT", str(gh_file))
|
||||
with patch.object(classify_changes_mod, "get_changed_files", return_value=[]):
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(main, ["--base", "v1.0", "--head", "HEAD", "--github-output"])
|
||||
assert result.exit_code == 0
|
||||
content = gh_file.read_text()
|
||||
assert "ansible-changed=false" in content
|
||||
assert "user-facing-changed=false" in content
|
||||
|
||||
def test_no_tags(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
gh_file = tmp_path / "output.txt"
|
||||
monkeypatch.setenv("GITHUB_OUTPUT", str(gh_file))
|
||||
with patch.object(classify_changes_mod, "get_latest_tag", return_value=""):
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(main, ["--github-output"])
|
||||
assert result.exit_code == 0
|
||||
content = gh_file.read_text()
|
||||
assert "ansible-changed=true" in content
|
||||
assert "user-facing-changed=true" in content
|
||||
|
||||
def test_no_env_var(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.delenv("GITHUB_OUTPUT", raising=False)
|
||||
with patch.object(classify_changes_mod, "get_changed_files", return_value=["src/cli.py"]):
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(main, ["--base", "v1.0", "--head", "HEAD", "--github-output"])
|
||||
assert result.exit_code != 0
|
||||
|
||||
def test_workflow_only(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
gh_file = tmp_path / "output.txt"
|
||||
monkeypatch.setenv("GITHUB_OUTPUT", str(gh_file))
|
||||
with patch.object(
|
||||
classify_changes_mod, "get_changed_files", return_value=[".gitea/workflows/ci.yml", "AGENTS.md"]
|
||||
):
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(main, ["--base", "v1.0", "--head", "HEAD", "--github-output"])
|
||||
assert result.exit_code == 0
|
||||
content = gh_file.read_text()
|
||||
assert "ansible-changed=false" in content
|
||||
assert "user-facing-changed=false" in content
|
||||
|
||||
@@ -0,0 +1,76 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
from click import ClickException
|
||||
from click.testing import CliRunner
|
||||
|
||||
import scripts.ci.detect_release_commit as detect_release_commit
|
||||
|
||||
|
||||
class TestGetCommitMessage:
|
||||
def test_success(self) -> None:
|
||||
mock_result = subprocess.CompletedProcess(args=[], returncode=0, stdout="feat: add feature\n", stderr="")
|
||||
with patch("subprocess.run", return_value=mock_result):
|
||||
assert detect_release_commit.get_commit_message() == "feat: add feature"
|
||||
|
||||
def test_failure(self) -> None:
|
||||
mock_result = subprocess.CompletedProcess(args=[], returncode=1, stdout="", stderr="git error")
|
||||
with patch("subprocess.run", return_value=mock_result):
|
||||
with pytest.raises(ClickException, match="git log failed"):
|
||||
detect_release_commit.get_commit_message()
|
||||
|
||||
|
||||
class TestIsReleaseCommit:
|
||||
def test_release_commit(self) -> None:
|
||||
assert detect_release_commit.is_release_commit("release: v1.0.0 [skip ci]") is True
|
||||
|
||||
def test_release_commit_no_skip(self) -> None:
|
||||
assert detect_release_commit.is_release_commit("release: v0.1.0") is True
|
||||
|
||||
def test_regular_commit(self) -> None:
|
||||
assert detect_release_commit.is_release_commit("feat: add feature") is False
|
||||
|
||||
def test_empty(self) -> None:
|
||||
assert detect_release_commit.is_release_commit("") is False
|
||||
|
||||
|
||||
class TestWriteGithubOutput:
|
||||
def test_write(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
gh_file = tmp_path / "output.txt"
|
||||
monkeypatch.setenv("GITHUB_OUTPUT", str(gh_file))
|
||||
detect_release_commit.write_github_output("is-release", "true")
|
||||
with open(gh_file) as f:
|
||||
assert f.read() == "is-release=true\n"
|
||||
|
||||
def test_no_env(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.delenv("GITHUB_OUTPUT", raising=False)
|
||||
with pytest.raises(ClickException, match="GITHUB_OUTPUT"):
|
||||
detect_release_commit.write_github_output("is-release", "true")
|
||||
|
||||
|
||||
class TestMain:
|
||||
def test_release_commit(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
gh_file = tmp_path / "output.txt"
|
||||
monkeypatch.setenv("GITHUB_OUTPUT", str(gh_file))
|
||||
with patch.object(detect_release_commit, "get_commit_message", return_value="release: v1.0.0 [skip ci]"):
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(detect_release_commit.main, [])
|
||||
assert result.exit_code == 0
|
||||
assert "Release commit" in result.output
|
||||
with open(gh_file) as f:
|
||||
assert "is-release=true" in f.read()
|
||||
|
||||
def test_regular_commit(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
gh_file = tmp_path / "output.txt"
|
||||
monkeypatch.setenv("GITHUB_OUTPUT", str(gh_file))
|
||||
with patch.object(detect_release_commit, "get_commit_message", return_value="feat: add feature"):
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(detect_release_commit.main, [])
|
||||
assert result.exit_code == 0
|
||||
assert "Regular merge commit" in result.output
|
||||
with open(gh_file) as f:
|
||||
assert "is-release=false" in f.read()
|
||||
@@ -1,8 +1,10 @@
|
||||
"""Unit tests for scripts/ci/discover_runners.py."""
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
from click.testing import CliRunner
|
||||
|
||||
from scripts.ci.discover_runners import (
|
||||
@@ -188,3 +190,21 @@ class TestMain:
|
||||
result = runner.invoke(main, ["--indices"])
|
||||
assert result.exit_code == 0
|
||||
assert json.loads(result.output.strip()) == ["1"]
|
||||
|
||||
@patch("scripts.ci.discover_runners.get_runner_count", return_value=3)
|
||||
def test_github_output(self, mock_count: MagicMock, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
gh_file = tmp_path / "output.txt"
|
||||
monkeypatch.setenv("GITHUB_OUTPUT", str(gh_file))
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(main, ["--github-output"])
|
||||
assert result.exit_code == 0
|
||||
content = gh_file.read_text()
|
||||
assert "runner-count=3" in content
|
||||
assert "runner-indices=" in content
|
||||
|
||||
@patch("scripts.ci.discover_runners.get_runner_count", return_value=3)
|
||||
def test_github_output_no_env(self, mock_count: MagicMock, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.delenv("GITHUB_OUTPUT", raising=False)
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(main, ["--github-output"])
|
||||
assert result.exit_code != 0
|
||||
|
||||
@@ -5,12 +5,14 @@ from unittest.mock import patch
|
||||
|
||||
import click
|
||||
import pytest
|
||||
from click.testing import CliRunner
|
||||
|
||||
from scripts.ci.distribute_molecule import (
|
||||
MOLECULE_ROOT,
|
||||
PLATFORMS,
|
||||
TestPair,
|
||||
build_pairs,
|
||||
cli,
|
||||
discover_scenarios,
|
||||
distribute,
|
||||
pairs_for_runner,
|
||||
@@ -192,6 +194,51 @@ class TestCli:
|
||||
assert "ubuntu-2204" in result.output
|
||||
|
||||
|
||||
class TestGithubEnv:
|
||||
def test_writes_env(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
gh_file = tmp_path / "env.txt"
|
||||
monkeypatch.setenv("GITHUB_ENV", str(gh_file))
|
||||
root = tmp_path / "molecule"
|
||||
scenario = root / "alpha"
|
||||
scenario.mkdir(parents=True)
|
||||
(scenario / "molecule.yml").write_text("name: alpha\n")
|
||||
with patch("scripts.ci.distribute_molecule.MOLECULE_ROOT", root):
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(cli, ["--runner-index", "1", "--max-runners", "3", "--github-env"])
|
||||
assert result.exit_code == 0
|
||||
content = gh_file.read_text()
|
||||
assert "TEST_PAIRS=" in content
|
||||
assert "SKIP=false" in content
|
||||
|
||||
def test_skip_if_excess(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
gh_file = tmp_path / "env.txt"
|
||||
monkeypatch.setenv("GITHUB_ENV", str(gh_file))
|
||||
root = tmp_path / "molecule"
|
||||
scenario = root / "alpha"
|
||||
scenario.mkdir(parents=True)
|
||||
(scenario / "molecule.yml").write_text("name: alpha\n")
|
||||
with patch("scripts.ci.distribute_molecule.MOLECULE_ROOT", root):
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(
|
||||
cli, ["--runner-index", "5", "--max-runners", "3", "--github-env", "--skip-if-excess"]
|
||||
)
|
||||
assert result.exit_code == 0
|
||||
content = gh_file.read_text()
|
||||
assert "TEST_PAIRS=\n" in content
|
||||
assert "SKIP=true" in content
|
||||
|
||||
def test_no_env_var(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.delenv("GITHUB_ENV", raising=False)
|
||||
root = tmp_path / "molecule"
|
||||
scenario = root / "alpha"
|
||||
scenario.mkdir(parents=True)
|
||||
(scenario / "molecule.yml").write_text("name: alpha\n")
|
||||
with patch("scripts.ci.distribute_molecule.MOLECULE_ROOT", root):
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(cli, ["--runner-index", "1", "--max-runners", "3", "--github-env"])
|
||||
assert result.exit_code != 0
|
||||
|
||||
|
||||
def test_main_module_block() -> None:
|
||||
import scripts.ci.distribute_molecule as dm
|
||||
|
||||
|
||||
@@ -77,7 +77,8 @@ class TestMain:
|
||||
# Write ci-cd-workflow.md with all scripts
|
||||
(docs / "tech" / "ci-cd-workflow.md").write_text(
|
||||
"auto_merge.py release.py publish.py review_pr.py "
|
||||
"notify_failure.py post_merge.py classify_changes.py discover_runners.py"
|
||||
"notify_failure.py post_merge.py classify_changes.py discover_runners.py "
|
||||
"detect_release_commit.py push_badges.py"
|
||||
)
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(main, ["--docs-dir", str(docs)])
|
||||
|
||||
@@ -0,0 +1,282 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import platform
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
from click import ClickException
|
||||
from click.testing import CliRunner
|
||||
|
||||
import scripts.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"
|
||||
|
||||
def _write_file(url: str, path: Path) -> tuple[str, None]:
|
||||
Path(path).write_bytes(b"data")
|
||||
return str(path), None
|
||||
|
||||
with patch("urllib.request.urlretrieve", side_effect=_write_file) as mock_retrieve:
|
||||
install_tools._download("https://example.com/file", dest)
|
||||
mock_retrieve.assert_called_once()
|
||||
assert dest.read_bytes() == b"data"
|
||||
|
||||
|
||||
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 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_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 == 3
|
||||
|
||||
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_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
|
||||
@@ -0,0 +1,130 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
from click.testing import CliRunner
|
||||
|
||||
import scripts.molecule_all as molecule_all
|
||||
|
||||
|
||||
class TestRunMolecule:
|
||||
def test_success(self) -> None:
|
||||
import subprocess
|
||||
|
||||
mock_result = subprocess.CompletedProcess(args=[], returncode=0, stdout="", stderr="")
|
||||
with patch("subprocess.run", return_value=mock_result) as mock_run:
|
||||
rc = molecule_all._run_molecule("/bin/molecule", "default", Path("/tmp/role"), {})
|
||||
assert rc == 0
|
||||
mock_run.assert_called_once()
|
||||
|
||||
def test_failure(self) -> None:
|
||||
import subprocess
|
||||
|
||||
mock_result = subprocess.CompletedProcess(args=[], returncode=1, stdout="", stderr="")
|
||||
with patch("subprocess.run", return_value=mock_result):
|
||||
rc = molecule_all._run_molecule("/bin/molecule", "default", Path("/tmp/role"), {})
|
||||
assert rc == 1
|
||||
|
||||
def test_non_default_scenario_adds_flag(self) -> None:
|
||||
import subprocess
|
||||
|
||||
mock_result = subprocess.CompletedProcess(args=[], returncode=0, stdout="", stderr="")
|
||||
with patch("subprocess.run", return_value=mock_result) as mock_run:
|
||||
molecule_all._run_molecule("/bin/molecule", "lifecycle", Path("/tmp/role"), {})
|
||||
cmd = mock_run.call_args[0][0]
|
||||
assert "-s" in cmd
|
||||
assert "lifecycle" in cmd
|
||||
|
||||
|
||||
class TestRunPlatform:
|
||||
def test_all_scenarios_pass(self) -> None:
|
||||
platform = {"name": "ubuntu-2204", "image": "ubuntu:22.04", "command": "/lib/systemd/systemd"}
|
||||
with patch("scripts.molecule_all._run_molecule", return_value=0) as mock_run:
|
||||
rc = molecule_all._run_platform("/bin/molecule", platform, Path("/tmp/role"), ["default", "lifecycle"], {})
|
||||
assert rc == 0
|
||||
assert mock_run.call_count == 2
|
||||
|
||||
def test_stops_on_failure(self) -> None:
|
||||
platform = {"name": "ubuntu-2204", "image": "ubuntu:22.04", "command": "/lib/systemd/systemd"}
|
||||
with patch("scripts.molecule_all._run_molecule", side_effect=[1, 0]) as mock_run:
|
||||
rc = molecule_all._run_platform("/bin/molecule", platform, Path("/tmp/role"), ["default", "lifecycle"], {})
|
||||
assert rc == 1
|
||||
assert mock_run.call_count == 1
|
||||
|
||||
def test_sets_env_vars(self) -> None:
|
||||
platform = {"name": "ubuntu-2204", "image": "ubuntu:22.04", "command": "/lib/systemd/systemd"}
|
||||
captured_env: dict[str, str] = {}
|
||||
|
||||
def _capture_env(cmd, cwd, env):
|
||||
captured_env.update(env)
|
||||
import subprocess
|
||||
|
||||
return subprocess.CompletedProcess(args=cmd, returncode=0, stdout="", stderr="")
|
||||
|
||||
with patch("subprocess.run", side_effect=_capture_env):
|
||||
molecule_all._run_platform("/bin/molecule", platform, Path("/tmp/role"), ["default"], {"PATH": "/usr/bin"})
|
||||
assert captured_env["MOLECULE_PLATFORM_NAME"] == "ubuntu-2204"
|
||||
assert captured_env["MOLECULE_PLATFORM_IMAGE"] == "ubuntu:22.04"
|
||||
assert captured_env["MOLECULE_PLATFORM_COMMAND"] == "/lib/systemd/systemd"
|
||||
|
||||
def test_empty_command_removes_env(self) -> None:
|
||||
platform = {"name": "custom", "image": "custom:latest", "command": ""}
|
||||
captured_env: dict[str, str] = {}
|
||||
|
||||
def _capture_env(cmd, cwd, env):
|
||||
captured_env.update(env)
|
||||
import subprocess
|
||||
|
||||
return subprocess.CompletedProcess(args=cmd, returncode=0, stdout="", stderr="")
|
||||
|
||||
with patch("subprocess.run", side_effect=_capture_env):
|
||||
molecule_all._run_platform(
|
||||
"/bin/molecule", platform, Path("/tmp/role"), ["default"], {"MOLECULE_PLATFORM_COMMAND": "old"}
|
||||
)
|
||||
assert "MOLECULE_PLATFORM_COMMAND" not in captured_env
|
||||
|
||||
|
||||
class TestMain:
|
||||
def test_molecule_not_found(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.chdir(tmp_path)
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(molecule_all.main, ["--bin", "nonexistent/bin"])
|
||||
assert result.exit_code != 0
|
||||
assert "molecule not found" in result.output
|
||||
|
||||
def test_role_dir_not_found(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.chdir(tmp_path)
|
||||
bin_dir = tmp_path / ".venv" / "bin"
|
||||
bin_dir.mkdir(parents=True)
|
||||
(bin_dir / "molecule").touch()
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(molecule_all.main, ["--bin", str(bin_dir)])
|
||||
assert result.exit_code != 0
|
||||
assert "Role directory not found" in result.output
|
||||
|
||||
def test_all_pass(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.chdir(tmp_path)
|
||||
bin_dir = tmp_path / ".venv" / "bin"
|
||||
bin_dir.mkdir(parents=True)
|
||||
(bin_dir / "molecule").touch()
|
||||
(tmp_path / "ansible" / "roles" / "gitea-runner").mkdir(parents=True)
|
||||
|
||||
runner = CliRunner()
|
||||
with patch("scripts.molecule_all._run_platform", return_value=0):
|
||||
result = runner.invoke(molecule_all.main, ["--bin", str(bin_dir)])
|
||||
assert result.exit_code == 0
|
||||
assert "All molecule scenarios passed" in result.output
|
||||
|
||||
def test_platform_failure(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.chdir(tmp_path)
|
||||
bin_dir = tmp_path / ".venv" / "bin"
|
||||
bin_dir.mkdir(parents=True)
|
||||
(bin_dir / "molecule").touch()
|
||||
(tmp_path / "ansible" / "roles" / "gitea-runner").mkdir(parents=True)
|
||||
|
||||
runner = CliRunner()
|
||||
with patch("scripts.molecule_all._run_platform", return_value=1):
|
||||
result = runner.invoke(molecule_all.main, ["--bin", str(bin_dir)])
|
||||
assert result.exit_code != 0
|
||||
@@ -1,6 +1,7 @@
|
||||
"""Unit tests for scripts/ci/post_merge.py."""
|
||||
|
||||
import http
|
||||
import subprocess
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import click
|
||||
@@ -9,6 +10,8 @@ from click.testing import CliRunner
|
||||
|
||||
from gitea_runner_manager.exceptions import APIError
|
||||
from scripts.ci.post_merge import (
|
||||
_get_git_commit_message,
|
||||
_get_git_commit_sha,
|
||||
build_comment,
|
||||
extract_conventional_msg,
|
||||
extract_task_id,
|
||||
@@ -206,3 +209,69 @@ class TestMain:
|
||||
assert result.exit_code == 0
|
||||
assert "Warning" in result.output
|
||||
assert "not updated" in result.output.lower()
|
||||
|
||||
|
||||
class TestGetGitCommitMessage:
|
||||
def test_success(self) -> None:
|
||||
mock_result = subprocess.CompletedProcess(args=[], returncode=0, stdout="GRM-20: fix: bug\n", stderr="")
|
||||
with patch("subprocess.run", return_value=mock_result):
|
||||
assert _get_git_commit_message() == "GRM-20: fix: bug"
|
||||
|
||||
def test_failure(self) -> None:
|
||||
mock_result = subprocess.CompletedProcess(args=[], returncode=1, stdout="", stderr="git error")
|
||||
with patch("subprocess.run", return_value=mock_result):
|
||||
with pytest.raises(click.ClickException, match="git log failed"):
|
||||
_get_git_commit_message()
|
||||
|
||||
|
||||
class TestGetGitCommitSha:
|
||||
def test_success(self) -> None:
|
||||
mock_result = subprocess.CompletedProcess(args=[], returncode=0, stdout="abc123\n", stderr="")
|
||||
with patch("subprocess.run", return_value=mock_result):
|
||||
assert _get_git_commit_sha() == "abc123"
|
||||
|
||||
def test_failure(self) -> None:
|
||||
mock_result = subprocess.CompletedProcess(args=[], returncode=1, stdout="", stderr="git error")
|
||||
with patch("subprocess.run", return_value=mock_result):
|
||||
with pytest.raises(click.ClickException, match="git rev-parse failed"):
|
||||
_get_git_commit_sha()
|
||||
|
||||
|
||||
class TestFromGit:
|
||||
@patch.dict("os.environ", {"VIKUNJA_TOKEN": "tok"})
|
||||
@patch("scripts.ci.post_merge.VikunjaClient")
|
||||
@patch("scripts.ci.post_merge._get_git_commit_sha", return_value="abc123")
|
||||
@patch("scripts.ci.post_merge._get_git_commit_message", return_value="GRM-20: fix: bug")
|
||||
def test_from_git(self, mock_msg: MagicMock, mock_sha: MagicMock, mock_client_cls: MagicMock) -> None:
|
||||
mock_client = MagicMock()
|
||||
mock_client.list_project_tasks.return_value = [
|
||||
{"id": 267, "identifier": "GRM-20"},
|
||||
]
|
||||
mock_client_cls.return_value = mock_client
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(main, ["--from-git"])
|
||||
assert result.exit_code == 0
|
||||
assert "updated and marked done" in result.output
|
||||
|
||||
@patch.dict("os.environ", {"VIKUNJA_TOKEN": "tok"})
|
||||
@patch("scripts.ci.post_merge.VikunjaClient")
|
||||
@patch("scripts.ci.post_merge._get_git_commit_sha", return_value="abc123")
|
||||
@patch("scripts.ci.post_merge._get_git_commit_message", return_value="GRM-20: fix: bug")
|
||||
def test_from_git_with_explicit_sha(
|
||||
self, mock_msg: MagicMock, mock_sha: MagicMock, mock_client_cls: MagicMock
|
||||
) -> None:
|
||||
mock_client = MagicMock()
|
||||
mock_client.list_project_tasks.return_value = [
|
||||
{"id": 267, "identifier": "GRM-20"},
|
||||
]
|
||||
mock_client_cls.return_value = mock_client
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(main, ["--from-git", "--commit-sha", "explicit_sha"])
|
||||
assert result.exit_code == 0
|
||||
|
||||
@patch.dict("os.environ", {"VIKUNJA_TOKEN": "tok"})
|
||||
def test_no_msg_and_no_from_git(self) -> None:
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(main, [])
|
||||
assert result.exit_code != 0
|
||||
assert "commit_msg" in result.output
|
||||
|
||||
@@ -0,0 +1,69 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
from click import ClickException
|
||||
from click.testing import CliRunner
|
||||
|
||||
import scripts.ci.push_badges as push_badges
|
||||
|
||||
|
||||
class TestGenerateBadges:
|
||||
def test_success(self, tmp_path: Path) -> None:
|
||||
output_dir = tmp_path / ".badges"
|
||||
output_dir.mkdir()
|
||||
(output_dir / "badge1.svg").touch()
|
||||
|
||||
with patch("subprocess.run") as mock_run:
|
||||
push_badges.generate_badges(str(output_dir))
|
||||
mock_run.assert_called_once()
|
||||
|
||||
def test_no_badges_generated(self, tmp_path: Path) -> None:
|
||||
output_dir = tmp_path / ".badges"
|
||||
output_dir.mkdir()
|
||||
|
||||
with patch("subprocess.run"):
|
||||
with pytest.raises(ClickException, match="No badge SVG files generated"):
|
||||
push_badges.generate_badges(str(output_dir))
|
||||
|
||||
|
||||
class TestPushToBadgesBranch:
|
||||
def test_success(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.chdir(tmp_path)
|
||||
badges_dir = tmp_path / ".badges"
|
||||
badges_dir.mkdir()
|
||||
svg = badges_dir / "badge1.svg"
|
||||
svg.write_text("<svg></svg>")
|
||||
|
||||
with patch("subprocess.run") as mock_run:
|
||||
push_badges.push_to_badges_branch(str(badges_dir))
|
||||
|
||||
# Should have called git config, checkout, rm, add, commit, push
|
||||
assert mock_run.call_count >= 6
|
||||
# Verify the SVG was copied to cwd
|
||||
assert (tmp_path / "badge1.svg").exists()
|
||||
|
||||
|
||||
class TestMain:
|
||||
def test_success(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.chdir(tmp_path)
|
||||
badges_dir = tmp_path / ".badges"
|
||||
badges_dir.mkdir()
|
||||
(badges_dir / "badge1.svg").touch()
|
||||
|
||||
runner = CliRunner()
|
||||
with patch("subprocess.run"):
|
||||
result = runner.invoke(push_badges.main, ["--output-dir", str(badges_dir)])
|
||||
assert result.exit_code == 0
|
||||
|
||||
def test_no_badges(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.chdir(tmp_path)
|
||||
badges_dir = tmp_path / ".badges"
|
||||
badges_dir.mkdir()
|
||||
|
||||
runner = CliRunner()
|
||||
with patch("subprocess.run"):
|
||||
result = runner.invoke(push_badges.main, ["--output-dir", str(badges_dir)])
|
||||
assert result.exit_code != 0
|
||||
@@ -0,0 +1,107 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
from unittest.mock import patch
|
||||
|
||||
import pytest
|
||||
from click.testing import CliRunner
|
||||
|
||||
import scripts.setup as setup
|
||||
|
||||
|
||||
class TestRun:
|
||||
def test_run_success(self) -> None:
|
||||
with patch("subprocess.run") as mock_run:
|
||||
setup._run(["echo", "hello"], ".venv/bin")
|
||||
mock_run.assert_called_once_with(["echo", "hello"], check=True)
|
||||
|
||||
def test_run_failure(self) -> None:
|
||||
import subprocess
|
||||
|
||||
with patch("subprocess.run", side_effect=subprocess.CalledProcessError(1, ["echo"])):
|
||||
with pytest.raises(subprocess.CalledProcessError):
|
||||
setup._run(["echo", "hello"], ".venv/bin")
|
||||
|
||||
|
||||
class TestInstallPythonDeps:
|
||||
def test_install(self) -> None:
|
||||
with patch("scripts.setup._run") as mock_run:
|
||||
setup._install_python_deps(".venv/bin")
|
||||
mock_run.assert_called_once_with([".venv/bin/pip", "install", "-e", ".[dev]"], ".venv/bin")
|
||||
|
||||
|
||||
class TestInstallAnsibleCollections:
|
||||
def test_install(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.chdir(tmp_path)
|
||||
(tmp_path / "ansible").mkdir()
|
||||
(tmp_path / "ansible/requirements.yml").write_text("collections: []")
|
||||
with patch("scripts.setup._run") as mock_run:
|
||||
setup._install_ansible_collections(".venv/bin")
|
||||
mock_run.assert_called_once()
|
||||
|
||||
def test_no_requirements(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.chdir(tmp_path)
|
||||
with patch("scripts.setup._run") as mock_run:
|
||||
setup._install_ansible_collections(".venv/bin")
|
||||
mock_run.assert_not_called()
|
||||
|
||||
|
||||
class TestInstallPreCommitHooks:
|
||||
def test_install(self) -> None:
|
||||
with patch("scripts.setup._run") as mock_run:
|
||||
setup._install_pre_commit_hooks(".venv/bin")
|
||||
assert mock_run.call_count == 3
|
||||
calls = [c.args[0] for c in mock_run.call_args_list]
|
||||
# Each call should have the pre-commit binary and --hook-type flag
|
||||
for call in calls:
|
||||
assert ".venv/bin/pre-commit" in call[0]
|
||||
assert "--hook-type" in call
|
||||
|
||||
|
||||
class TestVerify:
|
||||
def test_verify_success(self) -> None:
|
||||
import subprocess
|
||||
|
||||
mock_result = subprocess.CompletedProcess(
|
||||
args=["grm", "--version"], returncode=0, stdout="grm 1.0.0", stderr=""
|
||||
)
|
||||
with patch("subprocess.run", return_value=mock_result):
|
||||
setup._verify(".venv/bin")
|
||||
|
||||
def test_verify_not_found(self) -> None:
|
||||
|
||||
with patch("subprocess.run", side_effect=FileNotFoundError()):
|
||||
setup._verify(".venv/bin")
|
||||
|
||||
def test_verify_timeout(self) -> None:
|
||||
import subprocess
|
||||
|
||||
with patch("subprocess.run", side_effect=subprocess.TimeoutExpired(cmd=["grm", "--version"], timeout=10)):
|
||||
setup._verify(".venv/bin")
|
||||
|
||||
|
||||
class TestMain:
|
||||
def test_bin_not_found(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.chdir(tmp_path)
|
||||
runner = CliRunner()
|
||||
result = runner.invoke(setup.main, ["--bin", "nonexistent/bin"])
|
||||
assert result.exit_code != 0
|
||||
assert "Bin directory not found" in result.output
|
||||
|
||||
def test_success(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
monkeypatch.chdir(tmp_path)
|
||||
bin_dir = tmp_path / ".venv" / "bin"
|
||||
bin_dir.mkdir(parents=True)
|
||||
(bin_dir / "grm").touch()
|
||||
(bin_dir / "pre-commit").touch()
|
||||
(tmp_path / "ansible").mkdir()
|
||||
(tmp_path / "ansible/requirements.yml").write_text("collections: []")
|
||||
|
||||
runner = CliRunner()
|
||||
with patch("scripts.setup._install_python_deps"):
|
||||
with patch("scripts.setup._install_ansible_collections"):
|
||||
with patch("scripts.setup._install_pre_commit_hooks"):
|
||||
with patch("scripts.setup._verify"):
|
||||
result = runner.invoke(setup.main, ["--bin", str(bin_dir)])
|
||||
assert result.exit_code == 0
|
||||
assert "Setup complete" in result.output
|
||||
Reference in New Issue
Block a user