119 lines
4.8 KiB
Python
119 lines
4.8 KiB
Python
"""Реестр модулей: состав, конфликты имён, границы импортов ядра."""
|
||
import ast
|
||
import re
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
|
||
from jira_hub.core import registry
|
||
from jira_hub.core.modules import CommandHooks, Module
|
||
|
||
CORE = Path(__file__).resolve().parent.parent / "src" / "jira_hub" / "core"
|
||
|
||
# временно: R-1, список только сокращается; после Task 6 пуст
|
||
PENDING: set[str] = set()
|
||
|
||
|
||
def _noop(*a, **k):
|
||
return None
|
||
|
||
|
||
def _hooks(types):
|
||
names = [f for f in CommandHooks.__dataclass_fields__ if f not in ("types", "read_types")]
|
||
return CommandHooks(types=frozenset(types), read_types=frozenset(), **{n: _noop for n in names})
|
||
|
||
|
||
def test_jira_registered():
|
||
assert [m.name for m in registry.modules()] == ["jira"]
|
||
assert registry.module("jira").name == "jira"
|
||
assert registry.module("nope") is None
|
||
|
||
|
||
def test_command_type_conflict():
|
||
a = Module(name="jira", commands=_hooks({"refresh"}))
|
||
b = Module(name="x", commands=_hooks({"refresh"}))
|
||
with pytest.raises(RuntimeError) as e:
|
||
registry.check([a, b], [], {})
|
||
assert "refresh" in str(e.value) and "jira" in str(e.value) and "x" in str(e.value)
|
||
|
||
|
||
def test_settings_conflict():
|
||
with pytest.raises(RuntimeError, match="port"):
|
||
registry.check([Module(name="x", settings_defaults={"port": 1})], [], {"port": 8790})
|
||
|
||
|
||
def test_route_conflict():
|
||
core = [("GET", r"^/projects$", _noop, "client")]
|
||
m = Module(name="x", routes=(("GET", r"^/projects$", _noop, "client"),))
|
||
with pytest.raises(RuntimeError, match="/projects"):
|
||
registry.check([m], core, {})
|
||
|
||
|
||
def test_routes_assembled():
|
||
from jira_hub.core import api
|
||
from jira_hub.modules import jira
|
||
pairs = lambda routes: {(m, p) for m, p, _, _ in routes} # noqa: E731
|
||
assert pairs(api.ROUTES) == pairs(api.CORE_ROUTES) | pairs(jira.MODULE.routes)
|
||
assert not pairs(api.CORE_ROUTES) & pairs(jira.MODULE.routes)
|
||
assert len(api.ROUTES) == len(api.CORE_ROUTES) + len(jira.MODULE.routes)
|
||
|
||
|
||
def test_module_of(monkeypatch):
|
||
a = Module(name="a", commands=_hooks({"t1"}))
|
||
monkeypatch.setattr(registry, "_cache", [a])
|
||
assert registry.module_of("t1") is a and registry.module_of("t2") is None
|
||
|
||
|
||
def _module_imports(tree, top_level_only):
|
||
"""Импорты пакета jira_hub.modules (абсолютные и относительные из core/)."""
|
||
found = []
|
||
for n in (tree.body if top_level_only else ast.walk(tree)):
|
||
if isinstance(n, ast.Import):
|
||
names = [a.name for a in n.names]
|
||
elif isinstance(n, ast.ImportFrom):
|
||
prefix = {0: "", 1: "jira_hub.core.", 2: "jira_hub."}[n.level]
|
||
base = (prefix + (n.module or "")).rstrip(".")
|
||
names = [base] + [base + "." + a.name for a in n.names]
|
||
else:
|
||
continue
|
||
if any(x == "jira_hub.modules" or x.startswith("jira_hub.modules.") for x in names):
|
||
found.append(n)
|
||
return found
|
||
|
||
|
||
def test_core_imports():
|
||
dirty = set()
|
||
for path in sorted(CORE.glob("*.py")):
|
||
tree = ast.parse(path.read_text(encoding="utf-8"))
|
||
if path.name in ("registry.py", "db.py"):
|
||
# только внутри функций: на верхнем уровне модуля импортов модулей нет
|
||
assert not _module_imports(tree, top_level_only=True), path.name
|
||
continue
|
||
if _module_imports(tree, top_level_only=False):
|
||
dirty.add(path.name)
|
||
assert dirty - PENDING == set(), f"ядро импортирует модули: {sorted(dirty - PENDING)}"
|
||
assert PENDING - dirty == set(), f"PENDING устарел, уберите: {sorted(PENDING - dirty)}"
|
||
|
||
|
||
# типы команд и термины jira; ensure_defaults и json.dumps(ensure_ascii=...) — легитимные имена ядра
|
||
JIRA_TERMS = re.compile(
|
||
r"""['"]refresh['"]|\bensure_(?!defaults\b|ascii\b)\w+|\bfetch_attachment\b|\bsync_run\b|\bissue_attachment\b"""
|
||
r"""|\battachment_file\b|\bissue_comment\b|\bjql\b|\w_jql\b|\bjql_\w+""",
|
||
re.IGNORECASE)
|
||
|
||
|
||
def test_core_has_no_jira_terms():
|
||
"""Ядро не знает типов команд и терминов jira; исключения — db.py (история миграций) и места с меткой долга."""
|
||
hits = []
|
||
for path in sorted(CORE.glob("*.py")):
|
||
if path.name == "db.py":
|
||
continue
|
||
lines = path.read_text(encoding="utf-8").splitlines()
|
||
for i, line in enumerate(lines):
|
||
if not JIRA_TERMS.search(line):
|
||
continue
|
||
if "долг hub-modules" in line or (i > 0 and "долг hub-modules" in lines[i - 1]):
|
||
continue
|
||
hits.append(f"{path.name}:{i + 1}: {line.strip()}")
|
||
assert hits == [], "термины jira в ядре:\n" + "\n".join(hits)
|