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 random
import sys
import threading
import time
from collections import deque
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 乱序）
    # ---- 第三阶段 P2 压力模式（受控：须压力计划确认后由控制台注入）----
    load_mode: bool = False           # 压力模式开关（§19 实时停止/恢复正常）
    page_size: int = 0                # >0: 每次仅返回前 N 行（配置每次返回行数）
    total_rows_limit: int = 0         # >0: 响应数据总量上限（模拟大表截断）
    random_delay: float = 0.0         # >0: 每请求随机延迟 0..random_delay 秒
    error_rate: float = 0.0           # 0..1: 随机 500 比例（间歇 5xx）
    rate_limit: float = 0.0           # >0: 每秒最大请求数，超出返回 429（限流阈值）

    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
        self.load_mode = False
        self.page_size = 0
        self.total_rows_limit = 0
        self.random_delay = 0.0
        self.error_rate = 0.0
        self.rate_limit = 0.0


@dataclass
class MockApiLoadStats:
    """P2 压力模式请求统计（§19：统计收到的请求量、并发、QPS、耗时和错误）。"""
    requests: int = 0
    active: int = 0                 # 当前并发
    peak_active: int = 0            # 峰值并发
    status_counts: dict = field(default_factory=dict)
    latencies: list = field(default_factory=list)
    rate_limited: int = 0           # 429 次数
    recent_times: deque = field(default_factory=lambda: deque(maxlen=500))
    _lock: threading.Lock = field(default_factory=threading.Lock)

    def begin(self):
        with self._lock:
            self.active += 1
            self.peak_active = max(self.peak_active, self.active)

    def end(self, status: int, elapsed_s: float):
        with self._lock:
            self.active = max(0, self.active - 1)
            self.requests += 1
            self.status_counts[status] = self.status_counts.get(status, 0) + 1
            self.latencies.append(elapsed_s)
            self.recent_times.append(time.monotonic())
            if status == 429:
                self.rate_limited += 1

    def snapshot(self) -> dict:
        with self._lock:
            latencies = sorted(self.latencies) if self.latencies else []
            times = list(self.recent_times)
            qps = 0.0
            if len(times) >= 2:
                span = times[-1] - times[0]
                qps = round(len(times) / span, 2) if span > 0 else 0.0
            avg = round(sum(latencies) / len(latencies) * 1000, 1) if latencies else 0.0
            return {"requests": self.requests, "active": self.active,
                    "peakConcurrency": self.peak_active,
                    "qpsWindow": qps, "avgMs": avg,
                    "maxMs": round(max(latencies) * 1000, 1) if latencies else 0.0,
                    "statusCounts": dict(sorted(self.status_counts.items())),
                    "rateLimited": self.rate_limited}

    def reset(self):
        with self._lock:
            self.requests = 0
            self.active = 0
            self.peak_active = 0
            self.status_counts.clear()
            self.latencies.clear()
            self.rate_limited = 0
            self.recent_times.clear()


@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)
    load_stats: MockApiLoadStats = field(default_factory=MockApiLoadStats)
    rate_hits: deque = field(default_factory=deque)   # P2 限流滑动窗口

    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 _rate_allowed(state: MockApiState, rps: float) -> bool:
    """滑动窗口限流：每秒最多 rps 个请求，超出返回 False（间歇 429，§19）。"""
    now = time.monotonic()
    while state.rate_hits and state.rate_hits[0] <= now - 1.0:
        state.rate_hits.popleft()
    if len(state.rate_hits) >= rps:
        return False
    state.rate_hits.append(now)
    return True


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,
            "loadMode": s.load_mode, "pageSize": s.page_size,
            "totalRowsLimit": s.total_rows_limit, "randomDelay": s.random_delay,
            "errorRate": s.error_rate, "rateLimit": s.rate_limit,
        }

    @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"])
        if "load_mode" in payload:
            s.load_mode = bool(payload["load_mode"])
        if "page_size" in payload:
            s.page_size = max(0, int(payload["page_size"]))
        if "total_rows_limit" in payload:
            s.total_rows_limit = max(0, int(payload["total_rows_limit"]))
        if "random_delay" in payload:
            s.random_delay = max(0.0, float(payload["random_delay"]))
        if "error_rate" in payload:
            s.error_rate = min(1.0, max(0.0, float(payload["error_rate"])))
        if "rate_limit" in payload:
            s.rate_limit = max(0.0, float(payload["rate_limit"]))
        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("/__control/load-stats")
    def get_load_stats():
        """P2 压力模式统计快照（§19：请求量、并发、QPS、耗时和错误）。"""
        return {"loadMode": state.settings.load_mode, "stats": state.load_stats.snapshot()}

    @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
        status_code = 200
        load_on = s.load_mode
        if load_on:
            state.load_stats.begin()
        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.random_delay > 0:
                # P2 随机延迟（固定/随机延迟测试类型）
                time.sleep(random.uniform(0, s.random_delay))
            if s.force_status:
                raise HTTPException(status_code=s.force_status,
                                    detail=f"forced error {s.force_status}")
            if load_on and s.rate_limit > 0 and not _rate_allowed(state, s.rate_limit):
                raise HTTPException(status_code=429,
                                    detail=f"mock load rate limit {s.rate_limit}/s exceeded")
            if load_on and s.error_rate > 0 and random.random() < s.error_rate:
                raise HTTPException(status_code=500, detail="mock load error injection")
            _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]
            if load_on and s.total_rows_limit > 0:
                rows = rows[:s.total_rows_limit]
            if load_on and s.page_size > 0:
                rows = rows[:s.page_size]
            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
        except HTTPException as ex:
            status_code = ex.status_code
            raise
        finally:
            elapsed = time.time() - started
            if load_on:
                state.load_stats.end(status_code, elapsed)
            _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),
                "loadMode": load_on,
                "httpStatus": status_code,
                "elapsedMs": int(elapsed * 1000),
                "headers": redact_headers(dict(request.headers), state.secrets()),
            })

    return app