group_xinghuo_jinrong/tests/test_module_boundary.py

"""模块边界防呆测试（配合《风控Agent模块边界与合并接缝标注.md》§3）。

风控 Agent 按"独立封装模块"自治：模块私有实现自己维护，与宿主（main）的耦合
只走 4 个接缝。本文件把这层约定写成断言——**AL-09 合并 main 后必须全绿**，
任何一条红了都说明模块被宿主侵蚀或接缝被破坏，应先修边界再继续。

四类断言：
1. 模块私有文件未被误删；
2. 模块代码不得 import 宿主私有实现（防双套串味）；
3. 模块 AuthContext 契约未被宿主类替换；
4. 模块私有 settings 字段未被合并丢掉。
"""

from __future__ import annotations

import re
from pathlib import Path

import pytest

from app.api.deps import AuthContext
from app.config.settings import settings
from app.utils.authz import AGENT_TYPES

APP_DIR = Path(__file__).resolve().parents[1] / "app"

# 模块私有关键文件（A 类）：缺失即说明被误删或合并时被宿主覆盖
MODULE_PRIVATE_FILES = [
    "app/api/deps.py",
    "app/api/risk.py",
    "app/api/simulate.py",
    "app/service/auth_service.py",
    "app/service/input_guard.py",
    "app/service/suitability.py",
    "app/service/tool_service.py",
    "app/repository/core_ro.py",
    "app/repository/risk_repository.py",
    "app/repository/session_repository.py",
    "app/gateway/trade_gateway.py",
    "app/model/suitability.py",
    "app/utils/db.py",
    "app/utils/trace.py",
    "app/api/audit_middleware.py",
    "app/service/risk/engine.py",
    "app/service/risk/alert_service.py",
]

# 禁止模块代码导入的宿主私有实现（D 类 / B 类宿主侧）
FORBIDDEN_IMPORT_PREFIXES = (
    "from app.gateway.auth_deps",
    "from app.gateway.jwt_service",
    "from app.gateway.rbac",
    "from app.gateway.ownership",
    "from app.config.database",
    "from app.middleware",
    "from app.utils.input_guard",  # 模块用 app.service.input_guard（含限流与留痕）
    "from app.model.schemas",  # 宿主 AuthContext，须经 app.api.auth_adapter 转换
)

# 模块私有 settings 字段（合并时一个都不能丢）
MODULE_SETTINGS_FIELDS = (
    "mysql_core_database",
    "redis_url",
    "milvus_uri",
    "ollama_base_url",
    "embed_model",
    "embed_dim",
    "embed_timeout_seconds",
    "deepseek_api_key",
    "deepseek_base_url",
    "risk_large_amount",
    "risk_daily_total",
    "risk_freq_count",
    "risk_probe_window_minutes",
    "risk_probe_count",
    "risk_probe_amount",
    "risk_small_amount",
    "risk_small_count",
    "risk_aml_default_threshold",
    "guard_rate_limit_max",
    "guard_rate_limit_window_seconds",
)

_IMPORT_LINE = re.compile(r"^\s*(from|import)\s+")


def _iter_module_py_files():
    for path in APP_DIR.rglob("*.py"):
        if "__pycache__" in path.parts:
            continue
        yield path


@pytest.mark.parametrize("rel_path", MODULE_PRIVATE_FILES)
def test_module_private_file_exists(rel_path):
    """模块私有文件必须存在（防合并时被删除/覆盖）。"""
    assert (APP_DIR.parent / rel_path).is_file(), f"模块私有文件缺失：{rel_path}"


def test_no_host_private_imports():
    """模块代码不得直接 import 宿主私有实现——跨层一律走接缝。"""
    offenders: list[str] = []
    for path in _iter_module_py_files():
        for lineno, line in enumerate(
            path.read_text(encoding="utf-8", errors="ignore").splitlines(), start=1
        ):
            stripped = line.strip()
            if not _IMPORT_LINE.match(stripped):
                continue  # 只看真正的 import 行，避免注释/文档字符串误报
            if stripped.startswith(FORBIDDEN_IMPORT_PREFIXES):
                rel = path.relative_to(APP_DIR.parent).as_posix()
                offenders.append(f"{rel}:{lineno} -> {stripped}")

    assert not offenders, "发现跨层导入宿主私有实现（应改走接缝）：\n" + "\n".join(offenders)


def test_auth_context_contract_intact():
    """模块 AuthContext 契约必须完整——防被宿主的 schemas.AuthContext 替换。

    宿主用 sub/trace_id/agent_type、has_role 单参、has_perm；
    模块用 actor_id、has_role 多参、has_permission。字段名或方法名一变，
    模块内全量引用会静默失效，故在此锁死。
    """
    for field in (
        "actor_id",
        "roles",
        "customer_id",
        "token_type",
        "permissions",
        "tenant_id",
        "jti",
    ):
        assert field in AuthContext.model_fields, f"AuthContext 缺失字段：{field}"

    ctx = AuthContext(actor_id="STAFF-90001", roles=["risk_officer", "risk_demo"])
    assert ctx.has_role("risk_officer", "compliance") is True  # 多参语义
    assert ctx.has_role("advisor") is False
    assert ctx.has_permission("risk:alert:write") is False
    assert ctx.is_customer() is False


def test_module_settings_fields_present():
    """模块私有配置字段必须齐全（防合并 settings.py 时被丢）。"""
    missing = [f for f in MODULE_SETTINGS_FIELDS if not hasattr(settings, f)]
    assert not missing, f"settings 丢失模块私有字段：{missing}"


def test_agent_types_contract():
    """Agent 类型四值须与宿主一致（这是少数双方天然对齐的契约，不得漂移）。"""
    assert AGENT_TYPES == ("customer", "advisor", "analyst", "risk")