DealDocumentScreening/tests/unit/test_bot_boundary.py

"""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",
    ):
        importlib.import_module(mod)