"""Hexagonal boundary enforcement for the bot adapter (docs/ARCHITECTURE.md §4, §17). The bot must not import any core state/infra module or third-party driver it should not need: `core.db`, `core.s3`, `core.llm`, `core.mq`, `core.credits`, plus `sqlalchemy`, `asyncpg`, `minio`, `aio_pika`, `pymupdf`/`fitz`, `pytesseract`, `PIL`. Allowed: `core.logging`, `core.config` (type aliases only), aiogram, httpx, pydantic, structlog. A static AST scan is used (not an import probe) so the check still catches a regression even though those libs are installed in the local dev env — the bot Docker image does not ship them, so an accidental import would only explode in production. """ from __future__ import annotations import ast from pathlib import Path import pytest BOT_PKG = Path(__file__).resolve().parents[2] / "src" / "contract_check" / "bot" # Module names are rooted at `src/` so relative imports resolve to their full # `contract_check.bot.*` / `contract_check.core.*` dotted form. SRC_ROOT = BOT_PKG.parents[1] FORBIDDEN_PREFIXES = ( "contract_check.core.db", "contract_check.core.s3", "contract_check.core.llm", "contract_check.core.mq", "contract_check.core.credits", "sqlalchemy", "asyncpg", "alembic", "minio", "aio_pika", "aio-pika", "pymupdf", "fitz", "pytesseract", "PIL", ) # `core.config` is allowed (type aliases) but the bot must never *instantiate* # the infra `Settings` (which requires DB/MQ/S3 env vars). We assert the symbol # is not referenced by name. FORBIDDEN_NAMES = {"get_settings"} def _bot_files() -> list[Path]: return sorted(p for p in BOT_PKG.rglob("*.py") if p.is_file()) def _module_and_pkg(path: Path, src_root: Path = SRC_ROOT) -> tuple[str, str]: """Return (absolute module name, current package) for resolving relative imports.""" parts = path.relative_to(src_root).with_suffix("").parts if parts[-1] == "__init__": module_parts = list(parts[:-1]) current_pkg = ".".join(parts[:-1]) else: module_parts = list(parts) current_pkg = ".".join(parts[:-1]) return ".".join(module_parts), current_pkg def _resolve_relative(current_pkg: str, level: int, module: str | None) -> str: base_parts = current_pkg.split(".") if current_pkg else [] # Drop (level - 1) trailing components to find the base package. if level - 1 > 0: base_parts = base_parts[: len(base_parts) - (level - 1)] base = ".".join(base_parts) return f"{base}.{module}" if module else base def _imports_in(path: Path, src_root: Path = SRC_ROOT): tree = ast.parse(path.read_text(encoding="utf-8")) _, current_pkg = _module_and_pkg(path, src_root) for node in ast.walk(tree): if isinstance(node, ast.Import): for alias in node.names: yield alias.name, node.lineno elif isinstance(node, ast.ImportFrom): if node.level and node.level > 0: target = _resolve_relative(current_pkg, node.level, node.module) else: target = node.module or "" yield target, node.lineno def _forbidden_in(path: Path, src_root: Path = SRC_ROOT) -> list[str]: offenders = [] for target, lineno in _imports_in(path, src_root): for prefix in FORBIDDEN_PREFIXES: if target == prefix or target.startswith(prefix + "."): offenders.append(f"line {lineno}: imports {target!r}") return offenders def test_bot_package_exists() -> None: assert BOT_PKG.is_dir(), f"bot package not found at {BOT_PKG}" assert _bot_files(), "bot package has no .py files" def test_resolver_catches_forbidden_relative_import(tmp_path: Path) -> None: """The resolver must turn `from ..core.db import X` into `contract_check.core.db` so the forbidden-prefix match actually fires. Regression for an earlier bug where relative imports resolved without the `contract_check.` prefix.""" pkg = tmp_path / "contract_check" / "bot" pkg.mkdir(parents=True) (pkg / "__init__.py").write_text("", encoding="utf-8") bad = pkg / "leak.py" bad.write_text("from ..core.db import User\n", encoding="utf-8") offenders = _forbidden_in(bad, src_root=tmp_path) assert offenders, "expected forbidden import to be detected" assert "contract_check.core.db" in offenders[0] def test_resolver_allows_core_logging(tmp_path: Path) -> None: pkg = tmp_path / "contract_check" / "bot" pkg.mkdir(parents=True) (pkg / "__init__.py").write_text("", encoding="utf-8") good = pkg / "ok.py" good.write_text("from ..core.logging import get_logger\n", encoding="utf-8") assert _forbidden_in(good, src_root=tmp_path) == [] @pytest.mark.parametrize("path", _bot_files(), ids=lambda p: p.relative_to(BOT_PKG).as_posix()) def test_no_forbidden_imports(path: Path) -> None: offenders = _forbidden_in(path) assert not offenders, f"{path.relative_to(BOT_PKG)} violates boundary:\n " + "\n ".join( offenders ) @pytest.mark.parametrize("path", _bot_files(), ids=lambda p: p.relative_to(BOT_PKG).as_posix()) def test_does_not_instantiate_infra_settings(path: Path) -> None: tree = ast.parse(path.read_text(encoding="utf-8")) for node in ast.walk(tree): if isinstance(node, ast.Name) and node.id in FORBIDDEN_NAMES: pytest.fail( f"{path.relative_to(BOT_PKG)} line {node.lineno}: " f"references {node.id!r} (bot must use its own BotSettings)" ) def test_bot_imports_resolve() -> None: """The bot package imports cleanly with only the lean bot deps installed in dev.""" import importlib for mod in ( "contract_check.bot", "contract_check.bot.config", "contract_check.bot.client", "contract_check.bot.handlers", "contract_check.bot.webhook", ): importlib.import_module(mod)