generated from john/python-template
90 lines
3.3 KiB
Python
90 lines
3.3 KiB
Python
"""Structural rules for the services package.
|
|
|
|
The "Structure" section of `.github/instructions/services.instructions.md` requires
|
|
that service modules stay independent of one another. Shared behavior belongs in a
|
|
neutral module that defines no service class (`base.py`, `errors.py`, `registry.py`,
|
|
`source_media.py`, `media_storage.py`), and any operation that writes models owned by
|
|
two services belongs in an orchestration module.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import ast
|
|
from pathlib import Path
|
|
|
|
SERVICES_DIR = Path(__file__).resolve().parents[1] / "src" / "transcription" / "services"
|
|
|
|
# Modules that intentionally compose several services rather than owning one aggregate.
|
|
ORCHESTRATION_MODULES = frozenset({"store", "workflows", "__init__"})
|
|
|
|
|
|
def _module_paths() -> list[Path]:
|
|
return sorted(SERVICES_DIR.glob("*.py"))
|
|
|
|
|
|
def _defines_service_class(tree: ast.Module) -> bool:
|
|
return any(
|
|
isinstance(node, ast.ClassDef) and node.name.endswith("Service") and node.name != "RegistryService"
|
|
for node in tree.body
|
|
)
|
|
|
|
|
|
def _service_modules() -> dict[str, ast.Module]:
|
|
modules: dict[str, ast.Module] = {}
|
|
for path in _module_paths():
|
|
if path.stem in ORCHESTRATION_MODULES:
|
|
continue
|
|
tree = ast.parse(path.read_text(encoding="utf-8"))
|
|
if _defines_service_class(tree):
|
|
modules[path.stem] = tree
|
|
return modules
|
|
|
|
|
|
def _imported_sibling_modules(tree: ast.Module) -> set[str]:
|
|
imported: set[str] = set()
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.ImportFrom) and node.level == 1 and node.module:
|
|
imported.add(node.module.split(".")[0])
|
|
elif isinstance(node, ast.ImportFrom) and node.level == 0 and node.module:
|
|
parts = node.module.split(".")
|
|
if parts[:2] == ["transcription", "services"] and len(parts) > 2:
|
|
imported.add(parts[2])
|
|
return imported
|
|
|
|
|
|
def test_service_modules_are_discovered():
|
|
"""Guard the guard: the rule below is meaningless if nothing is scanned."""
|
|
assert set(_service_modules()) >= {"documents", "jobs", "people", "sources"}
|
|
|
|
|
|
def test_no_service_module_imports_another_service_module():
|
|
"""MED-14: a service module must not depend on a sibling service module."""
|
|
modules = _service_modules()
|
|
violations = {
|
|
name: sorted(_imported_sibling_modules(tree) & set(modules) - {name}) for name, tree in modules.items()
|
|
}
|
|
assert {name: found for name, found in violations.items() if found} == {}
|
|
|
|
|
|
def _foreign_session_scope_accesses(tree: ast.Module) -> list[int]:
|
|
lines: list[int] = []
|
|
for node in ast.walk(tree):
|
|
if isinstance(node, ast.Attribute) and node.attr == "_session_scope":
|
|
if isinstance(node.value, ast.Name) and node.value.id == "self":
|
|
continue
|
|
lines.append(node.lineno)
|
|
return sorted(lines)
|
|
|
|
|
|
def test_session_scope_is_not_accessed_via_other_services():
|
|
"""P4-1: orchestration must use a shared unit-of-work entry point, not private service scopes."""
|
|
violations: dict[str, list[int]] = {}
|
|
for path in _module_paths():
|
|
if path.stem == "base":
|
|
continue
|
|
tree = ast.parse(path.read_text(encoding="utf-8"))
|
|
found = _foreign_session_scope_accesses(tree)
|
|
if found:
|
|
violations[path.name] = found
|
|
assert violations == {}
|