|
| 1 | +"""Shared static architecture-test helpers.""" |
| 2 | + |
| 3 | +from __future__ import annotations |
| 4 | + |
| 5 | +import ast |
| 6 | +import re |
| 7 | +import tomllib |
| 8 | +from collections.abc import Iterable |
| 9 | +from dataclasses import dataclass |
| 10 | +from pathlib import Path |
| 11 | + |
| 12 | + |
| 13 | +IGNORED_PATH_PARTS = { |
| 14 | + ".git", |
| 15 | + ".local", |
| 16 | + ".mypy_cache", |
| 17 | + ".pytest_cache", |
| 18 | + ".ruff_cache", |
| 19 | + ".venv", |
| 20 | + "__pycache__", |
| 21 | +} |
| 22 | + |
| 23 | + |
| 24 | +@dataclass(frozen=True) |
| 25 | +class ImportReference: |
| 26 | + """One import statement resolved to its imported module prefix.""" |
| 27 | + |
| 28 | + path: Path |
| 29 | + line_no: int |
| 30 | + module_name: str |
| 31 | + |
| 32 | + |
| 33 | +@dataclass(frozen=True) |
| 34 | +class EffectImportRule: |
| 35 | + """Static rule for keeping an effect import inside its owner modules.""" |
| 36 | + |
| 37 | + name: str |
| 38 | + forbidden_prefixes: tuple[str, ...] |
| 39 | + allowed_module_prefixes: tuple[str, ...] |
| 40 | + |
| 41 | + |
| 42 | +def python_files(root: Path) -> list[Path]: |
| 43 | + """Return Python source files below root, excluding caches and local envs.""" |
| 44 | + |
| 45 | + if root.is_file(): |
| 46 | + return [root] if root.suffix == ".py" and not is_ignored_path(root) else [] |
| 47 | + return sorted( |
| 48 | + path |
| 49 | + for path in root.rglob("*.py") |
| 50 | + if path.is_file() and not is_ignored_path(path) |
| 51 | + ) |
| 52 | + |
| 53 | + |
| 54 | +def production_python_files( |
| 55 | + repo_root: Path, package_roots: Iterable[str] |
| 56 | +) -> list[Path]: |
| 57 | + """Return Python files for production package roots only.""" |
| 58 | + |
| 59 | + files: list[Path] = [] |
| 60 | + for package_root in package_roots: |
| 61 | + files.extend(python_files(repo_root / package_root)) |
| 62 | + return sorted(set(files)) |
| 63 | + |
| 64 | + |
| 65 | +def is_ignored_path(path: Path) -> bool: |
| 66 | + return any(part in IGNORED_PATH_PARTS for part in path.parts) |
| 67 | + |
| 68 | + |
| 69 | +def imported_modules(path: Path) -> list[tuple[int, str]]: |
| 70 | + """Return module prefixes imported by a Python source file.""" |
| 71 | + |
| 72 | + tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) |
| 73 | + imports: list[tuple[int, str]] = [] |
| 74 | + for node in ast.walk(tree): |
| 75 | + if isinstance(node, ast.Import): |
| 76 | + imports.extend((node.lineno, alias.name) for alias in node.names) |
| 77 | + elif isinstance(node, ast.ImportFrom) and node.module: |
| 78 | + imports.append((node.lineno, node.module)) |
| 79 | + return imports |
| 80 | + |
| 81 | + |
| 82 | +def iter_import_references(paths: Iterable[Path]) -> list[ImportReference]: |
| 83 | + references: list[ImportReference] = [] |
| 84 | + for path in paths: |
| 85 | + references.extend( |
| 86 | + ImportReference(path=path, line_no=line_no, module_name=module_name) |
| 87 | + for line_no, module_name in imported_modules(path) |
| 88 | + ) |
| 89 | + return references |
| 90 | + |
| 91 | + |
| 92 | +def module_name_for_path(path: Path, repo_root: Path) -> str: |
| 93 | + """Return the dotted module name for a Python file below the repo root.""" |
| 94 | + |
| 95 | + rel_path = path.relative_to(repo_root).with_suffix("") |
| 96 | + parts = rel_path.parts |
| 97 | + if parts[-1] == "__init__": |
| 98 | + parts = parts[:-1] |
| 99 | + return ".".join(parts) |
| 100 | + |
| 101 | + |
| 102 | +def module_matches_prefix(module_name: str, prefixes: tuple[str, ...]) -> bool: |
| 103 | + return any( |
| 104 | + module_name == prefix or module_name.startswith(f"{prefix}.") |
| 105 | + for prefix in prefixes |
| 106 | + ) |
| 107 | + |
| 108 | + |
| 109 | +def forbidden_import_violations( |
| 110 | + paths: Iterable[Path], |
| 111 | + forbidden_prefixes: tuple[str, ...], |
| 112 | + *, |
| 113 | + repo_root: Path, |
| 114 | +) -> list[str]: |
| 115 | + violations: list[str] = [] |
| 116 | + for ref in iter_import_references(paths): |
| 117 | + if module_matches_prefix(ref.module_name, forbidden_prefixes): |
| 118 | + violations.append( |
| 119 | + f"{ref.path.relative_to(repo_root)}:{ref.line_no} imports {ref.module_name}" |
| 120 | + ) |
| 121 | + return violations |
| 122 | + |
| 123 | + |
| 124 | +def effect_import_violations( |
| 125 | + paths: Iterable[Path], |
| 126 | + rules: Iterable[EffectImportRule], |
| 127 | + *, |
| 128 | + repo_root: Path, |
| 129 | +) -> list[str]: |
| 130 | + violations: list[str] = [] |
| 131 | + for ref in iter_import_references(paths): |
| 132 | + source_module = module_name_for_path(ref.path, repo_root) |
| 133 | + for rule in rules: |
| 134 | + if not module_matches_prefix(ref.module_name, rule.forbidden_prefixes): |
| 135 | + continue |
| 136 | + if module_matches_prefix(source_module, rule.allowed_module_prefixes): |
| 137 | + continue |
| 138 | + violations.append( |
| 139 | + f"{ref.path.relative_to(repo_root)}:{ref.line_no} imports " |
| 140 | + f"{ref.module_name} outside {rule.name} owners" |
| 141 | + ) |
| 142 | + return violations |
| 143 | + |
| 144 | + |
| 145 | +def pyproject_package_roots(pyproject_path: Path) -> set[str]: |
| 146 | + """Return top-level setuptools package roots declared in pyproject.toml.""" |
| 147 | + |
| 148 | + data = tomllib.loads(pyproject_path.read_text(encoding="utf-8")) |
| 149 | + include_patterns = ( |
| 150 | + data.get("tool", {}) |
| 151 | + .get("setuptools", {}) |
| 152 | + .get("packages", {}) |
| 153 | + .get("find", {}) |
| 154 | + .get("include", []) |
| 155 | + ) |
| 156 | + roots: set[str] = set() |
| 157 | + for pattern in include_patterns: |
| 158 | + root = pattern.split("*", 1)[0].rstrip(".") |
| 159 | + if root: |
| 160 | + roots.add(root.split(".", 1)[0]) |
| 161 | + return roots |
| 162 | + |
| 163 | + |
| 164 | +def pyproject_dependency_names(pyproject_path: Path) -> set[str]: |
| 165 | + """Return normalized project dependency names from pyproject.toml.""" |
| 166 | + |
| 167 | + data = tomllib.loads(pyproject_path.read_text(encoding="utf-8")) |
| 168 | + dependencies = data.get("project", {}).get("dependencies", []) |
| 169 | + return {_normalize_dependency_name(dependency) for dependency in dependencies} |
| 170 | + |
| 171 | + |
| 172 | +def direct_package_dirs(package_root: Path) -> set[str]: |
| 173 | + """Return direct child package directory names under a package root.""" |
| 174 | + |
| 175 | + return { |
| 176 | + path.name |
| 177 | + for path in package_root.iterdir() |
| 178 | + if path.is_dir() and not is_ignored_path(path) |
| 179 | + } |
| 180 | + |
| 181 | + |
| 182 | +def text_occurrence_violations( |
| 183 | + paths: Iterable[Path], |
| 184 | + needles: tuple[str, ...], |
| 185 | + *, |
| 186 | + repo_root: Path, |
| 187 | +) -> list[str]: |
| 188 | + violations: list[str] = [] |
| 189 | + for path in paths: |
| 190 | + text = path.read_text(encoding="utf-8", errors="ignore") |
| 191 | + for needle in needles: |
| 192 | + if needle in text: |
| 193 | + violations.append(f"{path.relative_to(repo_root)} contains {needle!r}") |
| 194 | + return violations |
| 195 | + |
| 196 | + |
| 197 | +def _normalize_dependency_name(dependency: str) -> str: |
| 198 | + raw_name = re.split(r"[<>=!~;\\[]", dependency, maxsplit=1)[0] |
| 199 | + return raw_name.strip().lower().replace("_", "-") |
0 commit comments