diff --git a/tests/run_tests.py b/tests/run_tests.py new file mode 100644 index 0000000..c6ec1f0 --- /dev/null +++ b/tests/run_tests.py @@ -0,0 +1,233 @@ +"""Select repository tests from changed paths or named subsystem groups.""" + +# ruff: noqa: T201 + +from __future__ import annotations + +import argparse +import fnmatch +import os +import shutil +import subprocess +import sys +import tomllib +import uuid +from dataclasses import dataclass, field +from pathlib import Path + + +REPOSITORY_ROOT = Path(__file__).resolve().parents[1] +COMFYUI_ROOT = REPOSITORY_ROOT.parent.parent +MANIFEST_PATH = Path(__file__).with_name("test_groups.toml") +TEMP_ROOT = REPOSITORY_ROOT / ".pytest-tmp" + + +@dataclass(frozen=True) +class TestGroup: + name: str + paths: tuple[str, ...] + python_tests: tuple[str, ...] + + +@dataclass +class Selection: + groups: set[str] = field(default_factory=set) + python_tests: set[str] = field(default_factory=set) + reasons: dict[str, set[str]] = field(default_factory=dict) + unmapped: set[str] = field(default_factory=set) + + +def _parse_groups(text: str) -> dict[str, TestGroup]: + data = tomllib.loads(text) + return { + name: TestGroup( + name=name, + paths=tuple(values.get("paths", ())), + python_tests=tuple(values.get("python_tests", ())), + ) + for name, values in data.get("groups", {}).items() + } + + +def load_groups(path: Path = MANIFEST_PATH) -> dict[str, TestGroup]: + return _parse_groups(path.read_text(encoding="utf-8")) + + +def load_groups_from_revision(revision: str) -> dict[str, TestGroup]: + manifest = MANIFEST_PATH.relative_to(REPOSITORY_ROOT).as_posix() + result = subprocess.run( + ["git", "show", f"{revision}:{manifest}"], cwd=REPOSITORY_ROOT, + check=False, capture_output=True, text=True, + ) + return _parse_groups(result.stdout) if result.returncode == 0 else {} + + +def git_lines(*args: str) -> set[str]: + result = subprocess.run( + ["git", *args], cwd=REPOSITORY_ROOT, check=True, + capture_output=True, text=True, + ) + return { + line.strip().replace("\\", "/") + for line in result.stdout.splitlines() if line.strip() + } + + +def changed_paths(base: str | None = None) -> set[str]: + paths = git_lines("diff", "--name-only", "--relative", "HEAD") + if base: + paths.update(git_lines("diff", "--name-only", "--relative", f"{base}...HEAD")) + return paths + + +def is_production_source(path: str) -> bool: + normalized = path.replace("\\", "/") + return normalized == "__init__.py" or ( + normalized.startswith("nodes/") and normalized.endswith(".py") + ) + + +def _add_group(selection: Selection, group: TestGroup, reason: str) -> None: + selection.groups.add(group.name) + selection.python_tests.update(group.python_tests) + selection.reasons.setdefault(group.name, set()).add(reason) + + +def _path_matches_group(path: str, group: TestGroup) -> bool: + return any(fnmatch.fnmatchcase(path, pattern) for pattern in group.paths) + + +def _existing_group(group: TestGroup) -> TestGroup: + return TestGroup( + name=group.name, paths=group.paths, + python_tests=tuple( + path for path in group.python_tests if (REPOSITORY_ROOT / path).is_file() + ), + ) + + +def select_tests( + paths: set[str], groups: dict[str, TestGroup], + explicit_groups: tuple[str, ...] = (), + historical_groups: tuple[dict[str, TestGroup], ...] = (), +) -> Selection: + selection = Selection() + for name in explicit_groups: + if name not in groups: + raise ValueError(f"Unknown test group: {name}") + _add_group(selection, groups[name], "explicit selection") + + for raw_path in sorted(paths): + path = raw_path.replace("\\", "/") + if path.startswith("tests/test_") and path.endswith(".py"): + if (REPOSITORY_ROOT / path).is_file(): + selection.python_tests.add(path) + selection.reasons.setdefault("direct test", set()).add(path) + continue + + matched = False + for group in groups.values(): + if _path_matches_group(path, group): + _add_group(selection, group, path) + matched = True + if not matched and not (REPOSITORY_ROOT / path).exists(): + for historical in historical_groups: + for historical_group in historical.values(): + if _path_matches_group(path, historical_group): + active = groups.get(historical_group.name) + _add_group( + selection, active or _existing_group(historical_group), + f"{path} (deleted)", + ) + matched = True + if not matched and is_production_source(path) and (REPOSITORY_ROOT / path).exists(): + selection.unmapped.add(path) + return selection + + +def tracked_final_tests() -> set[str]: + tracked = git_lines("ls-files", "tests/test_*.py") + return {path for path in tracked if (REPOSITORY_ROOT / path).is_file()} + + +def print_selection(selection: Selection) -> None: + for group in sorted(selection.groups): + reasons = ", ".join(sorted(selection.reasons.get(group, ()))) + print(f"group {group}: {reasons}") + for reason in sorted(selection.reasons.get("direct test", ())): + print(f"direct test: {reason}") + print("python tests:") + for path in sorted(selection.python_tests): + print(f" {path}") + + +def run_selection(selection: Selection) -> int: + TEMP_ROOT.mkdir(exist_ok=True) + basetemp = (TEMP_ROOT / f"run-{uuid.uuid4().hex}").resolve() + if TEMP_ROOT.resolve() not in basetemp.parents: + raise RuntimeError(f"Unsafe pytest temporary path: {basetemp}") + try: + environment = os.environ.copy() + environment["PYTHONPATH"] = os.pathsep.join( + filter(None, (str(REPOSITORY_ROOT), environment.get("PYTHONPATH", ""))) + ) + result = subprocess.run( + [sys.executable, "-m", "pytest", "-q", "--import-mode=importlib", + "--basetemp", str(basetemp), + *[str((REPOSITORY_ROOT / path).resolve()) + for path in sorted(selection.python_tests)]], + cwd=COMFYUI_ROOT, env=environment, check=False, + ) + return result.returncode + finally: + if basetemp.exists(): + shutil.rmtree(basetemp) + if TEMP_ROOT.exists() and not any(TEMP_ROOT.iterdir()): + TEMP_ROOT.rmdir() + + +def parser() -> argparse.ArgumentParser: + result = argparse.ArgumentParser(description=__doc__) + mode = result.add_mutually_exclusive_group() + mode.add_argument("--final", action="store_true", help="run every tracked test") + mode.add_argument("--list-groups", action="store_true", help="list configured groups") + result.add_argument("--changed", action="store_true", help="select changed paths (default)") + result.add_argument("--base", help="include committed changes since BASE") + result.add_argument("--group", action="append", default=[], help="add a named group") + result.add_argument("--dry-run", action="store_true", help="explain without executing") + return result + + +def main(argv: list[str] | None = None) -> int: + args = parser().parse_args(argv) + groups = load_groups() + if args.list_groups: + for name in sorted(groups): + print(name) + return 0 + if args.final: + selection = Selection( + groups={"final"}, python_tests=tracked_final_tests(), + reasons={"final": {"every tracked test"}}, + ) + else: + paths = changed_paths(args.base) if args.changed or args.base or not args.group else set() + revisions = ["HEAD", *([args.base] if args.base else [])] + historical = tuple(load_groups_from_revision(rev) for rev in revisions) if paths else () + selection = select_tests(paths, groups, tuple(args.group), historical) + + if selection.unmapped: + print("Unmapped production source files:", file=sys.stderr) + for path in sorted(selection.unmapped): + print(f" {path}", file=sys.stderr) + print("Update tests/test_groups.toml or use --final.", file=sys.stderr) + return 2 + print_selection(selection) + if not selection.python_tests: + print("No tests selected.") + return 0 + return 0 if args.dry_run else run_selection(selection) + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_groups.toml b/tests/test_groups.toml new file mode 100644 index 0000000..01f762e --- /dev/null +++ b/tests/test_groups.toml @@ -0,0 +1,53 @@ +[groups.repository_contract] +paths = ["__init__.py", "nodes/*.py"] +python_tests = [ + "tests/test_schema_tooltips.py", + "tests/test_uel_repository_contract.py", +] + +[groups.artifact_paths] +paths = [ + "nodes/artifact_paths.py", "nodes/dora_extract_wd.py", + "nodes/dora_learned_wd.py", "nodes/dtype_conversion.py", + "nodes/lora_extract_svd.py", "nodes/lora_merger.py", + "nodes/lora_rename.py", "nodes/lora_resize.py", "nodes/merger.py", + "nodes/prunekeys.py", "nodes/renamekeys.py", "nodes/text_encoder_extract.py", +] +python_tests = ["tests/test_artifact_paths.py"] + +[groups.cwb] +paths = ["nodes/consensus_merger.py", "run_cwb_api_workflow.py", "run_cwb_preset_sweep.py"] +python_tests = [ + "tests/test_consensus_merger.py", "tests/test_cwb_api_runner.py", + "tests/test_cwb_preset_sweep.py", +] + +[groups.model_analysis] +paths = ["nodes/model_analysis.py", "run_model_analysis_api_workflow.py"] +python_tests = ["tests/test_model_analysis.py", "tests/test_model_analysis_api_runner.py"] + +[groups.adaptive_extraction] +paths = [ + "nodes/adaptive_svd.py", "nodes/dora_extract_wd.py", "nodes/dora_learned_wd.py", + "nodes/extraction_stream.py", "nodes/lora_extract_svd.py", + "nodes/text_encoder_extract.py", +] +python_tests = ["tests/test_adaptive_extraction_pipeline.py", "tests/test_knee_partial_svd.py"] + +[groups.lora_merge] +paths = [ + "nodes/device_utils.py", "nodes/lora_alpha.py", "nodes/lora_merger.py", + "nodes/quantization_guard.py", "nodes/uel_io.py", +] +python_tests = [ + "tests/test_lora_diff_dtype_contract.py", "tests/test_lora_merger_async.py", + "tests/test_lora_quantization_guard.py", +] + +[groups.lora_resize] +paths = ["nodes/lora_resize.py"] +python_tests = ["tests/test_lora_resize_factor_space.py", "tests/test_lora_resize_min_rank.py"] + +[groups.dtype_conversion] +paths = ["nodes/dtype_conversion.py"] +python_tests = ["tests/test_dtype_conversion.py"]