"""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