95 lines
3.7 KiB
Python
95 lines
3.7 KiB
Python
"""Реестр модулей: состав, конфликты имён, границы импортов ядра."""
|
|
import ast
|
|
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)}"
|