AiDOPWarehouse/tools/integration-simulator/providers/mock_api.py

"""API_PULL 对端 Mock（任务书 P1-D）。

契约严格对齐 MdpApiPullExecutor.cs 与既有 doc/db/mdp/mock_api/mock_server.py：
- GET 单次请求，不分页；响应数组位于 data.list（可切换 responsePath）；
- 每行去重键 bizKey（dedup_key_path）；?cursor=<上批最后一行 bizKey> 只返回更大行；
- 鉴权模式 NONE / TOKEN(Bearer) / BASIC / APIKEY，对齐执行器 ApplyAuth 分支；
- 可注入延迟与强制 HTTP 错误（401/403/429/500）；
- 端点复用 doc/db/mdp/mock_api/endpoints.json 与 samples/*.json（不迁移、不修改），
  另支持运行时 override：把模拟器适配后的样例（带 SIM 前缀）挂到任意路径。

每次调用写入内存审计（Header 脱敏），Secret 值以 redact_text 二次遮蔽。
"""
from __future__ import annotations

import base64
import json
import sys
import time
from dataclasses import dataclass, field
from pathlib import Path

_SIM_ROOT = Path(__file__).resolve().parents[1]
if str(_SIM_ROOT) not in sys.path:
    sys.path.insert(0, str(_SIM_ROOT))

from fastapi import FastAPI, HTTPException, Query, Request, Response  # noqa: E402

from catalog.sample_adapters import MOCK_API_DIR  # noqa: E402
from clients.redaction import redact_headers  # noqa: E402


@dataclass
class MockApiSettings:
    auth_mode: str = "TOKEN"          # NONE | TOKEN | BASIC | APIKEY
    token: str = "sim-mock-token"
    basic_user: str = "sim"
    basic_password: str = "sim-basic-password"
    api_key_header: str = "X-Api-Key"
    api_key_value: str = "sim-api-key"
    delay_seconds: float = 0.0
    force_status: int | None = None   # 401/403/404/429/500/503 等
    response_path: str = "data.list"
    audit_limit: int = 200
    # ---- 第二阶段 N2 异常模式（内存态，恢复默认即清除）----
    timeout_seconds: float = 0.0      # >0 时按秒数睡眠，模拟超过 Ai-DOP 超时
    empty_response: bool = False      # 返回 200 空 body
    invalid_json: bool = False        # 返回非法 JSON 文本
    content_type: str | None = None   # 覆盖响应 Content-Type（如 text/html）
    ignore_cursor: bool = False       # 无视 cursor 返回全部行（cursor 不推进）
    no_sort: bool = False             # 不按 bizKey 排序（cursor 乱序）

    def reset_faults(self) -> None:
        """恢复默认：清空全部异常注入设置（鉴权/凭据类保留）。"""
        self.auth_mode = "TOKEN"
        self.delay_seconds = 0.0
        self.force_status = None
        self.response_path = "data.list"
        self.timeout_seconds = 0.0
        self.empty_response = False
        self.invalid_json = False
        self.content_type = None
        self.ignore_cursor = False
        self.no_sort = False


@dataclass
class MockApiState:
    settings: MockApiSettings = field(default_factory=MockApiSettings)
    overrides: dict[str, list[dict]] = field(default_factory=dict)
    audit: list[dict] = field(default_factory=list)

    def secrets(self) -> list[str]:
        s = self.settings
        return [v for v in (s.token, s.basic_password, s.api_key_value) if v]


def _load_endpoints() -> dict[str, str]:
    with (MOCK_API_DIR / "endpoints.json").open("r", encoding="utf-8") as f:
        return json.load(f)


def _load_rows(rel_path: str) -> list[dict]:
    p = MOCK_API_DIR / rel_path
    if not p.exists():
        raise HTTPException(status_code=404, detail=f"sample not found: {rel_path}")
    with p.open("r", encoding="utf-8") as f:
        return json.load(f)


def _build_response(rows: list[dict], response_path: str) -> dict:
    parts = [p for p in response_path.split(".") if p]
    body: dict = {}
    node = body
    for part in parts[:-1]:
        node[part] = {}
        node = node[part]
    node[parts[-1]] = rows if parts else rows
    body.setdefault("code", 0)
    body.setdefault("message", "ok")
    return body


def _check_auth(state: MockApiState, request: Request) -> None:
    s = state.settings
    mode = (s.auth_mode or "NONE").upper()
    if mode == "NONE":
        return
    authorization = request.headers.get("Authorization")
    if mode in ("TOKEN", "OAUTH2"):
        if authorization != f"Bearer {s.token}":
            raise HTTPException(status_code=401, detail="invalid or missing bearer token")
    elif mode == "BASIC":
        expected = "Basic " + base64.b64encode(f"{s.basic_user}:{s.basic_password}".encode()).decode()
        if authorization != expected:
            raise HTTPException(status_code=401, detail="invalid or missing basic credentials")
    elif mode == "APIKEY":
        if request.headers.get(s.api_key_header) != s.api_key_value:
            raise HTTPException(status_code=401, detail=f"invalid or missing api key header {s.api_key_header}")
    else:
        raise HTTPException(status_code=401, detail=f"unsupported auth mode on mock side: {mode}")


def _active_faults(s: MockApiSettings) -> list[str]:
    """当前生效的异常注入清单（写入审计，便于人工核对「选择的异常」）。"""
    faults: list[str] = []
    if s.timeout_seconds > 0:
        faults.append(f"timeout:{s.timeout_seconds}s")
    if s.empty_response:
        faults.append("emptyResponse")
    if s.invalid_json:
        faults.append("invalidJson")
    if s.content_type:
        faults.append(f"contentType:{s.content_type}")
    if s.ignore_cursor:
        faults.append("ignoreCursor")
    if s.no_sort:
        faults.append("noSort")
    return faults


def _audit(state: MockApiState, entry: dict) -> None:
    state.audit.append(entry)
    if len(state.audit) > state.settings.audit_limit:
        del state.audit[: len(state.audit) - state.settings.audit_limit]


def create_app(state: MockApiState | None = None) -> FastAPI:
    state = state or MockApiState()
    app = FastAPI(title="Ai-DOP Integration Simulator - Mock API (API_PULL)", version="1.0.0")
    app.state.sim = state

    @app.get("/health")
    def health():
        return {"status": "UP", "authMode": state.settings.auth_mode}

    @app.get("/")
    def root():
        # Ai-DOP 数据源测连对 api_base_url 发起 GET 根路径，返回 200 使测连显示成功
        return {"status": "UP", "name": "Ai-DOP Integration Simulator Mock API",
                "authMode": state.settings.auth_mode}

    @app.get("/__endpoints")
    def list_endpoints():
        return {
            "authMode": state.settings.auth_mode,
            "responsePath": state.settings.response_path,
            "endpoints": _load_endpoints(),
            "overrides": sorted(state.overrides.keys()),
        }

    @app.get("/__control/settings")
    def get_settings():
        s = state.settings
        # 不回显任何凭据值
        return {
            "authMode": s.auth_mode, "delaySeconds": s.delay_seconds,
            "forceStatus": s.force_status, "responsePath": s.response_path,
            "apiKeyHeader": s.api_key_header,
            "timeoutSeconds": s.timeout_seconds, "emptyResponse": s.empty_response,
            "invalidJson": s.invalid_json, "contentType": s.content_type,
            "ignoreCursor": s.ignore_cursor, "noSort": s.no_sort,
        }

    @app.post("/__control/settings")
    def update_settings(payload: dict):
        s = state.settings
        if payload.get("reset"):
            s.reset_faults()
            return get_settings()
        if "auth_mode" in payload:
            mode = str(payload["auth_mode"]).upper()
            if mode not in ("NONE", "TOKEN", "BASIC", "APIKEY", "OAUTH2"):
                raise HTTPException(status_code=400, detail=f"unknown auth_mode: {mode}")
            s.auth_mode = mode
        if "token" in payload and payload["token"]:
            s.token = str(payload["token"])
        if "basic_user" in payload and payload["basic_user"]:
            s.basic_user = str(payload["basic_user"])
        if "basic_password" in payload and payload["basic_password"]:
            s.basic_password = str(payload["basic_password"])
        if "api_key_header" in payload and payload["api_key_header"]:
            s.api_key_header = str(payload["api_key_header"])
        if "api_key_value" in payload and payload["api_key_value"]:
            s.api_key_value = str(payload["api_key_value"])
        if "delay_seconds" in payload:
            s.delay_seconds = max(0.0, float(payload["delay_seconds"]))
        if "force_status" in payload:
            s.force_status = int(payload["force_status"]) if payload["force_status"] else None
        if "response_path" in payload and payload["response_path"]:
            s.response_path = str(payload["response_path"])
        if "timeout_seconds" in payload:
            s.timeout_seconds = max(0.0, float(payload["timeout_seconds"]))
        if "empty_response" in payload:
            s.empty_response = bool(payload["empty_response"])
        if "invalid_json" in payload:
            s.invalid_json = bool(payload["invalid_json"])
        if "content_type" in payload:
            s.content_type = str(payload["content_type"]) if payload["content_type"] else None
        if "ignore_cursor" in payload:
            s.ignore_cursor = bool(payload["ignore_cursor"])
        if "no_sort" in payload:
            s.no_sort = bool(payload["no_sort"])
        return get_settings()

    @app.post("/__control/overrides")
    def set_override(payload: dict):
        path = str(payload.get("path") or "")
        rows = payload.get("rows")
        if not path.startswith("/api/") or not isinstance(rows, list):
            raise HTTPException(status_code=400, detail="need {path:/api/..., rows:[...]}")
        state.overrides[path] = rows
        return {"path": path, "rowCount": len(rows)}

    @app.delete("/__control/overrides")
    def clear_overrides():
        state.overrides.clear()
        return {"cleared": True}

    @app.get("/__control/audit")
    def get_audit():
        return {"entries": state.audit[-50:]}

    @app.get("/api/{obj:path}")
    def get_object(
        obj: str,
        request: Request,
        cursor: str | None = Query(default=None, description="上批最后一行 bizKey（增量游标）"),
    ):
        started = time.time()
        path = request.url.path
        s = state.settings
        try:
            if s.timeout_seconds > 0:
                # 超时模拟：睡眠超过 Ai-DOP 读取超时（由测试人员配置秒数）
                time.sleep(s.timeout_seconds)
            if s.empty_response:
                return Response(content="", media_type=s.content_type or "application/json")
            if s.invalid_json:
                return Response(content='{"data": {"list": [',
                                media_type=s.content_type or "application/json")
            if s.delay_seconds > 0:
                time.sleep(s.delay_seconds)
            if s.force_status:
                raise HTTPException(status_code=s.force_status,
                                    detail=f"forced error {s.force_status}")
            _check_auth(state, request)
            if path in state.overrides:
                rows = [dict(r) for r in state.overrides[path]]
            else:
                endpoints = _load_endpoints()
                if path not in endpoints:
                    raise HTTPException(status_code=404,
                                        detail=f"no mock for {path}; add it to endpoints.json or set an override")
                rows = _load_rows(endpoints[path])
            if not s.no_sort:
                rows.sort(key=lambda r: str(r.get("bizKey", "")))
            if cursor and not s.ignore_cursor:
                rows = [r for r in rows if str(r.get("bizKey", "")) > cursor]
            body = _build_response(rows, s.response_path)
            if s.content_type:
                return Response(content=json.dumps(body, ensure_ascii=False),
                                media_type=s.content_type)
            return body
        finally:
            _audit(state, {
                "time": time.strftime("%Y-%m-%dT%H:%M:%S%z"),
                "path": path,
                "cursor": cursor,
                "authMode": state.settings.auth_mode,
                "forceStatus": state.settings.force_status,
                "faults": _active_faults(state.settings),
                "elapsedMs": int((time.time() - started) * 1000),
                "headers": redact_headers(dict(request.headers), state.secrets()),
            })

    return app