Public Access
emil
e796b06a91
Post-merge / detect-type (push) Successful in 9s
Post-merge / validate-commit-msg (push) Successful in 9s
Post-merge / vikunja (push) Successful in 15s
Post-merge / configure-repo (push) Successful in 19s
Post-merge / release (push) Successful in 45s
Post-merge / sync-wiki (push) Successful in 46s
Post-merge / badges (push) Successful in 46s
Post-merge / publish (push) Successful in 17s
639 lines
27 KiB
Python
639 lines
27 KiB
Python
"""Unit tests for scripts/ci/distribute_molecule.py."""
|
|
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
import click
|
|
import pytest
|
|
from click.testing import CliRunner
|
|
|
|
from devx.molecule.distribute_molecule import (
|
|
DEFAULT_ROLES_ROOT,
|
|
MOLECULE_ROOT,
|
|
PLATFORMS,
|
|
MultiRoleTestPair,
|
|
TestPair,
|
|
_load_molecule_weights,
|
|
_lpt_distribute,
|
|
_scenario_weight,
|
|
build_multi_role_pairs,
|
|
build_pairs,
|
|
cli,
|
|
discover_multi_role_scenarios,
|
|
discover_scenarios,
|
|
distribute,
|
|
distribute_multi_role,
|
|
multi_role_pairs_for_runner,
|
|
pairs_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 TestPairEncoding:
|
|
def test_encode_roundtrip(self) -> None:
|
|
pair = TestPair("default", {"name": "ubuntu-2204", "image": "ubuntu:22.04", "command": ""})
|
|
encoded = pair.encode()
|
|
assert encoded == "default|ubuntu-2204|ubuntu:22.04|"
|
|
decoded = TestPair.decode(encoded)
|
|
assert decoded.scenario == "default"
|
|
assert decoded.platform["name"] == "ubuntu-2204"
|
|
assert decoded.platform["image"] == "ubuntu:22.04"
|
|
assert decoded.platform["command"] == ""
|
|
|
|
def test_encode_with_command(self) -> None:
|
|
pair = TestPair(
|
|
"default",
|
|
{"name": "archlinux", "image": "archlinux:latest", "command": "/usr/lib/systemd/systemd"},
|
|
)
|
|
encoded = pair.encode()
|
|
assert encoded == "default|archlinux|archlinux:latest|/usr/lib/systemd/systemd"
|
|
decoded = TestPair.decode(encoded)
|
|
assert decoded.platform["command"] == "/usr/lib/systemd/systemd"
|
|
|
|
|
|
class TestBuildPairs:
|
|
def test_cross_product(self) -> None:
|
|
scenarios = ["a", "b"]
|
|
platforms = [
|
|
{"name": "p1", "image": "img1", "command": ""},
|
|
{"name": "p2", "image": "img2", "command": ""},
|
|
]
|
|
pairs = build_pairs(scenarios, platforms)
|
|
assert len(pairs) == 4
|
|
assert pairs[0].scenario == "a"
|
|
assert pairs[0].platform["name"] == "p1"
|
|
assert pairs[1].scenario == "a"
|
|
assert pairs[1].platform["name"] == "p2"
|
|
assert pairs[2].scenario == "b"
|
|
assert pairs[2].platform["name"] == "p1"
|
|
assert pairs[3].scenario == "b"
|
|
assert pairs[3].platform["name"] == "p2"
|
|
|
|
def test_default_platforms(self) -> None:
|
|
pairs = build_pairs(["default"])
|
|
assert len(pairs) == len(PLATFORMS)
|
|
assert all(p.scenario == "default" for p in pairs)
|
|
|
|
|
|
class TestDistribute:
|
|
def test_even_split(self) -> None:
|
|
pairs = [TestPair(f"s{i}", {"name": "p", "image": "i", "command": ""}) for i in range(6)]
|
|
groups = distribute(pairs, 3)
|
|
assert len(groups) == 3
|
|
assert len(groups[0]) == 2
|
|
assert len(groups[1]) == 2
|
|
assert len(groups[2]) == 2
|
|
|
|
def test_uneven_split(self) -> None:
|
|
pairs = [TestPair(f"s{i}", {"name": "p", "image": "i", "command": ""}) for i in range(5)]
|
|
groups = distribute(pairs, 3)
|
|
assert len(groups[0]) == 2
|
|
assert len(groups[1]) == 2
|
|
assert len(groups[2]) == 1
|
|
|
|
def test_more_runners_than_pairs(self) -> None:
|
|
pairs = [TestPair("a", {"name": "p", "image": "i", "command": ""})]
|
|
groups = distribute(pairs, 5)
|
|
assert len(groups) == 5
|
|
assert len(groups[0]) == 1
|
|
assert all(len(g) == 0 for g in groups[1:])
|
|
|
|
def test_empty_pairs(self) -> None:
|
|
groups = distribute([], 3)
|
|
assert groups == [[], [], []]
|
|
|
|
|
|
class TestPairsForRunner:
|
|
def test_returns_correct_subset(self) -> None:
|
|
pairs = [TestPair(f"s{i}", {"name": "p", "image": "i", "command": ""}) for i in range(6)]
|
|
assert len(pairs_for_runner(pairs, 0, 3)) == 2
|
|
assert len(pairs_for_runner(pairs, 1, 3)) == 2
|
|
assert len(pairs_for_runner(pairs, 2, 3)) == 2
|
|
|
|
def test_out_of_range_raises(self) -> None:
|
|
pairs = [TestPair("a", {"name": "p", "image": "i", "command": ""})]
|
|
with pytest.raises(click.ClickException) as exc:
|
|
pairs_for_runner(pairs, 5, 3)
|
|
assert "out of range" in str(exc.value)
|
|
|
|
def test_negative_index_raises(self) -> None:
|
|
pairs = [TestPair("a", {"name": "p", "image": "i", "command": ""})]
|
|
with pytest.raises(click.ClickException) as exc:
|
|
pairs_for_runner(pairs, -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 devx.molecule.distribute_molecule import cli
|
|
|
|
root = tmp_path / "molecule"
|
|
(root / "alpha").mkdir(parents=True)
|
|
(root / "beta").mkdir(parents=True)
|
|
with patch("devx.molecule.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_list_platforms_flag(self) -> None:
|
|
from click.testing import CliRunner
|
|
|
|
from devx.molecule.distribute_molecule import cli
|
|
|
|
with patch("devx.molecule.distribute_molecule.discover_scenarios", return_value=["dummy"]):
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["--list-platforms"])
|
|
assert result.exit_code == 0
|
|
assert "ubuntu-2604" in result.output
|
|
|
|
def test_no_runner_index_prints_all_groups(self, tmp_path: Path) -> None:
|
|
from click.testing import CliRunner
|
|
|
|
from devx.molecule.distribute_molecule import cli
|
|
|
|
root = tmp_path / "molecule"
|
|
for s in ["a", "b", "c"]:
|
|
(root / s).mkdir(parents=True)
|
|
with patch("devx.molecule.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 devx.molecule.distribute_molecule import cli
|
|
|
|
root = tmp_path / "molecule"
|
|
(root / "alpha").mkdir(parents=True)
|
|
with patch("devx.molecule.distribute_molecule.MOLECULE_ROOT", root):
|
|
runner = CliRunner()
|
|
# 1-based index: "1" maps to internal 0
|
|
result = runner.invoke(cli, ["--runner-index", "1", "--max-runners", "3"])
|
|
assert result.exit_code == 0
|
|
# Output should contain encoded pairs with platform info
|
|
assert "alpha|" in result.output
|
|
assert "ubuntu-2604" 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("devx.molecule.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("devx.molecule.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("devx.molecule.distribute_molecule.MOLECULE_ROOT", root):
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["--runner-index", "1", "--max-runners", "3", "--github-env"])
|
|
assert result.exit_code != 0
|
|
|
|
|
|
class TestRunnerIndexValidation:
|
|
def test_runner_index_zero_raises(self) -> None:
|
|
"""Runner index < 1 should raise."""
|
|
with patch("devx.molecule.distribute_molecule.discover_scenarios", return_value=["dummy"]):
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["--runner-index", "0", "--max-runners", "3"])
|
|
assert result.exit_code != 0
|
|
assert "out of range" in result.output
|
|
|
|
|
|
def test_main_module_block() -> None:
|
|
import devx.molecule.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"])
|
|
|
|
|
|
class TestDiscoverMultiRole:
|
|
def test_discovers_role_scenario_pairs(self, tmp_path: Path) -> None:
|
|
roles = tmp_path / "roles"
|
|
for scenario in ["default", "binary"]:
|
|
(roles / "gitea-runner" / "molecule" / scenario).mkdir(parents=True)
|
|
(roles / "gitea-runner" / "molecule" / "common").mkdir(parents=True)
|
|
(roles / "gitea-runner" / "molecule" / "_shared").mkdir(parents=True)
|
|
(roles / "docker-base" / "molecule" / "default").mkdir(parents=True)
|
|
(roles / "no-molecule").mkdir(parents=True)
|
|
result = discover_multi_role_scenarios(roles)
|
|
assert ("docker-base", "default") in result
|
|
assert ("gitea-runner", "default") in result
|
|
assert ("gitea-runner", "binary") in result
|
|
assert ("gitea-runner", "common") not in result
|
|
assert ("gitea-runner", "_shared") not in result
|
|
assert len(result) == 3
|
|
|
|
def test_raises_when_dir_missing(self, tmp_path: Path) -> None:
|
|
with pytest.raises(click.ClickException) as exc:
|
|
discover_multi_role_scenarios(tmp_path / "nonexistent")
|
|
assert "not found" in str(exc.value)
|
|
|
|
def test_default_roles_root_raises_when_missing(self, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Calling with no args uses DEFAULT_ROLES_ROOT which doesn't exist in tests."""
|
|
with pytest.raises(click.ClickException):
|
|
discover_multi_role_scenarios()
|
|
|
|
def test_default_roles_root_constant(self) -> None:
|
|
assert Path("ansible/roles") == DEFAULT_ROLES_ROOT
|
|
|
|
|
|
class TestMultiRoleTestPair:
|
|
def test_encode_roundtrip(self) -> None:
|
|
pair = MultiRoleTestPair(
|
|
"docker-base", "default", {"name": "ubuntu-2204", "image": "ubuntu:22.04", "command": ""}
|
|
)
|
|
encoded = pair.encode()
|
|
assert encoded == "docker-base|default|ubuntu-2204|ubuntu:22.04|"
|
|
decoded = MultiRoleTestPair.decode(encoded)
|
|
assert decoded.role == "docker-base"
|
|
assert decoded.scenario == "default"
|
|
assert decoded.platform["name"] == "ubuntu-2204"
|
|
|
|
|
|
class TestBuildMultiRolePairs:
|
|
def test_cross_product(self) -> None:
|
|
role_scenarios = [("role-a", "default"), ("role-b", "binary")]
|
|
platforms = [{"name": "p1", "image": "i1", "command": ""}]
|
|
pairs = build_multi_role_pairs(role_scenarios, platforms)
|
|
assert len(pairs) == 2
|
|
assert pairs[0].role == "role-a"
|
|
assert pairs[1].role == "role-b"
|
|
|
|
def test_default_platforms(self) -> None:
|
|
pairs = build_multi_role_pairs([("r", "s")])
|
|
assert len(pairs) == len(PLATFORMS)
|
|
|
|
|
|
class TestDistributeMultiRole:
|
|
def test_even_split(self) -> None:
|
|
pairs = [MultiRoleTestPair(f"r{i}", "s", {"name": "p", "image": "i", "command": ""}) for i in range(6)]
|
|
groups = distribute_multi_role(pairs, 3)
|
|
assert all(len(g) == 2 for g in groups)
|
|
|
|
def test_out_of_range_raises(self) -> None:
|
|
pairs = [MultiRoleTestPair("r", "s", {"name": "p", "image": "i", "command": ""})]
|
|
with pytest.raises(click.ClickException):
|
|
multi_role_pairs_for_runner(pairs, 5, 3)
|
|
|
|
|
|
class TestCliMultiRole:
|
|
def test_roles_root_list(self, tmp_path: Path) -> None:
|
|
roles = tmp_path / "roles"
|
|
(roles / "role-a" / "molecule" / "default").mkdir(parents=True)
|
|
(roles / "role-b" / "molecule" / "binary").mkdir(parents=True)
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["--roles-root", str(roles), "--list"])
|
|
assert result.exit_code == 0
|
|
assert "role-a|default" in result.output
|
|
assert "role-b|binary" in result.output
|
|
|
|
def test_roles_root_runner_index(self, tmp_path: Path) -> None:
|
|
roles = tmp_path / "roles"
|
|
(roles / "role-a" / "molecule" / "default").mkdir(parents=True)
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["--roles-root", str(roles), "--runner-index", "1", "--max-runners", "3"])
|
|
assert result.exit_code == 0
|
|
assert "role-a|default|" in result.output
|
|
|
|
def test_roles_root_github_env(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
gh_file = tmp_path / "env.txt"
|
|
monkeypatch.setenv("GITHUB_ENV", str(gh_file))
|
|
roles = tmp_path / "roles"
|
|
(roles / "role-a" / "molecule" / "default").mkdir(parents=True)
|
|
runner = CliRunner()
|
|
result = runner.invoke(
|
|
cli,
|
|
["--roles-root", str(roles), "--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_roles_root_skip_if_excess(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
gh_file = tmp_path / "env.txt"
|
|
monkeypatch.setenv("GITHUB_ENV", str(gh_file))
|
|
roles = tmp_path / "roles"
|
|
(roles / "role-a" / "molecule" / "default").mkdir(parents=True)
|
|
runner = CliRunner()
|
|
result = runner.invoke(
|
|
cli,
|
|
[
|
|
"--roles-root",
|
|
str(roles),
|
|
"--runner-index",
|
|
"5",
|
|
"--max-runners",
|
|
"2",
|
|
"--github-env",
|
|
"--skip-if-excess",
|
|
],
|
|
)
|
|
assert result.exit_code == 0
|
|
content = gh_file.read_text()
|
|
assert "SKIP=true" in content
|
|
|
|
def test_molecule_root_option(self, tmp_path: Path) -> None:
|
|
root = tmp_path / "custom-molecule"
|
|
(root / "alpha").mkdir(parents=True)
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["--molecule-root", str(root), "--list"])
|
|
assert result.exit_code == 0
|
|
assert "alpha" in result.output
|
|
|
|
def test_roles_root_list_platforms(self, tmp_path: Path) -> None:
|
|
"""--roles-root --list-platforms prints platforms."""
|
|
roles = tmp_path / "roles"
|
|
(roles / "role-a" / "molecule" / "default").mkdir(parents=True)
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["--roles-root", str(roles), "--list-platforms"])
|
|
assert result.exit_code == 0
|
|
assert "ubuntu-2604" in result.output
|
|
|
|
def test_roles_root_no_runner_index_prints_groups(self, tmp_path: Path) -> None:
|
|
"""--roles-root without --runner-index prints all groups."""
|
|
roles = tmp_path / "roles"
|
|
(roles / "role-a" / "molecule" / "default").mkdir(parents=True)
|
|
(roles / "role-b" / "molecule" / "binary").mkdir(parents=True)
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["--roles-root", str(roles), "--max-runners", "2"])
|
|
assert result.exit_code == 0
|
|
assert "Runner 0:" in result.output
|
|
assert "Runner 1:" in result.output
|
|
|
|
def test_platforms_file_overrides_default(self, tmp_path: Path) -> None:
|
|
"""--platforms-file loads custom platforms from JSON."""
|
|
import json
|
|
|
|
from click.testing import CliRunner
|
|
|
|
from devx.molecule.distribute_molecule import cli
|
|
|
|
roles = tmp_path / "roles"
|
|
(roles / "role-a" / "molecule" / "default").mkdir(parents=True)
|
|
platforms_file = tmp_path / "platforms.json"
|
|
custom = [{"name": "custom-os", "image": "custom:latest", "command": "sleep infinity"}]
|
|
platforms_file.write_text(json.dumps(custom))
|
|
runner = CliRunner()
|
|
result = runner.invoke(
|
|
cli, ["--roles-root", str(roles), "--platforms-file", str(platforms_file), "--list-platforms"]
|
|
)
|
|
assert result.exit_code == 0
|
|
assert "custom-os" in result.output
|
|
assert "custom:latest" in result.output
|
|
|
|
def test_roles_root_skips_non_dir_role(self, tmp_path: Path) -> None:
|
|
"""Non-directory entries in roles root are skipped."""
|
|
roles = tmp_path / "roles"
|
|
roles.mkdir(parents=True)
|
|
(roles / "README.md").write_text("not a role")
|
|
(roles / "role-a" / "molecule" / "default").mkdir(parents=True)
|
|
result = discover_multi_role_scenarios(roles)
|
|
assert ("role-a", "default") in result
|
|
assert len(result) == 1
|
|
|
|
def test_roles_root_skips_non_dir_scenario(self, tmp_path: Path) -> None:
|
|
"""Non-directory entries in molecule dir are skipped."""
|
|
roles = tmp_path / "roles"
|
|
(roles / "role-a" / "molecule").mkdir(parents=True)
|
|
(roles / "role-a" / "molecule" / "default").mkdir(parents=True)
|
|
(roles / "role-a" / "molecule" / "file.txt").write_text("not a scenario")
|
|
result = discover_multi_role_scenarios(roles)
|
|
assert ("role-a", "default") in result
|
|
assert len(result) == 1
|
|
|
|
def test_roles_root_skips_role_without_molecule(self, tmp_path: Path) -> None:
|
|
"""Roles without a molecule/ directory are skipped."""
|
|
roles = tmp_path / "roles"
|
|
(roles / "role-a" / "molecule" / "default").mkdir(parents=True)
|
|
(roles / "no-molecule").mkdir(parents=True)
|
|
result = discover_multi_role_scenarios(roles)
|
|
assert ("role-a", "default") in result
|
|
assert len(result) == 1
|
|
|
|
def test_roles_root_runner_index_zero_raises(self, tmp_path: Path) -> None:
|
|
"""--roles-root --runner-index 0 should raise."""
|
|
roles = tmp_path / "roles"
|
|
(roles / "role-a" / "molecule" / "default").mkdir(parents=True)
|
|
runner = CliRunner()
|
|
result = runner.invoke(cli, ["--roles-root", str(roles), "--runner-index", "0", "--max-runners", "3"])
|
|
assert result.exit_code != 0
|
|
assert "out of range" in result.output
|
|
|
|
|
|
class TestScenarioWeight:
|
|
def test_default_weight_no_config(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Without pyproject.toml, all scenarios get the default weight."""
|
|
monkeypatch.chdir(tmp_path)
|
|
scenario_w, role_w = _load_molecule_weights()
|
|
assert scenario_w == {}
|
|
assert role_w == {}
|
|
assert _scenario_weight("unknown-scenario") == 3
|
|
|
|
def test_load_weights_from_pyproject(self, tmp_path: Path) -> None:
|
|
"""Weights are loaded from [tool.devx.molecule.weights] in pyproject.toml."""
|
|
pyproject = tmp_path / "pyproject.toml"
|
|
pyproject.write_text(
|
|
"[tool.devx.molecule.weights]\n"
|
|
'"nextcloud" = 15\n'
|
|
'"default" = 3\n'
|
|
'"binary" = 2\n'
|
|
'"app_container/customer-apps" = 11\n'
|
|
'"restore/default" = 11\n'
|
|
)
|
|
scenario_w, role_w = _load_molecule_weights(str(pyproject))
|
|
assert scenario_w == {"nextcloud": 15, "default": 3, "binary": 2}
|
|
assert role_w == {("app_container", "customer-apps"): 11, ("restore", "default"): 11}
|
|
|
|
def test_role_specific_takes_priority(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Role-specific weights take priority over scenario-name-only weights."""
|
|
pyproject = tmp_path / "pyproject.toml"
|
|
pyproject.write_text(
|
|
'[tool.devx.molecule.weights]\n"default" = 3\n"docker_base/default" = 8\n"restore/default" = 11\n'
|
|
)
|
|
scenario_w, role_w = _load_molecule_weights(str(pyproject))
|
|
monkeypatch.setattr("devx.molecule.distribute_molecule._SCENARIO_WEIGHTS", scenario_w)
|
|
monkeypatch.setattr("devx.molecule.distribute_molecule._ROLE_SCENARIO_WEIGHTS", role_w)
|
|
assert _scenario_weight("default", "docker_base") == 8
|
|
assert _scenario_weight("default", "restore") == 11
|
|
assert _scenario_weight("default", "app_container") == 3
|
|
|
|
def test_case_insensitive(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Weight keys are matched case-insensitively."""
|
|
pyproject = tmp_path / "pyproject.toml"
|
|
pyproject.write_text('[tool.devx.molecule.weights]\n"nextcloud" = 15\n')
|
|
scenario_w, role_w = _load_molecule_weights(str(pyproject))
|
|
monkeypatch.setattr("devx.molecule.distribute_molecule._SCENARIO_WEIGHTS", scenario_w)
|
|
monkeypatch.setattr("devx.molecule.distribute_molecule._ROLE_SCENARIO_WEIGHTS", role_w)
|
|
assert _scenario_weight("NextCloud") == 15
|
|
assert _scenario_weight("NEXTCLOUD") == 15
|
|
|
|
def test_substring_match(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
|
|
"""Scenario-name weights use substring matching."""
|
|
pyproject = tmp_path / "pyproject.toml"
|
|
pyproject.write_text('[tool.devx.molecule.weights]\n"nextcloud" = 15\n')
|
|
scenario_w, role_w = _load_molecule_weights(str(pyproject))
|
|
monkeypatch.setattr("devx.molecule.distribute_molecule._SCENARIO_WEIGHTS", scenario_w)
|
|
monkeypatch.setattr("devx.molecule.distribute_molecule._ROLE_SCENARIO_WEIGHTS", role_w)
|
|
assert _scenario_weight("nextcloud-with-redis") == 15
|
|
|
|
def test_no_pyproject_returns_empty(self, tmp_path: Path) -> None:
|
|
"""Missing pyproject.toml returns empty weight dicts."""
|
|
scenario_w, role_w = _load_molecule_weights(str(tmp_path / "nonexistent.toml"))
|
|
assert scenario_w == {}
|
|
assert role_w == {}
|
|
|
|
def test_invalid_weights_ignored(self, tmp_path: Path) -> None:
|
|
"""Non-integer weight values are silently ignored."""
|
|
pyproject = tmp_path / "pyproject.toml"
|
|
pyproject.write_text('[tool.devx.molecule.weights]\n"good" = 5\n"bad" = "not an int"\n')
|
|
scenario_w, role_w = _load_molecule_weights(str(pyproject))
|
|
assert scenario_w == {"good": 5}
|
|
assert role_w == {}
|
|
|
|
def test_malformed_toml_returns_empty(self, tmp_path: Path) -> None:
|
|
"""Malformed TOML returns empty weight dicts."""
|
|
pyproject = tmp_path / "pyproject.toml"
|
|
pyproject.write_text("this is not valid toml = = =")
|
|
scenario_w, role_w = _load_molecule_weights(str(pyproject))
|
|
assert scenario_w == {}
|
|
assert role_w == {}
|
|
|
|
def test_non_dict_weights_returns_empty(self, tmp_path: Path) -> None:
|
|
"""If [tool.devx.molecule.weights] is not a table, returns empty dicts."""
|
|
pyproject = tmp_path / "pyproject.toml"
|
|
pyproject.write_text('[tool.devx.molecule]\nweights = "not a table"\n')
|
|
scenario_w, role_w = _load_molecule_weights(str(pyproject))
|
|
assert scenario_w == {}
|
|
assert role_w == {}
|
|
|
|
|
|
class TestLptDistribute:
|
|
def test_equal_weights_produce_even_split(self) -> None:
|
|
items = list(range(6))
|
|
weights = [3, 3, 3, 3, 3, 3]
|
|
groups = _lpt_distribute(items, weights, 3)
|
|
assert all(len(g) == 2 for g in groups)
|
|
|
|
def test_heavy_items_on_different_runners(self) -> None:
|
|
"""Two heavy items should go to different runners."""
|
|
items = ["heavy-a", "heavy-b", "light-1", "light-2"]
|
|
weights = [10, 10, 1, 1]
|
|
groups = _lpt_distribute(items, weights, 2)
|
|
# Heavy items should be on different runners
|
|
flat = [item for group in groups for item in group]
|
|
assert "heavy-a" in flat
|
|
assert "heavy-b" in flat
|
|
runner_a = next(i for i, g in enumerate(groups) if "heavy-a" in g)
|
|
runner_b = next(i for i, g in enumerate(groups) if "heavy-b" in g)
|
|
assert runner_a != runner_b
|
|
|
|
def test_load_balance_with_varying_weights(self) -> None:
|
|
"""LPT should produce better load balance than round-robin."""
|
|
items = list(range(7))
|
|
# Simulate multi-role-like weights: 2 heavy, 2 medium, 3 light
|
|
weights = [10, 10, 7, 7, 3, 3, 3]
|
|
groups = _lpt_distribute(items, weights, 3)
|
|
loads = [sum(weights[i] for i in g) for g in groups]
|
|
# LPT should produce loads close to total/3 = 43/3 ≈ 14.3
|
|
# Round-robin would produce: 10+7+3=20, 10+7+3=20, 3=3 (terrible)
|
|
assert max(loads) - min(loads) <= 10 # Reasonably balanced
|
|
|
|
def test_more_runners_than_items(self) -> None:
|
|
items = ["a"]
|
|
weights = [5]
|
|
groups = _lpt_distribute(items, weights, 5)
|
|
assert len(groups) == 5
|
|
assert len(groups[0]) == 1
|
|
assert all(len(g) == 0 for g in groups[1:])
|
|
|
|
def test_empty_items(self) -> None:
|
|
groups = _lpt_distribute([], [], 3)
|
|
assert groups == [[], [], []]
|
|
|
|
def test_preserves_all_items(self) -> None:
|
|
items = ["a", "b", "c", "d", "e"]
|
|
weights = [5, 3, 8, 1, 2]
|
|
groups = _lpt_distribute(items, weights, 3)
|
|
flat = sorted(item for group in groups for item in group)
|
|
assert flat == sorted(items)
|
|
|
|
|
|
class TestDistributeLpt:
|
|
def test_nextcloud_on_separate_runners(self) -> None:
|
|
"""Two nextcloud scenarios should go to different runners."""
|
|
pairs = [
|
|
TestPair("nextcloud", {"name": "p", "image": "i", "command": ""}),
|
|
TestPair("nextcloud-backup", {"name": "p", "image": "i", "command": ""}),
|
|
TestPair("binary", {"name": "p", "image": "i", "command": ""}),
|
|
TestPair("default", {"name": "p", "image": "i", "command": ""}),
|
|
]
|
|
groups = distribute(pairs, 2)
|
|
# Both nextcloud scenarios (weight 10) should be on different runners
|
|
runner_0 = [p.scenario for p in groups[0]]
|
|
runner_1 = [p.scenario for p in groups[1]]
|
|
# nextcloud and nextcloud-backup should NOT be on the same runner
|
|
assert not ("nextcloud" in runner_0 and "nextcloud-backup" in runner_0)
|
|
assert not ("nextcloud" in runner_1 and "nextcloud-backup" in runner_1)
|