AiDOPWarehouse/doc/db/mdp/mock_api/mock_server.py

"""
配置驱动的多端点 Mock（AIDOP 全模块双模式契约测试）。

契约与 mock_receipt_api.py 完全一致，严格对齐 MdpApiPullExecutor.cs：
- GET，单次请求，不分页；
- Bearer Token 鉴权（api_auth_type='TOKEN'，token 见环境变量 MOCK_TOKEN）；
- 响应数组位于 data.list（mdp_entity.response_data_path='data.list'）；
- 每行去重键 bizKey（mdp_entity.dedup_key_path='bizKey'）；
- 可选 ?cursor=<上批最后一行 bizKey>：仅返回 bizKey 大于该游标的行。

端点由 endpoints.json 配置：{ "/api/<obj>": "samples/<file>.json" }。
新增对象只需在 endpoints.json 加一行 + 放一份样本 JSON（数组，每行含 bizKey），无需改本脚本。

启动：
    pip install -r requirements.txt
    set MOCK_TOKEN=uat-mock-token          # bash: export MOCK_TOKEN=uat-mock-token
    uvicorn mock_server:app --host 127.0.0.1 --port 8018
"""
import json
import os
from pathlib import Path

from fastapi import FastAPI, Header, HTTPException, Query, Request

app = FastAPI(title="AIDOP Mock Multi-Endpoint API", version="1.0.0")

_BASE = Path(__file__).parent
_TOKEN = os.environ.get("MOCK_TOKEN", "uat-mock-token")


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


def _load_rows(rel_path: str) -> list[dict]:
    p = _BASE / 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 _check_auth(authorization: str | None) -> None:
    if authorization != f"Bearer {_TOKEN}":
        raise HTTPException(status_code=401, detail="invalid or missing bearer token")


@app.get("/__endpoints")
def list_endpoints():
    return {"token_env": "MOCK_TOKEN", "endpoints": _load_endpoints()}


@app.get("/health")
def health():
    return {"status": "UP"}


@app.get("/api/{obj:path}")
def get_object(
    obj: str,
    request: Request,
    cursor: str | None = Query(default=None, description="上批最后一行 bizKey（增量游标）"),
    authorization: str | None = Header(default=None),
):
    _check_auth(authorization)
    endpoints = _load_endpoints()
    path = request.url.path
    if path not in endpoints:
        raise HTTPException(status_code=404, detail=f"no mock for {path}; add it to endpoints.json")

    rows = _load_rows(endpoints[path])
    rows.sort(key=lambda r: str(r.get("bizKey", "")))
    if cursor:
        rows = [r for r in rows if str(r.get("bizKey", "")) > cursor]
    return {"code": 0, "message": "ok", "data": {"list": rows}}