86 lines
3.3 KiB
Python
86 lines
3.3 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 4–6
|
||
PENDING = {"api.py", "cli.py", "mcp_server.py", "queue.py"}
|
||
|
||
|
||
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_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)}"
|