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  # 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/429/500 等
    response_path: str = "data.list"
    audit_limit: int = 200


@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 _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("/__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,
        }

    @app.post("/__control/settings")
    def update_settings(payload: dict):
        s = state.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"])
        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
        if state.settings.delay_seconds > 0:
            time.sleep(state.settings.delay_seconds)
        try:
            if state.settings.force_status:
                raise HTTPException(status_code=state.settings.force_status,
                                    detail=f"forced error {state.settings.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])
            rows.sort(key=lambda r: str(r.get("bizKey", "")))
            total = len(rows)
            if cursor:
                rows = [r for r in rows if str(r.get("bizKey", "")) > cursor]
            return _build_response(rows, state.settings.response_path)
        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,
                "elapsedMs": int((time.time() - started) * 1000),
                "headers": redact_headers(dict(request.headers), state.secrets()),
            })

    return app