diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 998f669f..14bcb8c1 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -149,7 +149,7 @@ jobs: needs: verified if: ${{ !cancelled() && needs.verified.outputs.skip != 'true' }} runs-on: ubuntu-latest - # Each module is run on its own by discovery, dealt out by size, a rough stand-in for its work: the tests that run Ansible playbooks cost a process each, so a module's runs, not its bytes, decide its time, and the suite no longer fits one job. The timeout is a guard against a hung test, set above the heaviest shard's run, never a budget the shards are fitted to. + # Each module is run on its own by discovery, and scripts/ci/shard_unit_tests.py deals the modules out by the work counted from their tests: the tests that run Ansible playbooks cost a process each, so the playbook runs a module's tests reach, not its bytes, decide its time, and the suite no longer fits one job. The timeout is a guard against a hung test, set above the heaviest shard's run, never a budget the shards are fitted to. timeout-minutes: 30 strategy: fail-fast: false @@ -191,7 +191,7 @@ jobs: tar -xzf "${dir}/etcd.tar.gz" -C "$dir" --strip-components=1 "etcd-${ETCD_VERSION}-linux-amd64/etcd" "etcd-${ETCD_VERSION}-linux-amd64/etcdctl" echo "ETCD_BINARY=${dir}/etcd" >> "$GITHUB_ENV" - name: Unit-test the collection - # Ubuntu's ansible package puts ansible-core on the system Python the filter plugins import from. The runner image can carry another Ansible earlier on PATH, so the tests run with /usr/bin first and ANSIBLE_PLAYBOOK pointing at Ubuntu's, which makes every playbook a test runs use the same ansible-core as the filters it imports; both versions are printed for the record. A module is a shard's when its position in the list of modules, largest first, leaves this shard's index modulo the number of shards. + # Ubuntu's ansible package puts ansible-core on the system Python the filter plugins import from. The runner image can carry another Ansible earlier on PATH, so the tests run with /usr/bin first and ANSIBLE_PLAYBOOK pointing at Ubuntu's, which makes every playbook a test runs use the same ansible-core as the filters it imports; both versions are printed for the record. The modules this shard runs are the ones scripts/ci/shard_unit_tests.py deals to it, imported from this checkout, so ansible must already be importable and ANSIBLE_PLAYBOOK set, as they are here. env: SHARD: ${{ matrix.shard }} SHARDS: ${{ strategy.job-total }} @@ -200,7 +200,9 @@ jobs: export PATH="/usr/bin:${PATH}" ANSIBLE_PLAYBOOK=/usr/bin/ansible-playbook ansible-playbook --version | head -n 1 python3 -c 'import ansible.release; print("ansible-core", ansible.release.__version__)' - mapfile -t modules < <(find tests/unit -maxdepth 1 -name 'test_*.py' -printf '%s %p\n' | LC_ALL=C sort -k1,1nr -k2,2 | cut -d' ' -f2- | awk -v shard="$SHARD" -v shards="$SHARDS" '(NR - 1) % shards == shard') + # An assignment, not a process substitution, so that a failed dealing stops the step instead of leaving a shard with no modules to pass. + shard_modules=$(python3 scripts/ci/shard_unit_tests.py "$SHARD" "$SHARDS") + mapfile -t modules <<< "$shard_modules" echo "Shard ${SHARD} of ${SHARDS}: ${#modules[@]} module(s)" # Every module runs whatever an earlier one did, so one failure never hides another's; the shard fails afterwards, naming each module that failed. failed=() diff --git a/scripts/ci/shard_unit_tests.py b/scripts/ci/shard_unit_tests.py new file mode 100755 index 00000000..0f0ce347 --- /dev/null +++ b/scripts/ci/shard_unit_tests.py @@ -0,0 +1,342 @@ +#!/usr/bin/env python3 +"""Deal the unit-test modules of a directory out to CI shards by the work each module does, and print one shard's modules. + + shard_unit_tests.py SHARD SHARDS [DIRECTORY] + +SHARD is the zero-based index of the shard asked for and SHARDS how many there are. The modules (``test_*.py`` directly in DIRECTORY, which defaults to ``tests/unit`` of this repository) are printed one path per line in the order they should run. The shards partition the modules: every module is in exactly one, and the same tree always gives the same assignment. + +The work of a module is counted from its tests, not stored, in units of one Ansible task. A test's time is spent in the processes it starts, chiefly ``ansible-playbook`` runs, so a module's work is that of the process launches (``subprocess.run`` and its siblings) that each of its tests reaches, plus the one that runs the module: through the test's own body, the helpers, base-class methods and ``super()`` calls it makes, and ``setUp``, ``tearDown`` and, once per class, ``setUpClass``. Every launch costs ``TASKS_PER_PROCESS_START``, and a launch of a playbook costs in addition the tasks that playbook reaches: its own, those of each role it includes (one task file where the include names it, the average task file of a role where the name is a variable, the whole role where it names none) and those of each task file it includes by a literal name. A launch site counts once however many times a loop runs it, and of the two branches of a conditional the dearer counts. A skipped test or class does no work. + +The modules are then dealt by longest-processing-time first: the heaviest module goes to whichever shard has the least work so far, ties going to the lower shard index, and equal modules in path order. Dealing to the lightest shard keeps the heaviest shard within ``mean + (1 - 1/SHARDS) * heaviest module`` of the mean (Graham's list-scheduling bound), which a dealing by position in a list cannot promise. +""" + +from __future__ import annotations + +import ast +import builtins +import inspect +import os +import subprocess +import sys +import unittest +from collections.abc import Iterator, Mapping +from dataclasses import dataclass +from pathlib import Path +from types import FunctionType, MethodType +from typing import Any + +import yaml + +REPO_ROOT = Path(__file__).resolve().parents[2] +DEFAULT_DIRECTORY = REPO_ROOT / "tests" / "unit" + +# The unit of work is one Ansible task. Starting a process costs about as much as this many tasks run: the ratio was fitted by least squares of each module's duration in CI logs against its process starts and its playbook tasks, and what the dealing needs of it is only that a playbook start is not negligible beside a task and that a start of any process is not free. +TASKS_PER_PROCESS_START = 64 + +# Every way the standard library starts a process that the tests use. +LAUNCHERS: tuple[object, ...] = (subprocess.run, subprocess.Popen, subprocess.check_output, subprocess.check_call, subprocess.call) + +FunctionNode = ast.FunctionDef | ast.AsyncFunctionDef +INCLUDE_ROLE = ("ansible.builtin.include_role", "ansible.builtin.import_role", "include_role", "import_role") +INCLUDE_TASKS = ("ansible.builtin.include_tasks", "ansible.builtin.import_tasks", "include_tasks", "import_tasks") + + +def count_tasks(tasks: object) -> int: + """The tasks of a task list, those inside its blocks, rescues and always sections included.""" + if not isinstance(tasks, list): + return 0 + total = 0 + for task in tasks: + if isinstance(task, dict): + total += 1 + sum(count_tasks(task.get(section)) for section in ("block", "rescue", "always")) + return total + + +class PlaybookTasks: + """How many tasks a playbook reaches, read from the YAML of the playbook and of the roles and task files it names.""" + + def __init__(self, repository: Path) -> None: + self.repository = repository + self.playbooks: dict[Path, int] = {} + + def task_file(self, path: Path) -> int: + return count_tasks(yaml.safe_load(path.read_text())) + + def role(self, name: str, tasks_from: object) -> int: + """The tasks of a role as an include of it reaches them: one task file when ``tasks_from`` names it, the average task file when it is a variable, every task file when there is none.""" + directory = self.repository / "roles" / name.rsplit(".", 1)[-1] / "tasks" + if isinstance(tasks_from, str) and "{{" not in tasks_from: + return self.task_file(directory / f"{tasks_from.removesuffix('.yml')}.yml") + total = sum(self.task_file(file) for file in directory.rglob("*.yml")) + if tasks_from is None: + return total + return -(-total // len(list(directory.rglob("*.yml")))) + + def includes(self, tasks: object, playbook: Path) -> int: + """The tasks that the includes among a task list reach, without those of the list itself.""" + if not isinstance(tasks, list): + return 0 + total = 0 + for task in tasks: + if not isinstance(task, dict): + continue + total += sum(self.includes(task.get(section), playbook) for section in ("block", "rescue", "always")) + for key in INCLUDE_ROLE: + if key in task: + total += self.role(task[key]["name"], task[key].get("tasks_from")) + for key in INCLUDE_TASKS: + if key in task: + target = task[key] if isinstance(task[key], str) else task[key]["file"] + if "{{" not in target: + total += self.task_file((playbook.parent / target).resolve()) + return total + + def playbook(self, path: Path) -> int: + """The tasks of a playbook and of what it includes.""" + if path not in self.playbooks: + total = 0 + for play in yaml.safe_load(path.read_text()): + for section in ("pre_tasks", "tasks", "post_tasks"): + total += count_tasks(play.get(section)) + self.includes(play.get(section), path) + for role in play.get("roles", []): + total += self.role(role if isinstance(role, str) else role["role"], None) + self.playbooks[path] = total + return self.playbooks[path] + + +def resolve(node: ast.expr, scope: Mapping[str, Any], receiver: type | None, receiver_name: str | None) -> object | None: + """The object a name or dotted attribute chain denotes in ``scope``, or None when the expression is anything else or names nothing.""" + if isinstance(node, ast.Name): + if receiver is not None and node.id == receiver_name: + return receiver + if node.id in scope: + value: object = scope[node.id] + return value + return getattr(builtins, node.id, None) + if isinstance(node, ast.Attribute): + base = resolve(node.value, scope, receiver, receiver_name) + return getattr(base, node.attr, None) if base is not None else None + return None + + +def path_of(node: ast.expr, scope: Mapping[str, Any]) -> Path | None: + """The path an expression of names, string constants, ``/``, ``+`` and ``str()`` or ``Path()`` of those denotes, or None for any other expression.""" + if isinstance(node, ast.Constant): + return Path(node.value) if isinstance(node.value, str) else None + if isinstance(node, ast.Name): + value = scope.get(node.id) + return Path(value) if isinstance(value, (str, os.PathLike)) else None + if isinstance(node, ast.BinOp) and isinstance(node.op, ast.Div): + left, right = path_of(node.left, scope), path_of(node.right, scope) + return left / right if left is not None and right is not None else None + if isinstance(node, ast.Call) and isinstance(node.func, ast.Name) and node.func.id in ("str", "Path") and len(node.args) == 1: + return path_of(node.args[0], scope) + return None + + +@dataclass(frozen=True) +class Frame: + """The function being read and what a name in it can mean.""" + + function: FunctionType + receiver: type | None + receiver_name: str | None + # Whether the function names ANSIBLE_PLAYBOOK or ansible-playbook, which makes the processes it starts playbook runs. + runs_playbooks: bool + # What each local name is assigned, for the argument list a process is started with. + assignments: Mapping[str, ast.expr] + + +class LaunchCounter: + """Counts the work of the process launches a function reaches, following the calls that resolve to functions defined under ``directory``.""" + + def __init__(self, directory: Path, repository: Path) -> None: + self.directory = directory.resolve() + self.playbooks = PlaybookTasks(repository) + self.definitions: dict[str, dict[int, FunctionNode]] = {} + self.memo: dict[tuple[FunctionType, type | None], int] = {} + self.active: set[tuple[FunctionType, type | None]] = set() + + def definition(self, function: FunctionType) -> FunctionNode | None: + """The syntax tree of a function's definition when its source file is under the directory.""" + file = Path(function.__code__.co_filename) + if not file.is_absolute() or not file.resolve().is_relative_to(self.directory): + return None + key = str(file) + if key not in self.definitions: + found: dict[int, FunctionNode] = {} + for node in ast.walk(ast.parse(file.read_text())): + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)): + # co_firstlineno is the first decorator's line when there are decorators, and the ``def`` line otherwise. + found[min([node.lineno, *(decorator.lineno for decorator in node.decorator_list)])] = node + self.definitions[key] = found + return self.definitions[key].get(function.__code__.co_firstlineno) + + def work(self, function: object, receiver: type | None) -> int: + """The work of one call of ``function``, as a method of ``receiver`` when that is a class (so that ``self.name`` resolves through its MRO).""" + if isinstance(function, MethodType): + # A bound method runs on whatever it is bound to: a class for a classmethod, an instance's class otherwise. + receiver = function.__self__ if isinstance(function.__self__, type) else type(function.__self__) + function = function.__func__ + if not isinstance(function, FunctionType): + return 0 + function = inspect.unwrap(function) + if not isinstance(function, FunctionType): + return 0 + key = (function, receiver) + if key in self.memo: + return self.memo[key] + if key in self.active: + # A recursive helper adds its own work once, not once per level. + return 0 + node = self.definition(function) + if node is None: + return 0 + self.active.add(key) + parameters = [argument.arg for argument in node.args.posonlyargs + node.args.args] + mentions = {name.id for name in ast.walk(node) if isinstance(name, ast.Name)} | {text.value for text in ast.walk(node) if isinstance(text, ast.Constant) and isinstance(text.value, str)} + frame = Frame( + function=function, + receiver=receiver, + receiver_name=parameters[0] if receiver is not None and parameters else None, + runs_playbooks=any("ANSIBLE_PLAYBOOK" in mention or "ansible-playbook" in mention for mention in mentions), + assignments={target.id: assignment.value for assignment in ast.walk(node) if isinstance(assignment, ast.Assign) for target in assignment.targets if isinstance(target, ast.Name)}, + ) + total = self.statements(node.body, frame) + self.active.discard(key) + self.memo[key] = total + return total + + def statements(self, nodes: list[ast.stmt], frame: Frame) -> int: + return sum(self.node(node, frame) for node in nodes) + + def node(self, node: ast.AST, frame: Frame) -> int: + # A nested definition starts nothing until it is called, and calls to it do not resolve. + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef, ast.Lambda, ast.ClassDef)): + return 0 + if isinstance(node, ast.If): + return self.node(node.test, frame) + max(self.statements(node.body, frame), self.statements(node.orelse, frame)) + total = sum(self.node(child, frame) for child in ast.iter_child_nodes(node)) + if isinstance(node, ast.Call): + total += self.call(node, frame) + return total + + def call(self, node: ast.Call, frame: Frame) -> int: + """What the call itself adds: the work of a launch, or of the function it runs.""" + callee = node.func + if isinstance(callee, ast.Attribute) and isinstance(callee.value, ast.Call) and isinstance(callee.value.func, ast.Name) and callee.value.func.id == "super" and frame.receiver is not None: + return self.work_after(frame, callee.attr) + target = resolve(callee, frame.function.__globals__, frame.receiver, frame.receiver_name) + if any(target is launcher for launcher in LAUNCHERS): + return TASKS_PER_PROCESS_START + (self.playbook_tasks(node, frame) if frame.runs_playbooks else 0) + return self.work(target, frame.receiver) + + def playbook_tasks(self, launch: ast.Call, frame: Frame) -> int: + """The tasks of the playbook a launch runs: the first element of its argument list that is the path of a YAML file, or none when no element is.""" + arguments = launch.args[0] if launch.args else None + if isinstance(arguments, ast.Name): + arguments = frame.assignments.get(arguments.id) + if not isinstance(arguments, ast.List): + return 0 + for element in arguments.elts: + path = path_of(element, frame.function.__globals__) + if path is not None and path.suffix == ".yml" and path.is_file(): + return self.playbooks.playbook(path) + return 0 + + def work_after(self, frame: Frame, name: str) -> int: + """The work of ``super().name()`` called from the frame's function: the next definition of ``name`` in the receiver's MRO after the class that defines the function.""" + assert frame.receiver is not None + owner = frame.function.__globals__.get(frame.function.__qualname__.split(".")[0]) + mro = frame.receiver.__mro__ + if owner not in mro: + return 0 + for cls in mro[mro.index(owner) + 1 :]: + if name in vars(cls): + inherited = vars(cls)[name] + return self.work(inherited.__func__ if isinstance(inherited, (classmethod, staticmethod)) else inherited, frame.receiver) + return 0 + +def flatten(suite: unittest.TestSuite) -> Iterator[unittest.TestCase]: + """The test cases of a suite, whatever its nesting.""" + for item in suite: + if isinstance(item, unittest.TestSuite): + yield from flatten(item) + else: + yield item + + +def skipped(candidate: object) -> bool: + return bool(getattr(candidate, "__unittest_skip__", False)) + + +def module_work(counter: LaunchCounter, module: str) -> int: + """The work of one importable test module: the start of the process that runs it, plus the work of each test it would run.""" + loader = unittest.TestLoader() + suite = loader.loadTestsFromName(module) + if loader.errors: + raise RuntimeError(f"{module} cannot be loaded:\n" + "\n".join(str(error) for error in loader.errors)) + total = TASKS_PER_PROCESS_START + classes: set[type] = set() + for case in flatten(suite): + cls = type(case) + method = getattr(cls, case._testMethodName) + if skipped(cls) or skipped(method): + continue + total += sum(counter.work(step, cls) for step in (cls.setUp, method, cls.tearDown)) + if cls not in classes: + classes.add(cls) + total += counter.work(cls.setUpClass, cls) + return total + + +def deal(work: Mapping[str, int], shards: int) -> list[list[str]]: + """Assign every module to one of ``shards`` shards, the heaviest first to the shard with the least work so far. + + Modules of equal work are taken in name order and ties between shards go to the lower index, so the same modules and work always give the same shards. Each shard lists its modules in name order. + """ + if shards < 1: + raise ValueError(f"at least one shard is needed, not {shards}") + loads = [0] * shards + dealt: list[list[str]] = [[] for _ in range(shards)] + for name, amount in sorted(work.items(), key=lambda item: (-item[1], item[0])): + index = min(range(shards), key=lambda shard: (loads[shard], shard)) + loads[index] += amount + dealt[index].append(name) + return [sorted(names) for names in dealt] + + +def modules_of(directory: Path) -> list[str]: + """The names of the test modules directly in a directory.""" + return sorted(path.stem for path in directory.glob("test_*.py")) + + +def work_of(directory: Path) -> dict[str, int]: + """The work of every test module directly in a directory, importing them the way unittest discovery does from it.""" + directory = directory.resolve() + if str(directory) not in sys.path: + sys.path.insert(0, str(directory)) + counter = LaunchCounter(directory, REPO_ROOT) + return {module: module_work(counter, module) for module in modules_of(directory)} + + +def main(arguments: list[str]) -> int: + if len(arguments) not in (2, 3): + print(__doc__, file=sys.stderr) + return 2 + shard, shards = int(arguments[0]), int(arguments[1]) + directory = Path(arguments[2]) if len(arguments) == 3 else DEFAULT_DIRECTORY + if not 0 <= shard < shards: + print(f"shard {shard} is not one of the {shards} shards", file=sys.stderr) + return 2 + modules = deal(work_of(directory), shards)[shard] + if not modules: + print(f"shard {shard} of {shards} has no modules: there are fewer modules than shards", file=sys.stderr) + return 1 + for module in modules: + print(directory / f"{module}.py") + return 0 + + +if __name__ == "__main__": + sys.exit(main(sys.argv[1:])) diff --git a/tests/unit/test_shard_unit_tests.py b/tests/unit/test_shard_unit_tests.py new file mode 100644 index 00000000..c4c8d8e2 --- /dev/null +++ b/tests/unit/test_shard_unit_tests.py @@ -0,0 +1,373 @@ +"""Tests for scripts/ci/shard_unit_tests.py: the work counted from a module's tests follows the processes they start, and the dealing puts every module in exactly one shard, the same way every time, with the heaviest shard within a bound of the mean.""" + +from __future__ import annotations + +import contextlib +import importlib +import importlib.util +import io +import itertools +import random +import shutil +import sys +import tempfile +import textwrap +import unittest +from pathlib import Path +from types import ModuleType + +REPO_ROOT = Path(__file__).resolve().parents[2] +SCRIPT = REPO_ROOT / "scripts" / "ci" / "shard_unit_tests.py" +UNIT_TESTS = REPO_ROOT / "tests" / "unit" + + +def load_script() -> ModuleType: + spec = importlib.util.spec_from_file_location("shard_unit_tests", SCRIPT) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + sys.modules[spec.name] = module + spec.loader.exec_module(module) + return module + + +shard = load_script() +START = shard.TASKS_PER_PROCESS_START +NAMES = itertools.count() + + +def heaviest_bound(work: dict[str, int], shards: int) -> float: + """Graham's bound on the heaviest of ``shards`` shards when each module goes to the lightest shard so far: the mean plus all but a shards-th part of the heaviest module.""" + return sum(work.values()) / shards + (1 - 1 / shards) * max(work.values()) + + +def loads(work: dict[str, int], dealt: list[list[str]]) -> list[int]: + return [sum(work[name] for name in names) for names in dealt] + + +def modules_went_to_the_lightest_shard(work: dict[str, int], dealt: list[list[str]]) -> list[str]: + """The shards that held more without their smallest module than another shard holds in all, which a module dealt to the lightest shard so far cannot leave behind: that shard was the lightest when the module went to it.""" + totals = loads(work, dealt) + return [ + f"shard {index}" + for index, names in enumerate(dealt) + if names and totals[index] - min(work[name] for name in names) > min(total for other, total in enumerate(totals) if other != index) + ] + + +class DealTest(unittest.TestCase): + SKEWED = {"heavy": 1000, **{f"light_{index:02d}": 1 + index % 7 for index in range(40)}, "middling_a": 300, "middling_b": 290, "middling_c": 40} + + def test_every_module_is_in_exactly_one_shard(self) -> None: + for shards in range(1, 9): + dealt = shard.deal(self.SKEWED, shards) + self.assertEqual(len(dealt), shards) + self.assertEqual(sorted(itertools.chain.from_iterable(dealt)), sorted(self.SKEWED)) + + def test_the_assignment_does_not_depend_on_the_order_the_modules_are_listed_in(self) -> None: + expected = shard.deal(self.SKEWED, 5) + names = list(self.SKEWED) + generator = random.Random(len(names)) + for _ in range(len(names)): + generator.shuffle(names) + self.assertEqual(shard.deal({name: self.SKEWED[name] for name in names}, 5), expected) + + def test_the_heaviest_shard_is_within_the_list_scheduling_bound_of_the_mean(self) -> None: + for shards in range(1, 9): + dealt = shard.deal(self.SKEWED, shards) + self.assertLessEqual(max(loads(self.SKEWED, dealt)), heaviest_bound(self.SKEWED, shards), f"{shards} shards") + + def test_every_module_went_to_the_shard_that_had_the_least_work_when_it_was_dealt(self) -> None: + for shards in range(2, 9): + self.assertEqual(modules_went_to_the_lightest_shard(self.SKEWED, shard.deal(self.SKEWED, shards)), [], f"{shards} shards") + + def test_equal_modules_are_dealt_in_name_order_to_the_lower_shard_first(self) -> None: + work = {name: 5 for name in ("test_d", "test_a", "test_c", "test_b", "test_e")} + self.assertEqual(shard.deal(work, 2), [["test_a", "test_c", "test_e"], ["test_b", "test_d"]]) + + def test_a_heavy_module_gets_a_shard_to_itself_while_the_lighter_ones_share_the_rest(self) -> None: + work = {"heavy": 100, **{f"light_{index}": 10 for index in range(10)}} + dealt = shard.deal(work, 2) + self.assertEqual(sorted(loads(work, dealt)), [100, 100]) + self.assertIn(["heavy"], dealt) + + def test_a_shard_count_below_one_is_refused(self) -> None: + with self.assertRaises(ValueError): + shard.deal(self.SKEWED, 0) + + +class RepositoryDealTest(unittest.TestCase): + """The dealing of this repository's own test modules.""" + + work: dict[str, int] + + @classmethod + def setUpClass(cls) -> None: + cls.work = shard.work_of(UNIT_TESTS) + + def test_the_work_covers_every_test_module_of_the_directory(self) -> None: + self.assertEqual(sorted(self.work), sorted(path.stem for path in UNIT_TESTS.glob("test_*.py"))) + + def test_every_module_does_at_least_the_work_of_the_process_that_runs_it(self) -> None: + self.assertEqual([name for name, amount in self.work.items() if amount < START], []) + + def test_every_module_is_in_exactly_one_shard_whatever_the_number_of_shards(self) -> None: + for shards in range(1, 9): + dealt = shard.deal(self.work, shards) + self.assertEqual(sorted(itertools.chain.from_iterable(dealt)), sorted(self.work), f"{shards} shards") + + def test_the_same_tree_is_dealt_the_same_way_again(self) -> None: + self.assertEqual(shard.deal(shard.work_of(UNIT_TESTS), 5), shard.deal(self.work, 5)) + + def test_the_heaviest_shard_is_within_the_list_scheduling_bound_of_the_mean(self) -> None: + for shards in range(1, 9): + dealt = shard.deal(self.work, shards) + self.assertLessEqual(max(loads(self.work, dealt)), heaviest_bound(self.work, shards), f"{shards} shards") + + def test_every_module_went_to_the_shard_that_had_the_least_work_when_it_was_dealt(self) -> None: + for shards in range(2, 9): + self.assertEqual(modules_went_to_the_lightest_shard(self.work, shard.deal(self.work, shards)), [], f"{shards} shards") + + def test_the_shards_work_differs_by_no_more_than_the_heaviest_module(self) -> None: + # The shard that ends heaviest took its last module while it was the lightest, so it exceeds the lightest by less than that module. + for shards in range(2, 9): + totals = loads(self.work, shard.deal(self.work, shards)) + self.assertLessEqual(max(totals) - min(totals), max(self.work.values()), f"{shards} shards") + + +class WorkTest(unittest.TestCase): + """The work counted from synthetic modules, each written into a repository of its own.""" + + def setUp(self) -> None: + self.repository = Path(tempfile.mkdtemp()).resolve() + self.addCleanup(shutil.rmtree, self.repository) + self.directory = self.repository / "tests" / "unit" + self.directory.mkdir(parents=True) + self.written: list[str] = [] + sys.path.insert(0, str(self.directory)) + self.addCleanup(sys.path.remove, str(self.directory)) + self.addCleanup(self.forget) + + def forget(self) -> None: + for name in self.written: + sys.modules.pop(name, None) + + def write(self, source: str, name: str | None = None) -> str: + name = name or f"synthetic_{next(NAMES)}" + (self.directory / f"{name}.py").write_text(textwrap.dedent(source)) + self.written.append(name) + importlib.invalidate_caches() + return name + + def work(self, module: str) -> int: + return shard.module_work(shard.LaunchCounter(self.directory, self.repository), module) + + def test_a_module_that_starts_nothing_costs_the_process_that_runs_it(self) -> None: + module = self.write( + """ + import unittest + + class Quiet(unittest.TestCase): + def test_one(self): self.assertTrue(True) + def test_two(self): self.assertTrue(True) + """ + ) + self.assertEqual(self.work(module), START) + + def test_each_process_a_test_starts_costs_a_start_whichever_function_of_the_subprocess_module_starts_it(self) -> None: + module = self.write( + """ + import subprocess + import unittest + from subprocess import run + + class Starts(unittest.TestCase): + def test_run(self): subprocess.run(["true"]) + def test_imported_name(self): run(["true"]) + def test_popen(self): subprocess.Popen(["true"]).wait() + def test_output(self): subprocess.check_output(["true"]) + def test_check(self): subprocess.check_call(["true"]) + def test_call(self): subprocess.call(["true"]) + """ + ) + self.assertEqual(self.work(module), START * (1 + 6)) + + def test_launches_through_helpers_base_classes_super_and_the_set_up_methods_are_reached(self) -> None: + support = self.write( + """ + import subprocess + import unittest + + def start(): + subprocess.run(["true"]) + + class Base(unittest.TestCase): + @classmethod + def setUpClass(cls): + start() + + def setUp(self): + start() + + def tearDown(self): + start() + + def helper(self): + self.inner() + + def inner(self): + start() + """ + ) + module = self.write( + f""" + import unittest + + import {support} + from {support} import Base, start + + class Child({support}.Base): + def setUp(self): + super().setUp() + start() + + def test_a(self): + self.helper() + start() + + def test_b(self): + {support}.start() + + class Sibling(Base): + def test_c(self): pass + """ + ) + # The module's own process; Child: setUpClass once; each of its two tests: both set-ups (Base's through super()), tearDown, and the body (test_a has two launches, test_b one); Sibling: setUpClass once, and for its one test setUp, tearDown and no body. The imported Base has no tests of its own beyond the TestCase machinery. + child = 1 + 2 * (2 + 1) + (2 + 1) + sibling = 1 + 2 + self.assertEqual(self.work(module), START * (1 + child + sibling)) + + def test_a_launch_site_counts_once_however_many_times_a_loop_runs_it_and_a_conditional_counts_its_dearer_branch(self) -> None: + module = self.write( + """ + import subprocess + import unittest + + class Shapes(unittest.TestCase): + def test_loop(self): + for _ in range(50): + subprocess.run(["true"]) + + def test_branches(self): + if self.id(): + subprocess.run(["true"]) + else: + subprocess.run(["true"]) + subprocess.run(["true"]) + """ + ) + self.assertEqual(self.work(module), START * (1 + 1 + 2)) + + def test_a_skipped_test_or_class_does_no_work_and_a_recursive_helper_counts_its_launch_once(self) -> None: + module = self.write( + """ + import subprocess + import unittest + + def again(depth): + subprocess.run(["true"]) + if depth: + again(depth - 1) + + class Partly(unittest.TestCase): + @unittest.skip("not here") + def test_skipped(self): subprocess.run(["true"]) + + def test_recursive(self): again(3) + + @unittest.skip("not here") + class Whole(unittest.TestCase): + def setUp(self): subprocess.run(["true"]) + def test_a(self): subprocess.run(["true"]) + """ + ) + self.assertEqual(self.work(module), START * (1 + 1)) + + def test_a_playbook_launch_costs_the_tasks_the_playbook_and_the_roles_it_includes_reach(self) -> None: + roles = self.repository / "roles" / "demo" / "tasks" + roles.mkdir(parents=True) + (roles / "main.yml").write_text("- name: a\n ansible.builtin.debug: {}\n- name: b\n block:\n - name: c\n ansible.builtin.debug: {}\n") + (roles / "other.yml").write_text("- name: d\n ansible.builtin.debug: {}\n- name: e\n ansible.builtin.debug: {}\n") + # Tasks of main.yml: a, b and c (inside b's block), 3; of other.yml, 2; of the role, 5, in two files. + for name, include, reached in ( + ("named", "name: exadev.demo.demo\n tasks_from: other", 2), + ("variable", "name: exadev.demo.demo\n tasks_from: \"{{ item }}\"", 3), + ("whole", "name: exadev.demo.demo", 5), + ): + playbook = self.repository / "tests" / f"{name}.yml" + playbook.write_text( + "- hosts: all\n tasks:\n - name: own\n ansible.builtin.debug: {}\n" + f" - name: include\n ansible.builtin.include_role:\n {include}\n" + ) + module = self.write( + f""" + import subprocess + import unittest + from pathlib import Path + + ANSIBLE_PLAYBOOK = "ansible-playbook" + PLAY = Path(__file__).resolve().parents[1] / "{name}.yml" + + class Runs(unittest.TestCase): + def test_it(self): + subprocess.run([ANSIBLE_PLAYBOOK, "-i", "inventory", str(PLAY)]) + """ + ) + # The play's own two tasks, those the include reaches, and the start of the process; the module's own process. + self.assertEqual(self.work(module), START * 2 + 2 + reached, name) + + def test_a_process_that_is_not_a_playbook_run_costs_its_start_alone(self) -> None: + playbook = self.repository / "tests" / "unrelated.yml" + playbook.write_text("- hosts: all\n tasks:\n - name: own\n ansible.builtin.debug: {}\n") + module = self.write( + """ + import subprocess + import unittest + from pathlib import Path + + PLAY = Path(__file__).resolve().parents[1] / "unrelated.yml" + + class Runs(unittest.TestCase): + def test_it(self): + subprocess.run(["cat", str(PLAY)]) + """ + ) + self.assertEqual(self.work(module), START * 2) + + def test_a_module_that_cannot_be_imported_is_an_error_not_a_module_of_no_work(self) -> None: + module = self.write("import a_module_that_does_not_exist\n") + with self.assertRaises(RuntimeError): + self.work(module) + + +class CommandLineTest(unittest.TestCase): + def test_a_shard_prints_the_paths_of_its_modules_and_the_shards_between_them_print_every_module(self) -> None: + directory = Path(tempfile.mkdtemp()) + self.addCleanup(shutil.rmtree, directory) + for name in ("test_one", "test_two", "test_three"): + (directory / f"{name}.py").write_text("import unittest\n\nclass T(unittest.TestCase):\n def test_it(self): pass\n") + self.addCleanup(lambda: [sys.modules.pop(name, None) for name in ("test_one", "test_two", "test_three")]) + self.addCleanup(lambda: sys.path.remove(str(directory.resolve()))) + printed: list[str] = [] + for index in range(2): + with contextlib.redirect_stdout(io.StringIO()) as out: + status = shard.main([str(index), "2", str(directory)]) + self.assertEqual(status, 0) + printed.extend(out.getvalue().split()) + self.assertEqual(sorted(printed), sorted(str(directory / f"{name}.py") for name in ("test_one", "test_two", "test_three"))) + + def test_a_shard_index_outside_the_shards_is_refused(self) -> None: + self.assertEqual(shard.main(["5", "5"]), 2) + + +if __name__ == "__main__": + unittest.main()