kevinveenbirkenbach/computer-playbook
1
0 Fork 0
mirror of https://github.com/kevinveenbirkenbach/computer-playbook.git synced 2026-11-06 10:59:59 +00:00
Code Issues Packages Projects Releases Wiki Activity

test(lint): cap production Python files at 500 lines

Browse Source 
Adds a SPOT-aware lint that walks the repo via the cached
project-tree iterator (no `git` subprocess) and fails on any tracked
non-test Python file longer than 500 lines. Files under `tests/` are
intentionally excluded — coverage breadth there reflects feature
matrix, not design weight. A per-file `# nocheck` (or `#nocheck`)
marker in the first 30 lines opts the file out; the marker is
intentionally visible so the cost of carrying long modules is not
silently buried.

Gitignore-aware filtering moves to its own SPOT
-----------------------------------------------

The fnmatch-based `.gitignore` parser/matcher previously lived inline
in `tests/lint/repository/test_test_files_location.py` (and in
near-duplicates across other tests). Pull it into
`utils.cache.gitignore` as a single source of truth:

* `load_gitignore_patterns(root)` — cached pattern loader.
* `is_path_gitignored(rel_path, patterns)` — pure matcher.
* `iter_non_ignored_files(...)` — convenience iterator in
  `utils.cache.files` that composes the cached project walk with the
  matcher.

`utils.cache._reset_cache_for_tests` now resets the gitignore module
too, so test fixtures that mutate the tree see consistent state.

Refactor / opt-out for existing offenders
-----------------------------------------

Three production files exceeded the cap; addressed each:

* `roles/sys-svc-compose-ca/files/compose_ca.py` — opted out via
  `# nocheck: file-size`. The script is delivered to target hosts
  via `ansible.builtin.copy` and runs with only the stdlib; splitting
  would require also delivering a package, which the role does not
  do.
* `tasks/utils/migrate_meta_layout.py` — converted to a package
  (`tasks/utils/migrate_meta_layout/`) with focused submodules:
    - `__main__` — entrypoint + run-orchestration.
    - `yaml_io` — load/dump/empty-dir/deep-merge helpers.
    - `credentials` — schema-vs-config collision detection +
      runtime-to-schema conversion.
    - `role_builder` — per-role meta file emission.
    - `indices` — networks + ports + relays index builders.
  Run via `python3 -m tasks.utils.migrate_meta_layout`. Largest
  submodule is `role_builder.py` at 250 lines.
* `cli/fix/move_unnecessary_dependencies/__main__.py` — split
  alongside two new sibling modules in the same package:
    - `analysis` — provider/consumer static analysis.
    - `apply` — meta rewrite + guarded include-block prepending.
    - `yaml_io` — ruamel round-trip helpers + role-tree discovery.
  Largest submodule is `analysis.py` at 174 lines.

`make test` green (162 tests, 1 skipped); the new lint passes.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
kevinveenbirkenbachandClaude Opus 4.7 committed 2026-04-29 17:15:13 +02:00
1 parent fe54854fe5
commit 4b37e123d3
18 files changed
+1317 -1142
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

cli/fix/move_unnecessary_dependencies/__init__.py
+7 -14
View File
@@ -1,17 +1,10 @@
"""Compatibility wrapper.
"""Move unnecessary meta dependencies to guarded include_role/import_role.
This package was migrated from a flat module (move_unnecessary_dependencies.py) to a package layout:
move_unnecessary_dependencies/__main__.py contains the original implementation.
Run via ``python3 -m cli.fix.move_unnecessary_dependencies``. The
implementation is split across:
We re-export the public API so existing imports keep working.
* ``__main__`` — argparse + per-role orchestration.
* ``analysis`` — provider / consumer static analysis.
* ``apply`` — YAML rewrite + include-block prepending.
* ``yaml_io`` — ruamel round-trip helpers + role-tree discovery.
"""
from __future__ import annotations
from . import __main__ as _main
__all__ = getattr(_main, "__all__", [n for n in dir(_main) if not n.startswith("_")])
globals().update({name: getattr(_main, name) for name in __all__})
# Prefer explicit __all__ if the original module defined it.
__all__ = getattr(_main, "__all__", [n for n in dir(_main) if not n.startswith("_")])
cli/fix/move_unnecessary_dependencies/__main__.py
+72 -453
View File
@@ -1,434 +1,66 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Move unnecessary meta dependencies to guarded include_role/import_role.
A dependency is considered unnecessary when the consumer:
* does NOT use provider variables in defaults/vars/handlers (no
early-var need), AND
* in tasks, any usage of provider vars or notifications of
provider handlers happens only AFTER an include/import of the
provider in the same file (or never).
Action per qualifying role:
* Remove the entry from ``roles/<role>/meta/main.yml``.
* Prepend a guarded include block to
``roles/<role>/tasks/01_core.yml`` (preferred) or
``roles/<role>/tasks/main.yml`` if the former is absent.
* When more than one dependency is moved, the include block uses a
``loop:``.
The heuristic mirrors
``tests/integration/test_unnecessary_role_dependencies.py``. YAML
round-trip preservation (comments, quotes, key order) is delegated
to ``ruamel.yaml`` via :mod:`.yaml_io`. Modified files get a
``.bak`` sibling.
"""
Move unnecessary meta dependencies to guarded include_role/import_role
for better performance, while preserving YAML comments, quotes, and layout.
Heuristic (matches tests/integration/test_unnecessary_role_dependencies.py):
- A dependency is considered UNNECESSARY if:
* The consumer does NOT use provider variables in defaults/vars/handlers
(no early-var need), AND
* In tasks, any usage of provider vars or provider-handler notifications
occurs only AFTER an include/import of the provider in the same file,
OR there is no usage at all.
Action:
- Remove such dependencies from roles/<role>/meta/main.yml.
- Prepend a guarded include block to roles/<role>/tasks/01_core.yml (preferred)
or roles/<role>/tasks/main.yml if 01_core.yml is absent.
- If multiple dependencies are moved for a role, use a loop over include_role.
Notes:
- Creates .bak backups for modified YAML files.
- Requires ruamel.yaml to preserve comments/quotes everywhere.
"""
from __future__ import annotations
import argparse
import glob
import os
import re
import shutil
import sys
from typing import Dict, Set, List, Tuple, Optional
import logging
from typing import Dict, List, Optional, Set, Tuple
# --- Require ruamel.yaml for full round-trip preservation ---
try:
from ruamel.yaml import YAML
from ruamel.yaml.comments import CommentedMap, CommentedSeq
from ruamel.yaml.scalarstring import SingleQuotedScalarString
_HAVE_RUAMEL = True
except Exception:
_HAVE_RUAMEL = False
if not _HAVE_RUAMEL:
logging.error(
"ruamel.yaml is required to preserve comments/quotes. Install with: pip install ruamel.yaml"
)
sys.exit(3)
yaml_rt = YAML()
yaml_rt.preserve_quotes = True
yaml_rt.width = 10**9 # prevent line wrapping
# ---------------- Utilities ----------------
from .analysis import (
collect_role_defined_vars,
collect_role_handler_names,
dependency_is_unnecessary,
parse_meta_dependencies,
)
from .apply import (
build_include_block_yaml,
prepend_tasks,
update_meta_remove_deps,
)
from .yaml_io import (
iter_role_dirs,
path_if_exists,
role_name_from_dir,
roles_root,
)
def _backup(path: str):
if os.path.exists(path):
shutil.copy2(path, path + ".bak")
def read_text(path: str) -> str:
try:
with open(path, "r", encoding="utf-8") as f:
return f.read()
except Exception:
return ""
def load_yaml_rt(path: str):
try:
with open(path, "r", encoding="utf-8") as f:
data = yaml_rt.load(f)
return data if data is not None else CommentedMap()
except FileNotFoundError:
return CommentedMap()
except Exception as e:
print(f"[WARN] Failed to parse YAML: {path}: {e}", file=sys.stderr)
return CommentedMap()
def dump_yaml_rt(data, path: str):
_backup(path)
with open(path, "w", encoding="utf-8") as f:
yaml_rt.dump(data, f)
def roles_root(project_root: str) -> str:
return os.path.join(project_root, "roles")
def iter_role_dirs(project_root: str) -> List[str]:
root = roles_root(project_root)
return [d for d in glob.glob(os.path.join(root, "*")) if os.path.isdir(d)]
def role_name_from_dir(role_dir: str) -> str:
return os.path.basename(role_dir.rstrip(os.sep))
def path_if_exists(*parts) -> Optional[str]:
p = os.path.join(*parts)
return p if os.path.exists(p) else None
def gather_yaml_files(base: str, patterns: List[str]) -> List[str]:
files: List[str] = []
for pat in patterns:
files.extend(glob.glob(os.path.join(base, pat), recursive=True))
return [f for f in files if os.path.isfile(f)]
def sq(v: str):
"""Return a single-quoted scalar (ruamel) for consistent quoting."""
return SingleQuotedScalarString(v)
# ---------------- Providers: vars & handlers ----------------
def flatten_keys(data) -> Set[str]:
out: Set[str] = set()
if isinstance(data, dict):
for k, v in data.items():
if isinstance(k, str):
out.add(k)
out |= flatten_keys(v)
elif isinstance(data, list):
for item in data:
out |= flatten_keys(item)
return out
def collect_role_defined_vars(role_dir: str) -> Set[str]:
"""Vars a role 'provides': defaults/vars keys + set_fact keys in tasks."""
provided: Set[str] = set()
for rel in ("defaults/main.yml", "vars/main.yml"):
p = path_if_exists(role_dir, rel)
if p:
data = load_yaml_rt(p)
provided |= flatten_keys(data)
# set_fact keys
task_files = gather_yaml_files(
os.path.join(role_dir, "tasks"), ["**/*.yml", "*.yml"]
)
for tf in task_files:
data = load_yaml_rt(tf)
if isinstance(data, list):
for task in data:
if (
isinstance(task, dict)
and "set_fact" in task
and isinstance(task["set_fact"], dict)
):
provided |= set(task["set_fact"].keys())
noisy = {"when", "name", "vars", "tags", "register"}
return {v for v in provided if isinstance(v, str) and v and v not in noisy}
def collect_role_handler_names(role_dir: str) -> Set[str]:
"""Handler names defined by a role (for notify detection)."""
handler_file = path_if_exists(role_dir, "handlers/main.yml")
if not handler_file:
return set()
data = load_yaml_rt(handler_file)
names: Set[str] = set()
if isinstance(data, list):
for task in data:
if isinstance(task, dict):
nm = task.get("name")
if isinstance(nm, str) and nm.strip():
names.add(nm.strip())
return names
# ---------------- Consumers: usage scanning ----------------
def find_var_positions(text: str, varname: str) -> List[int]:
"""Return byte offsets for occurrences of varname (word-ish boundary)."""
positions: List[int] = []
if not varname:
return positions
pattern = re.compile(rf"(?<!\w){re.escape(varname)}(?!\w)")
for m in pattern.finditer(text):
positions.append(m.start())
return positions
def first_var_use_offset_in_text(text: str, provided_vars: Set[str]) -> Optional[int]:
first: Optional[int] = None
for v in provided_vars:
for off in find_var_positions(text, v):
if first is None or off < first:
first = off
return first
def first_include_offset_for_role(text: str, producer_role: str) -> Optional[int]:
"""
Find earliest include/import of a given role in this YAML text.
Handles compact dict and block styles.
"""
pattern = re.compile(
r"(include_role|import_role)\s*:\s*\{[^}]*\bname\s*:\s*['\"]?"
+ re.escape(producer_role)
+ r"['\"]?[^}]*\}"
r"|"
r"(include_role|import_role)\s*:\s*\n(?:\s+[a-z_]+\s*:\s*.*\n)*\s*name\s*:\s*['\"]?"
+ re.escape(producer_role)
+ r"['\"]?",
re.IGNORECASE,
)
m = pattern.search(text)
return m.start() if m else None
def find_notify_offsets_for_handlers(text: str, handler_names: Set[str]) -> List[int]:
"""
Heuristic: for each handler name, find occurrences where 'notify' appears within
the preceding ~200 chars. Works for single string or list-style notify blocks.
"""
if not handler_names:
return []
offsets: List[int] = []
for h in handler_names:
for m in re.finditer(re.escape(h), text):
start = m.start()
back = max(0, start - 200)
context = text[back:start]
if re.search(r"notify\s*:", context):
offsets.append(start)
return sorted(offsets)
def parse_meta_dependencies(role_dir: str) -> List[str]:
meta = path_if_exists(role_dir, "meta/main.yml")
if not meta:
return []
data = load_yaml_rt(meta)
dd = data.get("dependencies")
deps: List[str] = []
if isinstance(dd, list):
for item in dd:
if isinstance(item, str):
deps.append(item)
elif isinstance(item, dict) and "role" in item:
deps.append(str(item["role"]))
elif isinstance(item, dict) and "name" in item:
deps.append(str(item["name"]))
return deps
# ---------------- Fix application ----------------
def sanitize_run_once_var(role_name: str) -> str:
"""
Generate run_once variable name from role name.
Example: 'sys-front-inj-logout' -> 'run_once_sys_front_inj_logout'
"""
return "run_once_" + role_name.replace("-", "_")
def build_include_block_yaml(consumer_role: str, moved_deps: List[str]) -> List[dict]:
"""
Build a guarded block that includes one or many roles.
This block will be prepended to tasks/01_core.yml or tasks/main.yml.
"""
guard_var = sanitize_run_once_var(consumer_role)
if len(moved_deps) == 1:
inner_tasks = [
{
"name": f"Include dependency '{moved_deps[0]}'",
"include_role": {"name": moved_deps[0]},
}
]
else:
inner_tasks = [
{
"name": "Include dependencies",
"include_role": {"name": "{{ item }}"},
"loop": moved_deps,
}
]
# Always set the run_once fact at the end
inner_tasks.append({"set_fact": {guard_var: True}})
# Correct Ansible block structure
block_task = {
"name": "Load former meta dependencies once",
"block": inner_tasks,
"when": f"{guard_var} is not defined",
}
return [block_task]
def prepend_tasks(tasks_path: str, new_tasks, dry_run: bool):
"""
Prepend new_tasks (CommentedSeq) to an existing tasks YAML list while preserving comments.
If the file does not exist, create it with new_tasks.
"""
if os.path.exists(tasks_path):
existing = load_yaml_rt(tasks_path)
if isinstance(existing, list):
combined = CommentedSeq()
for item in new_tasks:
combined.append(item)
for item in existing:
combined.append(item)
elif isinstance(existing, dict):
# Rare case: tasks file with a single mapping; coerce to list
combined = CommentedSeq()
for item in new_tasks:
combined.append(item)
combined.append(existing)
else:
combined = new_tasks
else:
os.makedirs(os.path.dirname(tasks_path), exist_ok=True)
combined = new_tasks
if dry_run:
print(
f"[DRY-RUN] Would write {tasks_path} with {len(new_tasks)} prepended task(s)."
def build_providers_index(
all_roles: List[str],
) -> Dict[str, Tuple[Set[str], Set[str]]]:
return {
role_name_from_dir(rd): (
collect_role_defined_vars(rd),
collect_role_handler_names(rd),
)
return
dump_yaml_rt(combined, tasks_path)
print(f"[OK] Updated {tasks_path} (prepended {len(new_tasks)} task(s)).")
def update_meta_remove_deps(meta_path: str, remove: List[str], dry_run: bool):
"""
Remove entries from meta.dependencies while leaving the rest of the file intact.
Quotes, comments, key order, and line breaks are preserved.
Returns True if a change would be made (or was made when not in dry-run).
"""
if not os.path.exists(meta_path):
return False
doc = load_yaml_rt(meta_path)
deps = doc.get("dependencies")
if not isinstance(deps, list):
return False
def dep_name(item):
if isinstance(item, dict):
return item.get("role") or item.get("name")
return item
keep = CommentedSeq()
removed = []
for item in deps:
name = dep_name(item)
if name in remove:
removed.append(name)
else:
keep.append(item)
if not removed:
return False
if keep:
doc["dependencies"] = keep
else:
if "dependencies" in doc:
del doc["dependencies"]
if dry_run:
print(f"[DRY-RUN] Would rewrite {meta_path}; removed: {', '.join(removed)}")
return True
dump_yaml_rt(doc, meta_path)
print(f"[OK] Rewrote {meta_path}; removed: {', '.join(removed)}")
return True
def dependency_is_unnecessary(
consumer_dir: str,
consumer_name: str,
producer_name: str,
provider_vars: Set[str],
provider_handlers: Set[str],
) -> bool:
"""Apply heuristic to decide if we can move this dependency."""
# 1) Early usage in defaults/vars/handlers? If yes -> necessary
defaults_files = [
p
for p in [
path_if_exists(consumer_dir, "defaults/main.yml"),
path_if_exists(consumer_dir, "vars/main.yml"),
path_if_exists(consumer_dir, "handlers/main.yml"),
]
if p
]
for p in defaults_files:
text = read_text(p)
if first_var_use_offset_in_text(text, provider_vars) is not None:
return False # needs meta dep
# 2) Tasks: any usage before include/import? If yes -> keep meta dep
task_files = gather_yaml_files(
os.path.join(consumer_dir, "tasks"), ["**/*.yml", "*.yml"]
)
for p in task_files:
text = read_text(p)
if not text:
continue
include_off = first_include_offset_for_role(text, producer_name)
var_use_off = first_var_use_offset_in_text(text, provider_vars)
notify_offs = find_notify_offsets_for_handlers(text, provider_handlers)
if var_use_off is not None:
if include_off is None or include_off > var_use_off:
return False # used before include
for noff in notify_offs:
if include_off is None or include_off > noff:
return False # notify before include
# If we get here: no early use, and either no usage at all or usage after include
return True
for rd in all_roles
}
def process_role(
@@ -437,9 +69,6 @@ def process_role(
only_role: Optional[str],
dry_run: bool,
) -> bool:
"""
Returns True if any change suggested/made for this role.
"""
consumer_name = role_name_from_dir(role_dir)
if only_role and only_role != consumer_name:
return False
@@ -448,12 +77,9 @@ def process_role(
if not meta_deps:
return False
# Build provider vars/handlers accessors
moved: List[str] = []
for producer in meta_deps:
# Only consider local roles we can analyze
if producer not in providers_index:
# Unknown/external role → skip (we cannot verify safety)
continue
pvars, phandlers = providers_index[producer]
if dependency_is_unnecessary(
@@ -464,33 +90,27 @@ def process_role(
if not moved:
return False
# 1) Remove from meta
meta_path = os.path.join(role_dir, "meta", "main.yml")
update_meta_remove_deps(meta_path, moved, dry_run=dry_run)
update_meta_remove_deps(
os.path.join(role_dir, "meta", "main.yml"), moved, dry_run=dry_run
)
# 2) Prepend include block to tasks/01_core.yml or tasks/main.yml
target_tasks = path_if_exists(role_dir, "tasks/01_core.yml")
if not target_tasks:
target_tasks = os.path.join(role_dir, "tasks", "main.yml")
include_block = build_include_block_yaml(consumer_name, moved)
prepend_tasks(target_tasks, include_block, dry_run=dry_run)
target_tasks = path_if_exists(role_dir, "tasks/01_core.yml") or os.path.join(
role_dir, "tasks", "main.yml"
)
prepend_tasks(
target_tasks,
build_include_block_yaml(consumer_name, moved),
dry_run=dry_run,
)
return True
def build_providers_index(all_roles: List[str]) -> Dict[str, Tuple[Set[str], Set[str]]]:
"""
Map role_name -> (provided_vars, handler_names)
"""
index: Dict[str, Tuple[Set[str], Set[str]]] = {}
for rd in all_roles:
rn = role_name_from_dir(rd)
index[rn] = (collect_role_defined_vars(rd), collect_role_handler_names(rd))
return index
def main():
def main() -> None:
parser = argparse.ArgumentParser(
description="Move unnecessary meta dependencies to guarded include_role for performance (preserve comments/quotes)."
description=(
"Move unnecessary meta dependencies to guarded include_role for "
"performance (preserve comments/quotes)."
)
)
parser.add_argument(
"--project-root",
@@ -522,16 +142,15 @@ def main():
changed_any = False
for role_dir in roles:
changed = process_role(role_dir, providers_index, args.only_role, args.dry_run)
changed_any = changed_any or changed
if process_role(role_dir, providers_index, args.only_role, args.dry_run):
changed_any = True
if not changed_any:
print("[OK] No unnecessary meta dependencies to move (per heuristic).")
elif args.dry_run:
print("[DRY-RUN] Completed analysis. No files were changed.")
else:
if args.dry_run:
print("[DRY-RUN] Completed analysis. No files were changed.")
else:
print("[OK] Finished moving unnecessary dependencies.")
print("[OK] Finished moving unnecessary dependencies.")
if __name__ == "__main__":
cli/fix/move_unnecessary_dependencies/analysis.py
+171
View File
@@ -0,0 +1,171 @@
"""Static analysis helpers: provider vars/handlers + consumer scan."""
from __future__ import annotations
import os
import re
from typing import List, Optional, Set
from .yaml_io import (
gather_yaml_files,
load_yaml_rt,
path_if_exists,
read_text,
)
def flatten_keys(data) -> Set[str]:
out: Set[str] = set()
if isinstance(data, dict):
for k, v in data.items():
if isinstance(k, str):
out.add(k)
out |= flatten_keys(v)
elif isinstance(data, list):
for item in data:
out |= flatten_keys(item)
return out
def collect_role_defined_vars(role_dir: str) -> Set[str]:
"""Vars a role 'provides': defaults/vars keys + set_fact keys in tasks."""
provided: Set[str] = set()
for rel in ("defaults/main.yml", "vars/main.yml"):
p = path_if_exists(role_dir, rel)
if p:
provided |= flatten_keys(load_yaml_rt(p))
task_files = gather_yaml_files(
os.path.join(role_dir, "tasks"), ["**/*.yml", "*.yml"]
)
for tf in task_files:
data = load_yaml_rt(tf)
if not isinstance(data, list):
continue
for task in data:
if isinstance(task, dict) and isinstance(task.get("set_fact"), dict):
provided |= set(task["set_fact"].keys())
noisy = {"when", "name", "vars", "tags", "register"}
return {v for v in provided if isinstance(v, str) and v and v not in noisy}
def collect_role_handler_names(role_dir: str) -> Set[str]:
handler_file = path_if_exists(role_dir, "handlers/main.yml")
if not handler_file:
return set()
data = load_yaml_rt(handler_file)
names: Set[str] = set()
if isinstance(data, list):
for task in data:
if isinstance(task, dict):
nm = task.get("name")
if isinstance(nm, str) and nm.strip():
names.add(nm.strip())
return names
def find_var_positions(text: str, varname: str) -> List[int]:
if not varname:
return []
pattern = re.compile(rf"(?<!\w){re.escape(varname)}(?!\w)")
return [m.start() for m in pattern.finditer(text)]
def first_var_use_offset_in_text(text: str, provided_vars: Set[str]) -> Optional[int]:
first: Optional[int] = None
for v in provided_vars:
for off in find_var_positions(text, v):
if first is None or off < first:
first = off
return first
def first_include_offset_for_role(text: str, producer_role: str) -> Optional[int]:
pattern = re.compile(
r"(include_role|import_role)\s*:\s*\{[^}]*\bname\s*:\s*['\"]?"
+ re.escape(producer_role)
+ r"['\"]?[^}]*\}"
r"|"
r"(include_role|import_role)\s*:\s*\n(?:\s+[a-z_]+\s*:\s*.*\n)*\s*name\s*:\s*['\"]?"
+ re.escape(producer_role)
+ r"['\"]?",
re.IGNORECASE,
)
m = pattern.search(text)
return m.start() if m else None
def find_notify_offsets_for_handlers(text: str, handler_names: Set[str]) -> List[int]:
"""For each handler name, return offsets that follow a `notify:` within ~200 chars."""
if not handler_names:
return []
offsets: List[int] = []
for h in handler_names:
for m in re.finditer(re.escape(h), text):
start = m.start()
context = text[max(0, start - 200) : start]
if re.search(r"notify\s*:", context):
offsets.append(start)
return sorted(offsets)
def parse_meta_dependencies(role_dir: str) -> List[str]:
meta = path_if_exists(role_dir, "meta/main.yml")
if not meta:
return []
data = load_yaml_rt(meta)
raw = data.get("dependencies")
deps: List[str] = []
if isinstance(raw, list):
for item in raw:
if isinstance(item, str):
deps.append(item)
elif isinstance(item, dict):
role = item.get("role") or item.get("name")
if role is not None:
deps.append(str(role))
return deps
def dependency_is_unnecessary(
consumer_dir: str,
consumer_name: str,
producer_name: str,
provider_vars: Set[str],
provider_handlers: Set[str],
) -> bool:
"""True iff the consumer can safely move the meta dep to a guarded include."""
early_files = [
p
for p in (
path_if_exists(consumer_dir, "defaults/main.yml"),
path_if_exists(consumer_dir, "vars/main.yml"),
path_if_exists(consumer_dir, "handlers/main.yml"),
)
if p
]
for p in early_files:
if first_var_use_offset_in_text(read_text(p), provider_vars) is not None:
return False
task_files = gather_yaml_files(
os.path.join(consumer_dir, "tasks"), ["**/*.yml", "*.yml"]
)
for p in task_files:
text = read_text(p)
if not text:
continue
include_off = first_include_offset_for_role(text, producer_name)
var_use_off = first_var_use_offset_in_text(text, provider_vars)
notify_offs = find_notify_offsets_for_handlers(text, provider_handlers)
if var_use_off is not None and (
include_off is None or include_off > var_use_off
):
return False
for noff in notify_offs:
if include_off is None or include_off > noff:
return False
return True
cli/fix/move_unnecessary_dependencies/apply.py
+108
View File
@@ -0,0 +1,108 @@
"""Apply the chosen fix: rewrite meta + prepend guarded include block."""
from __future__ import annotations
import os
from typing import List
from ruamel.yaml.comments import CommentedSeq
from .yaml_io import dump_yaml_rt, load_yaml_rt
def sanitize_run_once_var(role_name: str) -> str:
return "run_once_" + role_name.replace("-", "_")
def build_include_block_yaml(consumer_role: str, moved_deps: List[str]) -> List[dict]:
"""Guarded block prepended to tasks/01_core.yml or tasks/main.yml."""
guard_var = sanitize_run_once_var(consumer_role)
if len(moved_deps) == 1:
inner_tasks: List[dict] = [
{
"name": f"Include dependency '{moved_deps[0]}'",
"include_role": {"name": moved_deps[0]},
}
]
else:
inner_tasks = [
{
"name": "Include dependencies",
"include_role": {"name": "{{ item }}"},
"loop": moved_deps,
}
]
inner_tasks.append({"set_fact": {guard_var: True}})
return [
{
"name": "Load former meta dependencies once",
"block": inner_tasks,
"when": f"{guard_var} is not defined",
}
]
def prepend_tasks(tasks_path: str, new_tasks, dry_run: bool) -> None:
if os.path.exists(tasks_path):
existing = load_yaml_rt(tasks_path)
if isinstance(existing, list):
combined = CommentedSeq()
for item in new_tasks:
combined.append(item)
for item in existing:
combined.append(item)
elif isinstance(existing, dict):
combined = CommentedSeq()
for item in new_tasks:
combined.append(item)
combined.append(existing)
else:
combined = new_tasks
else:
os.makedirs(os.path.dirname(tasks_path), exist_ok=True)
combined = new_tasks
if dry_run:
print(
f"[DRY-RUN] Would write {tasks_path} with {len(new_tasks)} prepended task(s)."
)
return
dump_yaml_rt(combined, tasks_path)
print(f"[OK] Updated {tasks_path} (prepended {len(new_tasks)} task(s)).")
def update_meta_remove_deps(meta_path: str, remove: List[str], dry_run: bool) -> bool:
"""Drop given entries from meta.dependencies, preserving everything else."""
if not os.path.exists(meta_path):
return False
doc = load_yaml_rt(meta_path)
deps = doc.get("dependencies")
if not isinstance(deps, list):
return False
keep = CommentedSeq()
removed: List[str] = []
for item in deps:
name = item.get("role") or item.get("name") if isinstance(item, dict) else item
if name in remove:
removed.append(name)
else:
keep.append(item)
if not removed:
return False
if keep:
doc["dependencies"] = keep
elif "dependencies" in doc:
del doc["dependencies"]
if dry_run:
print(f"[DRY-RUN] Would rewrite {meta_path}; removed: {', '.join(removed)}")
return True
dump_yaml_rt(doc, meta_path)
print(f"[OK] Rewrote {meta_path}; removed: {', '.join(removed)}")
return True
cli/fix/move_unnecessary_dependencies/yaml_io.py
+94
View File
@@ -0,0 +1,94 @@
"""YAML round-trip + role-discovery helpers shared across the fix package."""
from __future__ import annotations
import glob
import logging
import os
import shutil
import sys
from typing import List, Optional
try:
from ruamel.yaml import YAML
from ruamel.yaml.comments import CommentedMap
from ruamel.yaml.scalarstring import SingleQuotedScalarString
_HAVE_RUAMEL = True
except Exception:
_HAVE_RUAMEL = False
if not _HAVE_RUAMEL:
logging.error(
"ruamel.yaml is required to preserve comments/quotes. "
"Install with: pip install ruamel.yaml"
)
sys.exit(3)
yaml_rt = YAML()
yaml_rt.preserve_quotes = True
yaml_rt.width = 10**9
def backup(path: str) -> None:
if os.path.exists(path):
shutil.copy2(path, path + ".bak")
def read_text(path: str) -> str:
try:
with open(path, "r", encoding="utf-8") as f:
return f.read()
except Exception:
return ""
def load_yaml_rt(path: str):
try:
with open(path, "r", encoding="utf-8") as f:
data = yaml_rt.load(f)
return data if data is not None else CommentedMap()
except FileNotFoundError:
return CommentedMap()
except Exception as e:
print(f"[WARN] Failed to parse YAML: {path}: {e}", file=sys.stderr)
return CommentedMap()
def dump_yaml_rt(data, path: str) -> None:
backup(path)
with open(path, "w", encoding="utf-8") as f:
yaml_rt.dump(data, f)
def sq(value: str):
"""Single-quoted scalar for consistent ruamel quoting."""
return SingleQuotedScalarString(value)
def roles_root(project_root: str) -> str:
return os.path.join(project_root, "roles")
def iter_role_dirs(project_root: str) -> List[str]:
return [
d
for d in glob.glob(os.path.join(roles_root(project_root), "*"))
if os.path.isdir(d)
]
def role_name_from_dir(role_dir: str) -> str:
return os.path.basename(role_dir.rstrip(os.sep))
def path_if_exists(*parts: str) -> Optional[str]:
p = os.path.join(*parts)
return p if os.path.exists(p) else None
def gather_yaml_files(base: str, patterns: List[str]) -> List[str]:
files: List[str] = []
for pat in patterns:
files.extend(glob.glob(os.path.join(base, pat), recursive=True))
return [f for f in files if os.path.isfile(f)]
roles/sys-svc-compose-ca/files/compose_ca.py
+4
View File
@@ -1,4 +1,8 @@
#!/usr/bin/env python3
# nocheck: file-size — single-host CA orchestration script. Ships as one
# file because it is delivered to the target via `ansible.builtin.copy`
# and must run with only the Python stdlib available; splitting it would
# require also delivering a package, which the role does not do.
from __future__ import annotations
import argparse
tasks/utils/migrate_meta_layout.py
-602
View File
@@ -1,602 +0,0 @@
#!/usr/bin/env python3
"""One-shot migration script for requirements 008 + 009 + 010.
Run from the repo root: `python3 tasks/utils/migrate_meta_layout.py`.
Idempotent: re-running on an already-migrated tree is a no-op for that role.
"""
from __future__ import annotations
import shutil
import sys
from collections import OrderedDict
from pathlib import Path
from typing import Any, Dict, Optional, Set, Tuple
import yaml
REPO_ROOT = Path(__file__).resolve().parents[2]
ROLES_DIR = REPO_ROOT / "roles"
NETWORKS_FILE = REPO_ROOT / "group_vars" / "all" / "08_networks.yml"
PORTS_FILE = REPO_ROOT / "group_vars" / "all" / "09_ports.yml"
sys.path.insert(0, str(REPO_ROOT))
from utils.entity_name_utils import get_entity_name # noqa: E402
ALLOWED_LIFECYCLES = {
"planned",
"pre-alpha",
"alpha",
"beta",
"stable",
"deprecated",
}
def _load_yaml(path: Path) -> Any:
if not path.is_file():
return None
text = path.read_text(encoding="utf-8")
if not text.strip():
return None
return yaml.safe_load(text) # noqa: direct-yaml — once-off migration script
def _dump_yaml(path: Path, data: Any) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
if data is None:
data = {}
text = yaml.safe_dump( # noqa: direct-yaml — once-off migration script
data,
sort_keys=False,
default_flow_style=False,
allow_unicode=True,
width=4096,
)
path.write_text(text, encoding="utf-8")
def _ordered_dump(data: Any) -> str:
return yaml.safe_dump( # noqa: direct-yaml — once-off migration script
data,
sort_keys=False,
default_flow_style=False,
allow_unicode=True,
width=4096,
)
def _empty_dir(path: Path) -> None:
if not path.is_dir():
return
for child in path.iterdir():
if child.is_file():
child.unlink()
elif child.is_dir():
shutil.rmtree(child)
path.rmdir()
def _deep_merge(base: Any, override: Any) -> Any:
if isinstance(base, dict) and isinstance(override, dict):
merged: Dict[str, Any] = {}
for key, value in base.items():
merged[key] = value
for key, value in override.items():
if key in merged:
merged[key] = _deep_merge(merged[key], value)
else:
merged[key] = value
return merged
return override
def _detect_credential_collision(
schema_creds: Dict[str, Any],
config_creds: Dict[str, Any],
role_name: str,
) -> None:
"""Walk both trees and fail loudly on shared paths."""
def _walk(prefix: str, schema_node: Any, config_node: Any) -> None:
if not isinstance(schema_node, dict) or not isinstance(config_node, dict):
if prefix:
raise SystemExit(
f"{role_name}: credential key collision at "
f"credentials.{prefix}: defined in both schema/main.yml and "
f"config/main.yml.credentials"
)
return
for key, schema_child in schema_node.items():
if key in config_node:
next_prefix = f"{prefix}.{key}" if prefix else key
config_child = config_node[key]
if isinstance(schema_child, dict) and isinstance(config_child, dict):
is_schema_leaf = (
"description" in schema_child
or "algorithm" in schema_child
or "validation" in schema_child
or "default" in schema_child
)
if is_schema_leaf:
raise SystemExit(
f"{role_name}: credential key collision at "
f"credentials.{next_prefix}: present in both "
f"schema/main.yml and config/main.yml.credentials"
)
_walk(next_prefix, schema_child, config_child)
else:
raise SystemExit(
f"{role_name}: credential key collision at "
f"credentials.{next_prefix}: present in both "
f"schema/main.yml and config/main.yml.credentials"
)
_walk("", schema_creds, config_creds)
def _convert_runtime_creds_to_schema(node: Any) -> Any:
"""Recursively turn `key: '{{ jinja }}'` runtime credentials into schema entries.
Each leaf becomes ``{description, algorithm: plain, default: <jinja>}``.
Nested dicts are walked.
"""
if not isinstance(node, dict):
return node
converted: Dict[str, Any] = {}
for key, value in node.items():
if isinstance(value, dict):
looks_like_schema = any(
inner_key in value
for inner_key in ("description", "algorithm", "validation", "default")
)
if looks_like_schema:
if "algorithm" not in value:
value = {"algorithm": "plain", **value}
converted[key] = value
else:
converted[key] = _convert_runtime_creds_to_schema(value)
elif isinstance(value, str):
converted[key] = {
"description": (
f"Runtime credential value imported from config/main.yml "
f"for '{key}'."
),
"algorithm": "plain",
"default": value,
}
else:
converted[key] = {
"description": (
f"Runtime credential value imported from config/main.yml "
f"for '{key}'."
),
"algorithm": "plain",
"default": value,
}
return converted
def _build_meta_files_for_role(
role_dir: Path,
networks_for_role: Optional[Dict[str, Any]],
ports_for_role: Dict[str, Dict[str, Any]],
relays_for_role: Dict[str, Dict[str, int]],
) -> None:
"""Run the file-move + content-rewrite for a single role.
Side effects (in `role_dir`):
* Reads ``schema/main.yml``, ``users/main.yml``, ``config/main.yml`` and
``meta/main.yml`` if present.
* Writes ``meta/{schema,users,server,rbac,services,volumes}.yml``
as required.
* Strips ``run_after`` and ``lifecycle`` from ``meta/main.yml`` and
re-emits it.
* Removes ``schema/``, ``users/``, ``config/`` directories on success.
"""
role_name = role_dir.name
primary_entity = get_entity_name(role_name) or role_name
config = _load_yaml(role_dir / "config" / "main.yml") or {}
schema = _load_yaml(role_dir / "schema" / "main.yml") or {}
users_data = _load_yaml(role_dir / "users" / "main.yml") or {}
meta_main = _load_yaml(role_dir / "meta" / "main.yml") or {}
if not isinstance(config, dict):
raise SystemExit(f"{role_name}: config/main.yml is not a mapping")
if not isinstance(schema, dict):
raise SystemExit(f"{role_name}: schema/main.yml is not a mapping")
if not isinstance(users_data, dict):
raise SystemExit(f"{role_name}: users/main.yml is not a mapping")
if not isinstance(meta_main, dict):
raise SystemExit(f"{role_name}: meta/main.yml is not a mapping")
# ---- credentials merge --------------------------------------------------
schema_creds = (
dict(schema.get("credentials") or {})
if isinstance(schema.get("credentials"), dict)
else {}
)
config_creds_raw = config.get("credentials") if isinstance(config, dict) else None
config_creds = dict(config_creds_raw) if isinstance(config_creds_raw, dict) else {}
converted_config_creds = _convert_runtime_creds_to_schema(config_creds)
_detect_credential_collision(schema_creds, converted_config_creds, role_name)
merged_creds = _deep_merge(schema_creds, converted_config_creds)
if merged_creds:
_dump_yaml(
role_dir / "meta" / "schema.yml",
{"credentials": merged_creds},
)
# ---- users.yml ----------------------------------------------------------
users_block = users_data.get("users") if isinstance(users_data, dict) else None
if isinstance(users_block, dict) and users_block:
_dump_yaml(role_dir / "meta" / "users.yml", users_block)
# ---- server, rbac -------------------------------------------------------
server_block = config.get("server")
if isinstance(server_block, dict) and server_block:
server_payload = dict(server_block)
if networks_for_role:
server_payload["networks"] = {"local": networks_for_role}
_dump_yaml(role_dir / "meta" / "server.yml", server_payload)
elif networks_for_role:
_dump_yaml(
role_dir / "meta" / "server.yml",
{"networks": {"local": networks_for_role}},
)
rbac_block = config.get("rbac")
if isinstance(rbac_block, dict) and rbac_block:
_dump_yaml(role_dir / "meta" / "rbac.yml", rbac_block)
# ---- compose volumes ----------------------------------------------------
compose = config.get("compose") if isinstance(config, dict) else None
compose = compose if isinstance(compose, dict) else {}
volumes_block = compose.get("volumes")
if isinstance(volumes_block, dict) and volumes_block:
_dump_yaml(role_dir / "meta" / "volumes.yml", volumes_block)
# ---- services + inlined keys + ports + run_after/lifecycle --------------
compose_services_raw = compose.get("services")
services: "OrderedDict[str, Dict[str, Any]]" = OrderedDict()
if isinstance(compose_services_raw, dict):
for service_key, service_value in compose_services_raw.items():
if isinstance(service_value, dict):
# rename `port:` (int or quoted) -> `ports.inter`
service_copy = dict(service_value)
if "port" in service_copy:
raw_port = service_copy.pop("port")
try:
inter_port = int(str(raw_port).strip())
except (TypeError, ValueError):
# Non-int (Jinja expression for example) -> stay as
# `inter` but raw, keep semantics
inter_port = raw_port
ports_field = service_copy.get("ports")
if not isinstance(ports_field, dict):
ports_field = {}
ports_field["inter"] = inter_port
service_copy["ports"] = ports_field
services[service_key] = service_copy
else:
services[service_key] = service_value
# Inline top-level non-special keys under primary_entity
EXCLUDED_TOP_LEVEL_KEYS = {"compose", "server", "rbac", "credentials"}
inlined: Dict[str, Any] = {}
for key, value in config.items():
if key in EXCLUDED_TOP_LEVEL_KEYS:
continue
inlined[key] = value
if inlined:
primary_node = services.setdefault(primary_entity, {})
if not isinstance(primary_node, dict):
raise SystemExit(
f"{role_name}: cannot inline keys under non-dict primary entity "
f"'{primary_entity}'."
)
for key, value in inlined.items():
if key in primary_node:
# Preserve compose-defined values, layer inlined under them
primary_node[key] = _deep_merge(value, primary_node[key])
else:
primary_node[key] = value
services[primary_entity] = primary_node
# Apply per-entity ports from 09_ports.yml
for entity_key, ports_payload in ports_for_role.items():
node = services.setdefault(entity_key, {})
if not isinstance(node, dict):
raise SystemExit(
f"{role_name}: cannot attach ports to non-dict entity '{entity_key}'."
)
node_ports = node.get("ports") if isinstance(node.get("ports"), dict) else {}
# ports_payload looks like {'local': {...}, 'public': {...}}
for scope_name, scope_map in ports_payload.items():
existing_scope = node_ports.get(scope_name)
if isinstance(existing_scope, dict):
merged_scope = dict(existing_scope)
merged_scope.update(scope_map)
node_ports[scope_name] = merged_scope
else:
node_ports[scope_name] = dict(scope_map)
node["ports"] = node_ports
services[entity_key] = node
# Apply relay ranges
for entity_key, relay in relays_for_role.items():
node = services.setdefault(entity_key, {})
if not isinstance(node, dict):
raise SystemExit(
f"{role_name}: cannot attach relay range to non-dict entity "
f"'{entity_key}'."
)
node_ports = node.get("ports") if isinstance(node.get("ports"), dict) else {}
public_scope = (
node_ports.get("public")
if isinstance(node_ports.get("public"), dict)
else {}
)
public_scope["relay"] = {"start": relay["start"], "end": relay["end"]}
node_ports["public"] = public_scope
node["ports"] = node_ports
services[entity_key] = node
# Migrate run_after / lifecycle from meta/main.yml
galaxy_info_changed = False
galaxy_info = meta_main.get("galaxy_info")
if isinstance(galaxy_info, dict):
run_after_value = galaxy_info.pop("run_after", None)
lifecycle_value = galaxy_info.pop("lifecycle", None)
if run_after_value is not None or lifecycle_value is not None:
galaxy_info_changed = True
primary_node = services.setdefault(primary_entity, {})
if not isinstance(primary_node, dict):
raise SystemExit(
f"{role_name}: cannot attach run_after/lifecycle to non-dict "
f"primary entity '{primary_entity}'."
)
if isinstance(run_after_value, list) and run_after_value:
primary_node["run_after"] = list(run_after_value)
if isinstance(lifecycle_value, str) and lifecycle_value.strip():
stripped = lifecycle_value.strip().lower()
if stripped not in ALLOWED_LIFECYCLES:
raise SystemExit(
f"{role_name}: meta/main.yml.galaxy_info.lifecycle has "
f"unknown value {lifecycle_value!r}. Allowed: "
f"{sorted(ALLOWED_LIFECYCLES)}."
)
primary_node["lifecycle"] = stripped
services[primary_entity] = primary_node
if services:
_dump_yaml(role_dir / "meta" / "services.yml", dict(services))
if galaxy_info_changed:
_dump_yaml(role_dir / "meta" / "main.yml", meta_main)
# ---- cleanup old directories -------------------------------------------
for legacy in ("config", "schema", "users"):
legacy_dir = role_dir / legacy
if legacy_dir.is_dir():
_empty_dir(legacy_dir)
# --------------------------------------------------------------------------
# Centralised network/port extraction
# --------------------------------------------------------------------------
def _build_network_index() -> Dict[str, Dict[str, Any]]:
networks = _load_yaml(NETWORKS_FILE) or {}
raw = networks.get("defaults_networks", {}) if isinstance(networks, dict) else {}
local = raw.get("local", {}) if isinstance(raw, dict) else {}
if not isinstance(local, dict):
return {}
result: Dict[str, Dict[str, Any]] = {}
for role_name, value in local.items():
if isinstance(value, str):
result[role_name] = {"subnet": value}
elif isinstance(value, dict):
entry: Dict[str, Any] = {}
if "subnet" in value:
entry["subnet"] = value["subnet"]
if "dns_resolver" in value:
entry["dns_resolver"] = value["dns_resolver"]
if entry:
result[role_name] = entry
return result
# Roles where multi-entity port keys use a non-default split. Empty by default;
# `split_role_entity` falls back to a deterministic suffix split rooted in the
# real role directory listing.
_KNOWN_ROLES_CACHE: Optional[Set[str]] = None
def _all_role_names() -> Set[str]:
global _KNOWN_ROLES_CACHE
if _KNOWN_ROLES_CACHE is None:
_KNOWN_ROLES_CACHE = {
child.name for child in ROLES_DIR.iterdir() if child.is_dir()
}
return _KNOWN_ROLES_CACHE
def _split_role_and_entity(combined_key: str) -> Tuple[str, Optional[str]]:
"""Return ``(role_name, entity_or_None)`` for a `09_ports.yml` key.
The convention is ``<role>_<entity>`` for multi-entity roles, e.g.
``web-app-bluesky_api`` -> ``("web-app-bluesky", "api")``. Single-entity
roles use the bare role name, e.g. ``web-app-gitea`` -> ``("web-app-gitea", None)``.
Resolution is deterministic: if the bare key is itself a known role, no
split is attempted; otherwise the suffix after the last ``_`` is treated
as the entity name when the prefix matches a known role.
"""
known = _all_role_names()
if combined_key in known:
return combined_key, None
if "_" in combined_key:
prefix, suffix = combined_key.rsplit("_", 1)
if prefix in known:
return prefix, suffix
# Unknown role; carry the key through verbatim so the caller can flag it.
return combined_key, None
def _build_port_index() -> Tuple[
Dict[str, Dict[str, Dict[str, Any]]], Dict[str, Dict[str, Dict[str, int]]]
]:
"""Return ``(ports_for_role, relays_for_role)``.
Shape::
ports_for_role[role][entity] = {"local": {<cat>: <int>}, "public": {...}}
relays_for_role[role][entity] = {"start": <int>, "end": <int>}
"""
ports_doc = _load_yaml(PORTS_FILE) or {}
ports_root = ports_doc.get("ports", {}) if isinstance(ports_doc, dict) else {}
if not isinstance(ports_root, dict):
return {}, {}
ports_for_role: Dict[str, Dict[str, Dict[str, Any]]] = {}
relays_for_role: Dict[str, Dict[str, Dict[str, int]]] = {}
for scope in ("localhost", "public"):
scope_block = ports_root.get(scope, {})
if not isinstance(scope_block, dict):
continue
scope_label = "local" if scope == "localhost" else "public"
for category, entries in scope_block.items():
if category == "relay_port_ranges":
if not isinstance(entries, dict):
continue
for combined_key, port_value in entries.items():
if not isinstance(port_value, int):
continue
if combined_key.endswith("_start"):
base = combined_key[: -len("_start")]
kind = "start"
elif combined_key.endswith("_end"):
base = combined_key[: -len("_end")]
kind = "end"
else:
continue
role_name, entity = _split_role_and_entity(base)
entity_key = entity or get_entity_name(role_name) or role_name
role_relays = relays_for_role.setdefault(role_name, {})
relay_entry = role_relays.setdefault(entity_key, {})
relay_entry[kind] = port_value
continue
if not isinstance(entries, dict):
continue
for combined_key, port_value in entries.items():
if not isinstance(port_value, int):
continue
role_name, entity = _split_role_and_entity(combined_key)
entity_key = entity or get_entity_name(role_name) or role_name
role_ports = ports_for_role.setdefault(role_name, {})
entity_ports = role_ports.setdefault(entity_key, {})
scope_map = entity_ports.setdefault(scope_label, {})
scope_map[category] = port_value
return ports_for_role, relays_for_role
# --------------------------------------------------------------------------
# Entry point
# --------------------------------------------------------------------------
def main() -> int:
if not ROLES_DIR.is_dir():
print(f"roles directory not found at {ROLES_DIR}", file=sys.stderr)
return 1
networks = _build_network_index()
ports_for_role, relays_for_role = _build_port_index()
role_dirs = sorted(p for p in ROLES_DIR.iterdir() if p.is_dir())
touched = 0
for role_dir in role_dirs:
legacy_dirs_present = any(
(role_dir / sub).is_dir() for sub in ("config", "schema", "users")
)
meta_main_path = role_dir / "meta" / "main.yml"
meta_main = _load_yaml(meta_main_path) if meta_main_path.is_file() else None
gi = (
meta_main.get("galaxy_info") if isinstance(meta_main, dict) else None
) or {}
meta_main_carries_legacy = isinstance(gi, dict) and (
"run_after" in gi or "lifecycle" in gi
)
has_centralised_data = (
role_dir.name in networks
or role_dir.name in ports_for_role
or role_dir.name in relays_for_role
)
if not (
legacy_dirs_present or meta_main_carries_legacy or has_centralised_data
):
continue
_build_meta_files_for_role(
role_dir,
networks.get(role_dir.name),
ports_for_role.get(role_dir.name, {}),
relays_for_role.get(role_dir.name, {}),
)
touched += 1
print(f"migrated {touched} roles", file=sys.stderr)
# Strip migrated keys from group_vars/all/{08_networks,09_ports}.yml.
# 09_ports.yml is deleted entirely; 08_networks.yml keeps NETWORK_IPV6_ENABLED
# plus defaults_networks.internet.
networks_doc = _load_yaml(NETWORKS_FILE) or {}
if isinstance(networks_doc, dict):
defaults = networks_doc.get("defaults_networks")
if isinstance(defaults, dict) and "local" in defaults:
defaults.pop("local", None)
if not defaults:
networks_doc.pop("defaults_networks", None)
else:
networks_doc["defaults_networks"] = defaults
# Inject PORT_BANDS map (per req-009).
networks_doc["PORT_BANDS"] = {
"local": {
"http": {"start": 8001, "end": 8099},
"websocket": {"start": 4001, "end": 4099},
"oauth2": {"start": 16480, "end": 16499},
"database": {"start": 3306, "end": 5432},
"ldap": {"start": 389, "end": 389},
},
"public": {
"ssh": {"start": 2201, "end": 2299},
"stun_turn": {"start": 3478, "end": 3499},
"stun_turn_tls": {"start": 5349, "end": 5399},
"federation": {"start": 8448, "end": 8499},
"ldaps": {"start": 636, "end": 636},
"relay": {"start": 20000, "end": 59999},
},
}
_dump_yaml(NETWORKS_FILE, networks_doc)
if PORTS_FILE.is_file():
PORTS_FILE.unlink()
return 0
if __name__ == "__main__": # pragma: no cover
sys.exit(main())
tasks/utils/migrate_meta_layout/__init__.py
+8
View File
@@ -0,0 +1,8 @@
"""One-shot migration package for requirements 008 + 009 + 010.
Invoke from the repo root:
python3 -m tasks.utils.migrate_meta_layout
Idempotent per role: re-running on an already-migrated tree is a no-op.
"""
tasks/utils/migrate_meta_layout/__main__.py
+105
View File
@@ -0,0 +1,105 @@
"""Entry point: ``python3 -m tasks.utils.migrate_meta_layout``."""
from __future__ import annotations
import sys
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parents[3]
ROLES_DIR = REPO_ROOT / "roles"
NETWORKS_FILE = REPO_ROOT / "group_vars" / "all" / "08_networks.yml"
PORTS_FILE = REPO_ROOT / "group_vars" / "all" / "09_ports.yml"
sys.path.insert(0, str(REPO_ROOT))
from . import yaml_io # noqa: E402
from .indices import build_network_index, build_port_index # noqa: E402
from .role_builder import build as build_role_meta # noqa: E402
_PORT_BANDS = {
"local": {
"http": {"start": 8001, "end": 8099},
"websocket": {"start": 4001, "end": 4099},
"oauth2": {"start": 16480, "end": 16499},
"database": {"start": 3306, "end": 5432},
"ldap": {"start": 389, "end": 389},
},
"public": {
"ssh": {"start": 2201, "end": 2299},
"stun_turn": {"start": 3478, "end": 3499},
"stun_turn_tls": {"start": 5349, "end": 5399},
"federation": {"start": 8448, "end": 8499},
"ldaps": {"start": 636, "end": 636},
"relay": {"start": 20000, "end": 59999},
},
}
def _role_needs_migration(
role_dir: Path,
networks: dict,
ports_for_role: dict,
relays_for_role: dict,
) -> bool:
legacy_dirs = any(
(role_dir / sub).is_dir() for sub in ("config", "schema", "users")
)
meta_main_path = role_dir / "meta" / "main.yml"
meta_main = yaml_io.load(meta_main_path) if meta_main_path.is_file() else None
galaxy_info = meta_main.get("galaxy_info") if isinstance(meta_main, dict) else None
galaxy_info = galaxy_info if isinstance(galaxy_info, dict) else {}
legacy_galaxy = "run_after" in galaxy_info or "lifecycle" in galaxy_info
has_centralised = (
role_dir.name in networks
or role_dir.name in ports_for_role
or role_dir.name in relays_for_role
)
return legacy_dirs or legacy_galaxy or has_centralised
def _strip_centralised_files() -> None:
networks_doc = yaml_io.load(NETWORKS_FILE) or {}
if isinstance(networks_doc, dict):
defaults = networks_doc.get("defaults_networks")
if isinstance(defaults, dict) and "local" in defaults:
defaults.pop("local", None)
if defaults:
networks_doc["defaults_networks"] = defaults
else:
networks_doc.pop("defaults_networks", None)
networks_doc["PORT_BANDS"] = _PORT_BANDS
yaml_io.dump(NETWORKS_FILE, networks_doc)
if PORTS_FILE.is_file():
PORTS_FILE.unlink()
def main() -> int:
if not ROLES_DIR.is_dir():
print(f"roles directory not found at {ROLES_DIR}", file=sys.stderr)
return 1
networks = build_network_index(NETWORKS_FILE)
ports_for_role, relays_for_role = build_port_index(PORTS_FILE, ROLES_DIR)
touched = 0
for role_dir in sorted(p for p in ROLES_DIR.iterdir() if p.is_dir()):
if not _role_needs_migration(
role_dir, networks, ports_for_role, relays_for_role
):
continue
build_role_meta(
role_dir,
networks.get(role_dir.name),
ports_for_role.get(role_dir.name, {}),
relays_for_role.get(role_dir.name, {}),
)
touched += 1
print(f"migrated {touched} roles", file=sys.stderr)
_strip_centralised_files()
return 0
if __name__ == "__main__": # pragma: no cover
sys.exit(main())
tasks/utils/migrate_meta_layout/credentials.py
+66
View File
@@ -0,0 +1,66 @@
from __future__ import annotations
from typing import Any, Dict
_SCHEMA_LEAF_KEYS = ("description", "algorithm", "validation", "default")
def detect_collision(
schema_creds: Dict[str, Any],
config_creds: Dict[str, Any],
role_name: str,
) -> None:
"""Raise SystemExit if any path is defined in both schema and config."""
def _walk(prefix: str, schema_node: Any, config_node: Any) -> None:
if not isinstance(schema_node, dict) or not isinstance(config_node, dict):
if prefix:
_fail(role_name, prefix)
return
for key, schema_child in schema_node.items():
if key not in config_node:
continue
next_prefix = f"{prefix}.{key}" if prefix else key
config_child = config_node[key]
if isinstance(schema_child, dict) and isinstance(config_child, dict):
if any(k in schema_child for k in _SCHEMA_LEAF_KEYS):
_fail(role_name, next_prefix)
_walk(next_prefix, schema_child, config_child)
else:
_fail(role_name, next_prefix)
_walk("", schema_creds, config_creds)
def _fail(role_name: str, dotted_path: str) -> None:
raise SystemExit(
f"{role_name}: credential key collision at "
f"credentials.{dotted_path}: present in both schema/main.yml and "
f"config/main.yml.credentials"
)
def convert_runtime_to_schema(node: Any) -> Any:
"""Turn `key: '{{ jinja }}'` runtime credentials into schema entries."""
if not isinstance(node, dict):
return node
converted: Dict[str, Any] = {}
for key, value in node.items():
if isinstance(value, dict):
looks_like_schema = any(k in value for k in _SCHEMA_LEAF_KEYS)
if looks_like_schema:
if "algorithm" not in value:
value = {"algorithm": "plain", **value}
converted[key] = value
else:
converted[key] = convert_runtime_to_schema(value)
else:
converted[key] = {
"description": (
f"Runtime credential value imported from config/main.yml "
f"for '{key}'."
),
"algorithm": "plain",
"default": value,
}
return converted
tasks/utils/migrate_meta_layout/indices.py
+115
View File
@@ -0,0 +1,115 @@
from __future__ import annotations
from pathlib import Path
from typing import Any, Dict, Optional, Set, Tuple
from utils.entity_name_utils import get_entity_name
from . import yaml_io
_KNOWN_ROLES_CACHE: Optional[Set[str]] = None
def build_network_index(networks_file: Path) -> Dict[str, Dict[str, Any]]:
networks = yaml_io.load(networks_file) or {}
raw = networks.get("defaults_networks", {}) if isinstance(networks, dict) else {}
local = raw.get("local", {}) if isinstance(raw, dict) else {}
if not isinstance(local, dict):
return {}
result: Dict[str, Dict[str, Any]] = {}
for role_name, value in local.items():
if isinstance(value, str):
result[role_name] = {"subnet": value}
elif isinstance(value, dict):
entry: Dict[str, Any] = {}
for k in ("subnet", "dns_resolver"):
if k in value:
entry[k] = value[k]
if entry:
result[role_name] = entry
return result
def build_port_index(
ports_file: Path, roles_dir: Path
) -> Tuple[
Dict[str, Dict[str, Dict[str, Any]]],
Dict[str, Dict[str, Dict[str, int]]],
]:
"""Return ``(ports_for_role, relays_for_role)`` indexed by role -> entity."""
ports_doc = yaml_io.load(ports_file) or {}
ports_root = ports_doc.get("ports", {}) if isinstance(ports_doc, dict) else {}
if not isinstance(ports_root, dict):
return {}, {}
ports_for_role: Dict[str, Dict[str, Dict[str, Any]]] = {}
relays_for_role: Dict[str, Dict[str, Dict[str, int]]] = {}
for scope in ("localhost", "public"):
scope_block = ports_root.get(scope, {})
if not isinstance(scope_block, dict):
continue
scope_label = "local" if scope == "localhost" else "public"
for category, entries in scope_block.items():
if category == "relay_port_ranges":
_absorb_relays(entries, relays_for_role, roles_dir)
continue
if not isinstance(entries, dict):
continue
for combined_key, port_value in entries.items():
if not isinstance(port_value, int):
continue
role_name, entity = _split_role_and_entity(combined_key, roles_dir)
entity_key = entity or get_entity_name(role_name) or role_name
role_ports = ports_for_role.setdefault(role_name, {})
entity_ports = role_ports.setdefault(entity_key, {})
scope_map = entity_ports.setdefault(scope_label, {})
scope_map[category] = port_value
return ports_for_role, relays_for_role
def _absorb_relays(
entries: Any,
relays_for_role: Dict[str, Dict[str, Dict[str, int]]],
roles_dir: Path,
) -> None:
if not isinstance(entries, dict):
return
for combined_key, port_value in entries.items():
if not isinstance(port_value, int):
continue
if combined_key.endswith("_start"):
base, kind = combined_key[: -len("_start")], "start"
elif combined_key.endswith("_end"):
base, kind = combined_key[: -len("_end")], "end"
else:
continue
role_name, entity = _split_role_and_entity(base, roles_dir)
entity_key = entity or get_entity_name(role_name) or role_name
role_relays = relays_for_role.setdefault(role_name, {})
role_relays.setdefault(entity_key, {})[kind] = port_value
def _split_role_and_entity(
combined_key: str, roles_dir: Path
) -> Tuple[str, Optional[str]]:
"""Return ``(role_name, entity_or_None)`` for a `09_ports.yml` key."""
known = _all_role_names(roles_dir)
if combined_key in known:
return combined_key, None
if "_" in combined_key:
prefix, suffix = combined_key.rsplit("_", 1)
if prefix in known:
return prefix, suffix
return combined_key, None
def _all_role_names(roles_dir: Path) -> Set[str]:
global _KNOWN_ROLES_CACHE
if _KNOWN_ROLES_CACHE is None:
_KNOWN_ROLES_CACHE = {
child.name for child in roles_dir.iterdir() if child.is_dir()
}
return _KNOWN_ROLES_CACHE
tasks/utils/migrate_meta_layout/role_builder.py
+250
View File
@@ -0,0 +1,250 @@
from __future__ import annotations
from collections import OrderedDict
from pathlib import Path
from typing import Any, Dict, Optional
from utils.entity_name_utils import get_entity_name
from . import yaml_io
from .credentials import convert_runtime_to_schema, detect_collision
ALLOWED_LIFECYCLES = {
"planned",
"pre-alpha",
"alpha",
"beta",
"stable",
"deprecated",
}
_EXCLUDED_TOP_LEVEL_KEYS = {"compose", "server", "rbac", "credentials"}
def build(
role_dir: Path,
networks_for_role: Optional[Dict[str, Any]],
ports_for_role: Dict[str, Dict[str, Any]],
relays_for_role: Dict[str, Dict[str, int]],
) -> None:
role_name = role_dir.name
primary_entity = get_entity_name(role_name) or role_name
config = _load_mapping(role_dir / "config" / "main.yml", role_name)
schema = _load_mapping(role_dir / "schema" / "main.yml", role_name)
users_data = _load_mapping(role_dir / "users" / "main.yml", role_name)
meta_main = _load_mapping(role_dir / "meta" / "main.yml", role_name)
_emit_credentials(role_dir, role_name, schema, config)
_emit_users(role_dir, users_data)
_emit_server(role_dir, config, networks_for_role)
_emit_rbac(role_dir, config)
_emit_volumes(role_dir, config)
services = _build_services(config, primary_entity, role_name)
_apply_centralised_ports(services, ports_for_role, role_name)
_apply_relays(services, relays_for_role, role_name)
galaxy_changed = _migrate_run_after_lifecycle(
meta_main, services, primary_entity, role_name
)
if services:
yaml_io.dump(role_dir / "meta" / "services.yml", dict(services))
if galaxy_changed:
yaml_io.dump(role_dir / "meta" / "main.yml", meta_main)
for legacy in ("config", "schema", "users"):
yaml_io.empty_dir(role_dir / legacy)
def _load_mapping(path: Path, role_name: str) -> Dict[str, Any]:
data = yaml_io.load(path) or {}
if not isinstance(data, dict):
raise SystemExit(f"{role_name}: {path} is not a mapping")
return data
def _emit_credentials(
role_dir: Path,
role_name: str,
schema: Dict[str, Any],
config: Dict[str, Any],
) -> None:
schema_creds = (
dict(schema.get("credentials") or {})
if isinstance(schema.get("credentials"), dict)
else {}
)
config_creds_raw = config.get("credentials")
config_creds = dict(config_creds_raw) if isinstance(config_creds_raw, dict) else {}
converted = convert_runtime_to_schema(config_creds)
detect_collision(schema_creds, converted, role_name)
merged = yaml_io.deep_merge(schema_creds, converted)
if merged:
yaml_io.dump(role_dir / "meta" / "schema.yml", {"credentials": merged})
def _emit_users(role_dir: Path, users_data: Dict[str, Any]) -> None:
users_block = users_data.get("users")
if isinstance(users_block, dict) and users_block:
yaml_io.dump(role_dir / "meta" / "users.yml", users_block)
def _emit_server(
role_dir: Path,
config: Dict[str, Any],
networks_for_role: Optional[Dict[str, Any]],
) -> None:
server_block = config.get("server")
if isinstance(server_block, dict) and server_block:
payload = dict(server_block)
if networks_for_role:
payload["networks"] = {"local": networks_for_role}
yaml_io.dump(role_dir / "meta" / "server.yml", payload)
elif networks_for_role:
yaml_io.dump(
role_dir / "meta" / "server.yml",
{"networks": {"local": networks_for_role}},
)
def _emit_rbac(role_dir: Path, config: Dict[str, Any]) -> None:
rbac_block = config.get("rbac")
if isinstance(rbac_block, dict) and rbac_block:
yaml_io.dump(role_dir / "meta" / "rbac.yml", rbac_block)
def _emit_volumes(role_dir: Path, config: Dict[str, Any]) -> None:
compose = config.get("compose")
if not isinstance(compose, dict):
return
volumes_block = compose.get("volumes")
if isinstance(volumes_block, dict) and volumes_block:
yaml_io.dump(role_dir / "meta" / "volumes.yml", volumes_block)
def _build_services(
config: Dict[str, Any], primary_entity: str, role_name: str
) -> "OrderedDict[str, Dict[str, Any]]":
compose = config.get("compose") if isinstance(config.get("compose"), dict) else {}
services: "OrderedDict[str, Dict[str, Any]]" = OrderedDict()
raw_services = compose.get("services") if isinstance(compose, dict) else None
if isinstance(raw_services, dict):
for key, value in raw_services.items():
services[key] = _normalise_service_entry(value)
inlined = {k: v for k, v in config.items() if k not in _EXCLUDED_TOP_LEVEL_KEYS}
if inlined:
primary_node = services.setdefault(primary_entity, {})
if not isinstance(primary_node, dict):
raise SystemExit(
f"{role_name}: cannot inline keys under non-dict primary entity "
f"'{primary_entity}'."
)
for key, value in inlined.items():
primary_node[key] = (
yaml_io.deep_merge(value, primary_node[key])
if key in primary_node
else value
)
services[primary_entity] = primary_node
return services
def _normalise_service_entry(value: Any) -> Any:
if not isinstance(value, dict):
return value
entry = dict(value)
if "port" in entry:
raw_port = entry.pop("port")
try:
inter_port: Any = int(str(raw_port).strip())
except (TypeError, ValueError):
inter_port = raw_port
ports_field = entry.get("ports")
if not isinstance(ports_field, dict):
ports_field = {}
ports_field["inter"] = inter_port
entry["ports"] = ports_field
return entry
def _apply_centralised_ports(
services: "OrderedDict[str, Dict[str, Any]]",
ports_for_role: Dict[str, Dict[str, Any]],
role_name: str,
) -> None:
for entity_key, ports_payload in ports_for_role.items():
node = services.setdefault(entity_key, {})
if not isinstance(node, dict):
raise SystemExit(
f"{role_name}: cannot attach ports to non-dict entity '{entity_key}'."
)
node_ports = node.get("ports") if isinstance(node.get("ports"), dict) else {}
for scope_name, scope_map in ports_payload.items():
existing = node_ports.get(scope_name)
if isinstance(existing, dict):
merged = dict(existing)
merged.update(scope_map)
node_ports[scope_name] = merged
else:
node_ports[scope_name] = dict(scope_map)
node["ports"] = node_ports
services[entity_key] = node
def _apply_relays(
services: "OrderedDict[str, Dict[str, Any]]",
relays_for_role: Dict[str, Dict[str, int]],
role_name: str,
) -> None:
for entity_key, relay in relays_for_role.items():
node = services.setdefault(entity_key, {})
if not isinstance(node, dict):
raise SystemExit(
f"{role_name}: cannot attach relay range to non-dict entity "
f"'{entity_key}'."
)
node_ports = node.get("ports") if isinstance(node.get("ports"), dict) else {}
public_scope = (
node_ports.get("public")
if isinstance(node_ports.get("public"), dict)
else {}
)
public_scope["relay"] = {"start": relay["start"], "end": relay["end"]}
node_ports["public"] = public_scope
node["ports"] = node_ports
services[entity_key] = node
def _migrate_run_after_lifecycle(
meta_main: Dict[str, Any],
services: "OrderedDict[str, Dict[str, Any]]",
primary_entity: str,
role_name: str,
) -> bool:
galaxy_info = meta_main.get("galaxy_info")
if not isinstance(galaxy_info, dict):
return False
run_after = galaxy_info.pop("run_after", None)
lifecycle = galaxy_info.pop("lifecycle", None)
if run_after is None and lifecycle is None:
return False
primary_node = services.setdefault(primary_entity, {})
if not isinstance(primary_node, dict):
raise SystemExit(
f"{role_name}: cannot attach run_after/lifecycle to non-dict primary "
f"entity '{primary_entity}'."
)
if isinstance(run_after, list) and run_after:
primary_node["run_after"] = list(run_after)
if isinstance(lifecycle, str) and lifecycle.strip():
stripped = lifecycle.strip().lower()
if stripped not in ALLOWED_LIFECYCLES:
raise SystemExit(
f"{role_name}: meta/main.yml.galaxy_info.lifecycle has unknown "
f"value {lifecycle!r}. Allowed: {sorted(ALLOWED_LIFECYCLES)}."
)
primary_node["lifecycle"] = stripped
services[primary_entity] = primary_node
return True
tasks/utils/migrate_meta_layout/yaml_io.py
+48
View File
@@ -0,0 +1,48 @@
from __future__ import annotations
import shutil
from pathlib import Path
from typing import Any, Dict
import yaml
def load(path: Path) -> Any:
if not path.is_file():
return None
text = path.read_text(encoding="utf-8")
if not text.strip():
return None
return yaml.safe_load(text) # noqa: direct-yaml — once-off migration script
def dump(path: Path, data: Any) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
text = yaml.safe_dump( # noqa: direct-yaml — once-off migration script
data if data is not None else {},
sort_keys=False,
default_flow_style=False,
allow_unicode=True,
width=4096,
)
path.write_text(text, encoding="utf-8")
def empty_dir(path: Path) -> None:
if not path.is_dir():
return
for child in path.iterdir():
if child.is_file():
child.unlink()
elif child.is_dir():
shutil.rmtree(child)
path.rmdir()
def deep_merge(base: Any, override: Any) -> Any:
if isinstance(base, dict) and isinstance(override, dict):
merged: Dict[str, Any] = dict(base)
for key, value in override.items():
merged[key] = deep_merge(merged[key], value) if key in merged else value
return merged
return override
tests/lint/repository/test_python_file_size.py
+109
View File
@@ -0,0 +1,109 @@
"""Lint: cap Python file length at 500 lines (production code only).
Every production ``.py`` file tracked by the repository (i.e. not
matched by ``.gitignore``) must be at most :data:`MAX_PY_LINES` lines
long. The threshold is a soft architectural cue — a file approaching
the cap is usually doing too many things and is a candidate for a
split.
Files under the top-level ``tests/`` directory are intentionally
excluded: tests cluster fixtures and parametrised cases by feature,
so length here reflects coverage breadth rather than design weight.
Per-file opt-out
----------------
A file may opt out of this cap by including the marker
:data:`NOCHECK_MARKER_REGEX` (matches ``#nocheck`` or ``# nocheck``)
inside the first :data:`NOCHECK_SCAN_LINES` lines — typically the
module docstring or a comment at the top. The opt-out is intentionally
visible at the top of the file so the cost of carrying long modules is
not silently buried.
The walker uses the SPOT helpers in :mod:`utils.cache.files` so the
lint runs in environments where ``.git/`` is not mounted (the
``make exec`` container) and stays consistent with other gitignore-
aware lint tests.
"""
from __future__ import annotations
import re
import unittest
from pathlib import Path
from utils.cache.base import PROJECT_ROOT
from utils.cache.files import iter_non_ignored_files
# A file at this length is generally too coupled / too scope-broad and
# should be split. Bumping this number requires a corresponding cleanup
# pass — do not raise it just to silence the linter.
MAX_PY_LINES: int = 500
# Number of leading lines to scan for the per-file opt-out marker. Kept
# small so the marker has to live near the top of the file (visible,
# not buried) and so the scan stays cheap.
NOCHECK_SCAN_LINES: int = 30
# Matches `#nocheck` and `# nocheck`. The trailing word boundary keeps
# unrelated tokens (`#nocheckpoint`, `#nochecksum`) from accidentally
# silencing the lint.
NOCHECK_MARKER_REGEX = re.compile(r"#\s*nocheck\b")
def _line_count(path: Path) -> int:
# Read as bytes + count newlines so the check works on files that
# accidentally ship with a non-UTF-8 byte (lint must not fail on a
# decode error before reporting the size violation).
with path.open("rb") as fh:
data = fh.read()
if not data:
return 0
n = data.count(b"\n")
# Match `wc -l` only when the file ends with a newline; a file that
# ends without a final newline still has that last "line".
return n if data.endswith(b"\n") else n + 1
def _has_nocheck_marker(path: Path) -> bool:
try:
with path.open("r", encoding="utf-8", errors="replace") as fh:
head_lines: list[str] = []
for idx, line in enumerate(fh):
if idx >= NOCHECK_SCAN_LINES:
break
head_lines.append(line)
except OSError:
return False
head = "".join(head_lines)
return bool(NOCHECK_MARKER_REGEX.search(head))
class TestPythonFileSize(unittest.TestCase):
def test_python_files_at_or_below_max_lines(self):
root = PROJECT_ROOT
offenders: list[tuple[str, int]] = []
for path_str in iter_non_ignored_files(extensions=(".py",), exclude_tests=True):
path = Path(path_str)
lines = _line_count(path)
if lines <= MAX_PY_LINES:
continue
if _has_nocheck_marker(path):
continue
rel = path.relative_to(root).as_posix()
offenders.append((rel, lines))
if offenders:
offenders.sort(key=lambda item: (-item[1], item[0]))
self.fail(
f"Python files exceeding {MAX_PY_LINES} lines "
f"({len(offenders)}):\n"
+ "\n".join(f" - {rel} ({lines} lines)" for rel, lines in offenders)
+ "\n\nSplit the file or move sections into focused modules. "
"Raising MAX_PY_LINES requires a corresponding cleanup pass."
)
if __name__ == "__main__":
unittest.main()
tests/lint/repository/test_test_files_location.py
+8 -71
View File
@@ -1,74 +1,12 @@
from __future__ import annotations
import fnmatch
import os
import unittest
from pathlib import Path
from utils.cache.base import PROJECT_ROOT
from utils.cache.files import iter_project_files
def repo_root() -> Path:
for candidate in Path(__file__).resolve().parents:
if (candidate / "pyproject.toml").is_file():
return candidate
raise AssertionError("Repository root not found from test path.")
def _load_gitignore_patterns(root: Path) -> list[str]:
gitignore = root / ".gitignore"
if not gitignore.exists():
return []
patterns: list[str] = []
for raw in gitignore.read_text(encoding="utf-8").splitlines():
line = raw.strip()
if not line or line.startswith("#"):
continue
if line.startswith("!"):
# Negation patterns (re-include) are not used in this repo's .gitignore.
# If someone adds one later, extending this helper is cheap; for now
# we ignore them so they don't accidentally widen the ignore set.
continue
patterns.append(line)
return patterns
def _is_ignored(rel_path: str, patterns: list[str]) -> bool:
"""Minimal gitignore matcher covering the shapes used in this repo.
Supports:
- Blank lines and `#` comments (filtered in `_load_gitignore_patterns`).
- Bare-basename patterns (no slash): match any path component by `fnmatch`.
- Path-qualified patterns (with slash): match from repo root via `fnmatch`,
including `prefix/*` shorthand for everything under `prefix/`, and plain
directory prefixes (`foo/bar` ignores anything under `foo/bar/`).
- Trailing `/` denotes directory-only; treated the same as the stripped
pattern because we only test against file paths here.
Not supported (intentionally): negation (`!`), `**` globstar, character
classes with locale semantics, or anchoring differences from
`fnmatch.fnmatch` (`*` matches across `/`). Extend if the repo's
.gitignore grows richer.
"""
parts = rel_path.split("/")
for raw in patterns:
pattern = raw.lstrip("/").rstrip("/")
if not pattern:
continue
if "/" in pattern:
if fnmatch.fnmatch(rel_path, pattern):
return True
if pattern.endswith("/*"):
prefix = pattern[:-2]
if rel_path.startswith(prefix + "/"):
return True
if rel_path.startswith(pattern + "/"):
return True
else:
for part in parts:
if fnmatch.fnmatch(part, pattern):
return True
return False
from utils.cache.gitignore import is_path_gitignored, load_gitignore_patterns
class TestTestFilesLocation(unittest.TestCase):
@@ -76,24 +14,23 @@ class TestTestFilesLocation(unittest.TestCase):
Test-linter that enforces every `test_*.py` file to live under the
top-level `tests/` directory.
Files matched by `.gitignore` are skipped. The matcher is implemented
inline (no `git` subprocess), so the lint works in environments where
Files matched by `.gitignore` are skipped via the SPOT helpers in
:mod:`utils.cache.files`, so the lint works in environments where
`.git/` is not mounted (e.g. the `make exec` container).
"""
def test_no_test_files_outside_tests_dir(self):
root = repo_root()
patterns = _load_gitignore_patterns(root)
root = PROJECT_ROOT
patterns = load_gitignore_patterns(str(root))
offenders: list[str] = []
for path_str in iter_project_files(extensions=(".py",)):
if os.path.basename(path_str).startswith("test_") is False:
continue
path = Path(path_str)
rel = path.relative_to(root).as_posix()
rel = Path(path_str).relative_to(root).as_posix()
if rel == "tests" or rel.startswith("tests/"):
continue
if _is_ignored(rel, patterns):
if is_path_gitignored(rel, patterns):
continue
offenders.append(rel)
utils/cache/__init__.py
Vendored
+7 -2
View File
@@ -12,7 +12,11 @@ Domain modules:
``load_yaml_any`` parser cache shared by everything that reads YAML
off-disk.
- ``utils.cache.files`` — process-wide project-tree walk +
file-content cache (``read_text``, ``iter_project_files``).
file-content cache (``read_text``, ``iter_project_files``,
``iter_non_ignored_files``).
- ``utils.cache.gitignore`` — cached ``.gitignore`` pattern loader
+ path matcher used by lint/integration tests that need a
"files git would track" view without spawning git.
Use the domain module directly. The package-level
``_reset_cache_for_tests`` clears every domain's caches plus the shared
@@ -27,7 +31,7 @@ def _reset_cache_for_tests() -> None:
"""Orchestrate per-domain cache resets plus the shared fingerprint
memo. Test fixtures rely on a single entry point so they can stay
agnostic of how the cache is partitioned across modules."""
from . import applications, base, domains, files, users
from . import applications, base, domains, files, gitignore, users
from . import yaml as _yaml_cache
applications._reset()
@@ -36,3 +40,4 @@ def _reset_cache_for_tests() -> None:
base._reset()
_yaml_cache._reset()
files._reset()
gitignore._reset()
utils/cache/files.py
Vendored
+40
View File
@@ -15,15 +15,26 @@ Usage:
...
`PROJECT_ROOT` is the repo root derived from this file's location.
Gitignore-filtered iteration
----------------------------
The matcher itself lives in :mod:`utils.cache.gitignore` (single SPOT for
``.gitignore`` parsing and matching). :func:`iter_non_ignored_files`
here composes the cached project walk with that matcher and is the
recommended helper for lint tests that want the "files git would track"
view without subprocessing.
"""
from __future__ import annotations
import os
from functools import lru_cache
from pathlib import Path
from typing import Iterable, Iterator, Tuple
from .base import PROJECT_ROOT
from .gitignore import is_path_gitignored, load_gitignore_patterns
# Directories never worth descending into during project-tree scans.
_DEFAULT_SKIP_DIRS: frozenset[str] = frozenset(
@@ -131,6 +142,35 @@ def iter_project_files_with_content(
yield path, content
def iter_non_ignored_files(
*,
extensions: Iterable[str] | None = None,
exclude_tests: bool = False,
exclude_dirs: Iterable[str] | None = None,
root: str | None = None,
) -> Iterator[str]:
"""Yield project files filtered through ``.gitignore``.
Combines :func:`iter_project_files` with :func:`is_path_gitignored`
so callers get the "files git would track" view without spawning a
git subprocess. ``root`` defaults to :data:`PROJECT_ROOT`.
A path is yielded iff it survives both the in-process skip-dir
pruning of :func:`iter_project_files` and the ``.gitignore`` match.
"""
base = Path(root) if root else PROJECT_ROOT
patterns = load_gitignore_patterns(str(base))
for path in iter_project_files(
extensions=extensions,
exclude_tests=exclude_tests,
exclude_dirs=exclude_dirs,
):
rel = os.path.relpath(path, str(base)).replace(os.sep, "/")
if is_path_gitignored(rel, patterns):
continue
yield path
def _reset() -> None:
"""Test-only helper: clear the project-walk and file-content caches.
utils/cache/gitignore.py
Vendored
+105
View File
@@ -0,0 +1,105 @@
"""Gitignore-aware path matching for project-tree scans.
SPOT for the fnmatch-based ``.gitignore``-style filter used by lint and
integration tests that need to walk the repository without spawning
``git``. Covers the shapes the repo's ``.gitignore`` actually uses
(basename patterns, path-qualified patterns, trailing-slash-as-directory)
and intentionally omits negation (``!``) and ``**`` globstar — adding
either would require a real pathspec implementation.
Public API:
* :func:`load_gitignore_patterns` — read non-empty, non-comment,
non-negated lines from ``<root>/.gitignore``. Cached process-wide.
* :func:`is_path_gitignored` — pure matcher: does a repo-relative
POSIX-style path match any of the supplied patterns?
The combined "yield project files filtered through ``.gitignore``"
helper lives in :mod:`utils.cache.files` (``iter_non_ignored_files``)
because it composes the cached project walk with this matcher.
"""
from __future__ import annotations
import fnmatch
from functools import lru_cache
from pathlib import Path
from typing import Iterable, Tuple
from .base import PROJECT_ROOT
@lru_cache(maxsize=4)
def load_gitignore_patterns(root: str | None = None) -> Tuple[str, ...]:
"""Load non-comment, non-blank, non-negated patterns from ``<root>/.gitignore``.
Returns a tuple (immutable, hashable for ``lru_cache``). Pass ``root``
as a string so the cache key is plainly hashable; defaults to
:data:`PROJECT_ROOT` when omitted.
Negation (``!``) and ``**`` globstar are intentionally not supported —
the repo's ``.gitignore`` does not use them, and supporting either
would require a real pathspec implementation. If a future
``.gitignore`` adds them, extend this helper rather than duplicating
the logic in callers.
"""
base = Path(root) if root else PROJECT_ROOT
gitignore = base / ".gitignore"
if not gitignore.is_file():
return ()
patterns: list[str] = []
for raw in gitignore.read_text(encoding="utf-8").splitlines():
line = raw.strip()
if not line or line.startswith("#") or line.startswith("!"):
continue
patterns.append(line)
return tuple(patterns)
def is_path_gitignored(rel_path: str, patterns: Iterable[str]) -> bool:
"""Return True iff ``rel_path`` (POSIX-style, repo-root-relative) is
matched by any of ``patterns``.
Supports the shapes the repo's ``.gitignore`` actually uses:
* Bare-basename patterns (no slash): match any path component via
``fnmatch``.
* Path-qualified patterns (with slash): match from the repo root via
``fnmatch``, plus ``prefix/*`` shorthand and bare-prefix directory
matching.
* Trailing ``/`` (directory-only): treated as the stripped pattern
because we only test against file paths.
Behaviour intentionally diverges from real git ``pathspec``: ``*``
here matches across ``/`` (``fnmatch`` semantics). If you need full
fidelity, use a dedicated pathspec library.
"""
parts = rel_path.split("/")
for raw in patterns:
pattern = raw.lstrip("/").rstrip("/")
if not pattern:
continue
if "/" in pattern:
if fnmatch.fnmatch(rel_path, pattern):
return True
if pattern.endswith("/*"):
prefix = pattern[:-2]
if rel_path.startswith(prefix + "/"):
return True
if rel_path.startswith(pattern + "/"):
return True
else:
for part in parts:
if fnmatch.fnmatch(part, pattern):
return True
return False
def _reset() -> None:
"""Test-only helper: clear the cached gitignore patterns.
Named ``_reset`` for parity with the per-domain reset helpers in
sibling modules; ``utils.cache._reset_cache_for_tests`` orchestrates
calls to all of them.
"""
load_gitignore_patterns.cache_clear()