136 lines
4.7 KiB
Python
136 lines
4.7 KiB
Python
"""Unit tests for scripts/distribute_molecule.py."""
|
|
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
import click
|
|
import pytest
|
|
|
|
from scripts.distribute_molecule import (
|
|
MOLECULE_ROOT,
|
|
discover_scenarios,
|
|
distribute,
|
|
scenarios_for_runner,
|
|
)
|
|
|
|
|
|
class TestDiscoverScenarios:
|
|
def test_discovers_scenarios(self, tmp_path: Path) -> None:
|
|
root = tmp_path / "molecule"
|
|
(root / "default").mkdir(parents=True)
|
|
(root / "binary").mkdir(parents=True)
|
|
(root / "common").mkdir(parents=True)
|
|
(root / "_shared").mkdir(parents=True)
|
|
result = discover_scenarios(root)
|
|
assert result == ["binary", "default"]
|
|
|
|
def test_raises_when_dir_missing(self, tmp_path: Path) -> None:
|
|
with pytest.raises(click.ClickException) as exc:
|
|
discover_scenarios(tmp_path / "nonexistent")
|
|
assert "not found" in str(exc.value)
|
|
|
|
def test_default_root_constant(self) -> None:
|
|
assert Path("ansible/roles/gitea-runner/molecule") == MOLECULE_ROOT
|
|
|
|
|
|
class TestDistribute:
|
|
def test_even_split(self) -> None:
|
|
scenarios = ["a", "b", "c", "d", "e", "f"]
|
|
groups = distribute(scenarios, 3)
|
|
assert groups == [["a", "d"], ["b", "e"], ["c", "f"]]
|
|
|
|
def test_uneven_split(self) -> None:
|
|
scenarios = ["a", "b", "c", "d", "e"]
|
|
groups = distribute(scenarios, 3)
|
|
assert groups == [["a", "d"], ["b", "e"], ["c"]]
|
|
|
|
def test_more_runners_than_scenarios(self) -> None:
|
|
scenarios = ["a", "b"]
|
|
groups = distribute(scenarios, 5)
|
|
assert groups == [["a"], ["b"], [], [], []]
|
|
|
|
def test_single_runner(self) -> None:
|
|
scenarios = ["a", "b", "c"]
|
|
groups = distribute(scenarios, 1)
|
|
assert groups == [["a", "b", "c"]]
|
|
|
|
def test_empty_scenarios(self) -> None:
|
|
groups = distribute([], 3)
|
|
assert groups == [[], [], []]
|
|
|
|
|
|
class TestScenariosForRunner:
|
|
def test_returns_correct_subset(self) -> None:
|
|
scenarios = ["a", "b", "c", "d", "e", "f"]
|
|
assert scenarios_for_runner(scenarios, 0, 3) == ["a", "d"]
|
|
assert scenarios_for_runner(scenarios, 1, 3) == ["b", "e"]
|
|
assert scenarios_for_runner(scenarios, 2, 3) == ["c", "f"]
|
|
|
|
def test_out_of_range_raises(self) -> None:
|
|
with pytest.raises(click.ClickException) as exc:
|
|
scenarios_for_runner(["a"], 5, 3)
|
|
assert "out of range" in str(exc.value)
|
|
|
|
def test_negative_index_raises(self) -> None:
|
|
with pytest.raises(click.ClickException) as exc:
|
|
scenarios_for_runner(["a"], -1, 3)
|
|
assert "out of range" in str(exc.value)
|
|
|
|
|
|
class TestCli:
|
|
def test_list_flag(self, tmp_path: Path) -> None:
|
|
from click.testing import CliRunner
|
|
|
|
from scripts.distribute_molecule import cli
|
|
|
|
root = tmp_path / "molecule"
|
|
(root / "alpha").mkdir(parents=True)
|
|
(root / "beta").mkdir(parents=True)
|
|
with patch("scripts.distribute_molecule.MOLECULE_ROOT", root):
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["--list"])
|
|
assert result.exit_code == 0
|
|
assert "alpha" in result.output
|
|
assert "beta" in result.output
|
|
|
|
def test_no_runner_index_prints_all_groups(self, tmp_path: Path) -> None:
|
|
from click.testing import CliRunner
|
|
|
|
from scripts.distribute_molecule import cli
|
|
|
|
root = tmp_path / "molecule"
|
|
for s in ["a", "b", "c"]:
|
|
(root / s).mkdir(parents=True)
|
|
with patch("scripts.distribute_molecule.MOLECULE_ROOT", root):
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["--max-runners", "3"])
|
|
assert result.exit_code == 0
|
|
assert "Runner 0:" in result.output
|
|
assert "Runner 1:" in result.output
|
|
assert "Runner 2:" in result.output
|
|
|
|
def test_runner_index_prints_assigned(self, tmp_path: Path) -> None:
|
|
from click.testing import CliRunner
|
|
|
|
from scripts.distribute_molecule import cli
|
|
|
|
root = tmp_path / "molecule"
|
|
for s in ["a", "b", "c"]:
|
|
(root / s).mkdir(parents=True)
|
|
with patch("scripts.distribute_molecule.MOLECULE_ROOT", root):
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["--runner-index", "1", "--max-runners", "3"])
|
|
assert result.exit_code == 0
|
|
assert result.output.strip() == "b"
|
|
|
|
|
|
def test_main_module_block() -> None:
|
|
import scripts.distribute_molecule as dm
|
|
|
|
with open(dm.__file__) as f:
|
|
source = f.read()
|
|
source = source.replace('if __name__ == "__main__":\n cli()\n', "")
|
|
namespace = dict(dm.__dict__)
|
|
exec(compile(source, dm.__file__, "exec"), namespace)
|
|
assert callable(namespace["cli"])
|