Public Access
emil
0228fce5b9
Post-merge / detect-type (push) Successful in 10s
Post-merge / validate-commit-msg (push) Successful in 10s
Post-merge / configure-repo (push) Successful in 11s
Post-merge / sync-wiki (push) Successful in 17s
Post-merge / vikunja (push) Successful in 18s
Post-merge / release (push) Successful in 36s
Post-merge / publish (push) Successful in 20s
Post-merge / badges (push) Successful in 35s
234 lines
11 KiB
Python
234 lines
11 KiB
Python
"""Unit tests for devx/molecule/discover_runners.py."""
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import click
|
|
import pytest
|
|
from click.testing import CliRunner
|
|
|
|
from devx.molecule.discover_runners import (
|
|
DEFAULT_MAX_RUNNERS,
|
|
generate_indices,
|
|
get_runner_count,
|
|
main,
|
|
query_runners,
|
|
)
|
|
|
|
|
|
class TestGenerateIndices:
|
|
def test_zero(self) -> None:
|
|
assert generate_indices(0) == []
|
|
|
|
def test_one(self) -> None:
|
|
assert generate_indices(1) == ["1"]
|
|
|
|
def test_three(self) -> None:
|
|
assert generate_indices(3) == ["1", "2", "3"]
|
|
|
|
def test_five(self) -> None:
|
|
assert generate_indices(5) == ["1", "2", "3", "4", "5"]
|
|
|
|
|
|
class TestQueryRunners:
|
|
@patch("devx.molecule.discover_runners.requests.get")
|
|
def test_returns_total_from_all_levels(self, mock_get: MagicMock) -> None:
|
|
"""Runners from repo, org, and admin levels are summed."""
|
|
responses = [
|
|
MagicMock(status_code=200, json=lambda: {"runners": [], "total_count": 2}),
|
|
MagicMock(status_code=200, json=lambda: {"runners": [], "total_count": 1}),
|
|
MagicMock(status_code=200, json=lambda: {"runners": [], "total_count": 3}),
|
|
]
|
|
mock_get.side_effect = responses
|
|
result = query_runners("https://api.example.com", "token", "owner", "repo")
|
|
assert result == 6
|
|
|
|
@patch("devx.molecule.discover_runners.requests.get")
|
|
def test_skips_non_200(self, mock_get: MagicMock) -> None:
|
|
"""Non-200 responses (e.g., 403 for admin) are skipped."""
|
|
responses = [
|
|
MagicMock(status_code=200, json=lambda: {"total_count": 2}),
|
|
MagicMock(status_code=200, json=lambda: {"total_count": 1}),
|
|
MagicMock(status_code=403, json=lambda: {"message": "forbidden"}),
|
|
]
|
|
mock_get.side_effect = responses
|
|
result = query_runners("https://api.example.com", "token", "owner", "repo")
|
|
assert result == 3
|
|
|
|
@patch("devx.molecule.discover_runners.requests.get")
|
|
def test_handles_request_exception(self, mock_get: MagicMock) -> None:
|
|
"""Network errors are caught and don't crash."""
|
|
mock_get.side_effect = [
|
|
MagicMock(status_code=200, json=lambda: {"total_count": 1}),
|
|
MagicMock(side_effect=__import__("requests").RequestException("network error")),
|
|
MagicMock(status_code=200, json=lambda: {"total_count": 2}),
|
|
]
|
|
result = query_runners("https://api.example.com", "token", "owner", "repo")
|
|
assert result == 3
|
|
|
|
@patch("devx.molecule.discover_runners.requests.get")
|
|
def test_all_failures_return_zero(self, mock_get: MagicMock) -> None:
|
|
"""When all API calls fail, returns 0."""
|
|
mock_get.side_effect = [
|
|
MagicMock(status_code=404),
|
|
MagicMock(status_code=404),
|
|
MagicMock(status_code=403),
|
|
]
|
|
result = query_runners("https://api.example.com", "token", "owner", "repo")
|
|
assert result == 0
|
|
|
|
@patch("devx.molecule.discover_runners.requests.get")
|
|
def test_value_error_on_repo_level(self, mock_get: MagicMock) -> None:
|
|
"""JSON parse error on repo level is caught."""
|
|
responses = [
|
|
MagicMock(status_code=200, json=MagicMock(side_effect=ValueError("bad json"))),
|
|
MagicMock(status_code=200, json=lambda: {"total_count": 2}),
|
|
MagicMock(status_code=200, json=lambda: {"total_count": 1}),
|
|
]
|
|
mock_get.side_effect = responses
|
|
result = query_runners("https://api.example.com", "token", "owner", "repo")
|
|
assert result == 3
|
|
|
|
@patch("devx.molecule.discover_runners.requests.get")
|
|
def test_value_error_on_org_level(self, mock_get: MagicMock) -> None:
|
|
"""JSON parse error on org level is caught."""
|
|
responses = [
|
|
MagicMock(status_code=200, json=lambda: {"total_count": 1}),
|
|
MagicMock(status_code=200, json=MagicMock(side_effect=ValueError("bad json"))),
|
|
MagicMock(status_code=200, json=lambda: {"total_count": 2}),
|
|
]
|
|
mock_get.side_effect = responses
|
|
result = query_runners("https://api.example.com", "token", "owner", "repo")
|
|
assert result == 3
|
|
|
|
@patch("devx.molecule.discover_runners.requests.get")
|
|
def test_value_error_on_admin_level(self, mock_get: MagicMock) -> None:
|
|
"""JSON parse error on admin level is caught."""
|
|
responses = [
|
|
MagicMock(status_code=200, json=lambda: {"total_count": 1}),
|
|
MagicMock(status_code=200, json=lambda: {"total_count": 2}),
|
|
MagicMock(status_code=200, json=MagicMock(side_effect=ValueError("bad json"))),
|
|
]
|
|
mock_get.side_effect = responses
|
|
result = query_runners("https://api.example.com", "token", "owner", "repo")
|
|
assert result == 3
|
|
|
|
@patch("devx.molecule.discover_runners.requests.get")
|
|
def test_request_exception_on_all_levels(self, mock_get: MagicMock) -> None:
|
|
"""Network errors on all levels return 0."""
|
|
mock_get.side_effect = __import__("requests").RequestException("network error")
|
|
result = query_runners("https://api.example.com", "token", "owner", "repo")
|
|
assert result == 0
|
|
|
|
|
|
class TestGetRunnerCount:
|
|
@patch("devx.molecule.discover_runners.query_runners", return_value=5)
|
|
def test_uses_api_count_when_positive(self, mock_query: MagicMock) -> None:
|
|
result = get_runner_count("https://api.example.com", "token", "owner", "repo")
|
|
assert result == 5
|
|
|
|
@patch("devx.molecule.discover_runners.query_runners", return_value=0)
|
|
@patch.dict("os.environ", {"MOLECULE_RUNNERS": "4"})
|
|
def test_falls_back_to_env_var(self, mock_query: MagicMock) -> None:
|
|
result = get_runner_count("https://api.example.com", "token", "owner", "repo")
|
|
assert result == 4
|
|
|
|
@patch("devx.molecule.discover_runners.query_runners", return_value=0)
|
|
@patch.dict("os.environ", {"MOLECULE_RUNNERS": "invalid"})
|
|
def test_falls_back_to_default_on_invalid_env(self, mock_query: MagicMock) -> None:
|
|
result = get_runner_count("https://api.example.com", "token", "owner", "repo")
|
|
assert result == DEFAULT_MAX_RUNNERS
|
|
|
|
@patch("devx.molecule.discover_runners.query_runners", return_value=0)
|
|
@patch.dict("os.environ", {}, clear=True)
|
|
def test_falls_back_to_default_when_no_env(self, mock_query: MagicMock) -> None:
|
|
result = get_runner_count("https://api.example.com", "token", "owner", "repo")
|
|
assert result == DEFAULT_MAX_RUNNERS
|
|
|
|
@patch("devx.molecule.discover_runners.query_runners", return_value=0)
|
|
@patch.dict("os.environ", {"MOLECULE_RUNNERS": "0"})
|
|
def test_env_var_zero_falls_back_to_default(self, mock_query: MagicMock) -> None:
|
|
"""MOLECULE_RUNNERS=0 is invalid, falls back to default."""
|
|
result = get_runner_count("https://api.example.com", "token", "owner", "repo")
|
|
assert result == DEFAULT_MAX_RUNNERS
|
|
|
|
@patch("devx.molecule.discover_runners.query_runners", return_value=0)
|
|
@patch.dict("os.environ", {}, clear=True)
|
|
def test_no_token_uses_env_var(self, mock_query: MagicMock) -> None:
|
|
"""When no token, skips API and uses env/default."""
|
|
with patch.dict("os.environ", {"MOLECULE_RUNNERS": "2"}):
|
|
result = get_runner_count("https://api.example.com", "", "owner", "repo")
|
|
assert result == 2
|
|
|
|
|
|
class TestMain:
|
|
@patch("devx.molecule.discover_runners.get_runner_count", return_value=3)
|
|
def test_default_output(self, mock_count: MagicMock) -> None:
|
|
runner = CliRunner()
|
|
result = runner.invoke(main, [])
|
|
assert result.exit_code == 0
|
|
assert "count=3" in result.output
|
|
assert 'indices=["1", "2", "3"]' in result.output
|
|
|
|
@patch("devx.molecule.discover_runners.get_runner_count", return_value=5)
|
|
def test_count_only(self, mock_count: MagicMock) -> None:
|
|
runner = CliRunner()
|
|
result = runner.invoke(main, ["--count"])
|
|
assert result.exit_code == 0
|
|
assert result.output.strip() == "5"
|
|
|
|
@patch("devx.molecule.discover_runners.get_runner_count", return_value=4)
|
|
def test_indices_only(self, mock_count: MagicMock) -> None:
|
|
runner = CliRunner()
|
|
result = runner.invoke(main, ["--indices"])
|
|
assert result.exit_code == 0
|
|
assert json.loads(result.output.strip()) == ["1", "2", "3", "4"]
|
|
|
|
@patch("devx.molecule.discover_runners.get_runner_count", return_value=1)
|
|
def test_single_runner(self, mock_count: MagicMock) -> None:
|
|
runner = CliRunner()
|
|
result = runner.invoke(main, ["--indices"])
|
|
assert result.exit_code == 0
|
|
assert json.loads(result.output.strip()) == ["1"]
|
|
|
|
@patch("devx.molecule.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("devx.molecule.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
|
|
|
|
@patch("devx.molecule.discover_runners.get_runner_count", return_value=2)
|
|
def test_explicit_owner_and_repo(self, mock_count: MagicMock) -> None:
|
|
"""When --owner and --repo are provided, env vars are not used."""
|
|
runner = CliRunner()
|
|
result = runner.invoke(main, ["--owner", "myorg", "--repo", "myrepo"])
|
|
assert result.exit_code == 0
|
|
mock_count.assert_called_once()
|
|
args, kwargs = mock_count.call_args
|
|
assert "myorg" in args
|
|
assert "myrepo" in args
|
|
|
|
@patch("devx.molecule.discover_runners.get_ci_token", side_effect=click.ClickException("no token"))
|
|
@patch("devx.molecule.discover_runners.get_runner_count", return_value=3)
|
|
def test_missing_token_runs_without_api(self, mock_count: MagicMock, mock_token: MagicMock) -> None:
|
|
"""When no token is available, runner discovery falls back to env/default."""
|
|
runner = CliRunner()
|
|
result = runner.invoke(main, ["--count"])
|
|
assert result.exit_code == 0
|
|
assert result.output.strip() == "3"
|
|
args, _ = mock_count.call_args
|
|
assert args[1] is None # token passed as None when missing
|