AiDOPWarehouse/tools/integration-simulator/tests/test_mock_api.py

"""Mock API（API_PULL 对端）行为守卫：cursor 过滤、鉴权模式、错误注入、响应路径、审计脱敏。

用 FastAPI TestClient 进程内验证，不占用端口、不需要 Ai-DOP 在线。
契约对齐 MdpApiPullExecutor.cs：GET 单次请求、data.list、dedup_key_path=bizKey、?cursor=。
"""
from __future__ import annotations

import base64

import pytest

fastapi_testclient = pytest.importorskip("fastapi.testclient")
TestClient = fastapi_testclient.TestClient

from providers import mock_api  # noqa: E402


@pytest.fixture()
def client():
    state = mock_api.MockApiState()
    state.settings.auth_mode = "NONE"
    return TestClient(mock_api.create_app(state)), state


def test_health_and_endpoints_listed(client):
    c, _ = client
    assert c.get("/health").json()["status"] == "UP"
    body = c.get("/__endpoints").json()
    assert "/api/item" in body["endpoints"]


def test_data_list_shape_and_dedup_key(client):
    c, _ = client
    body = c.get("/api/item").json()
    rows = body["data"]["list"]
    assert body["code"] == 0
    assert rows and all("bizKey" in r for r in rows)


def test_cursor_returns_only_greater_bizkeys(client):
    c, _ = client
    rows = c.get("/api/sales-order-entry").json()["data"]["list"]
    keys = [r["bizKey"] for r in rows]
    assert keys == sorted(keys)
    cut = keys[0]
    # cursor 是查询参数，必须 URL 编码：样例 bizKey 含 '#'（片段分隔符），
    # 直接拼接 URL 会被截断，故用 params= 让 httpx 正确编码。
    after = c.get("/api/sales-order-entry", params={"cursor": cut}).json()["data"]["list"]
    assert [r["bizKey"] for r in after] == [k for k in keys if k > cut]


def test_unknown_path_is_404(client):
    c, _ = client
    assert c.get("/api/not-registered-object").status_code == 404


def test_token_auth_rejects_and_accepts(client):
    c, state = client
    state.settings.auth_mode = "TOKEN"
    state.settings.token = "sim-token"
    assert c.get("/api/item").status_code == 401
    assert c.get("/api/item", headers={"Authorization": "Bearer wrong"}).status_code == 401
    assert c.get("/api/item", headers={"Authorization": "Bearer sim-token"}).status_code == 200


def test_basic_auth(client):
    c, state = client
    state.settings.auth_mode = "BASIC"
    state.settings.basic_user = "sim"
    state.settings.basic_password = "pw"
    token = base64.b64encode(b"sim:pw").decode()
    assert c.get("/api/item").status_code == 401
    assert c.get("/api/item", headers={"Authorization": f"Basic {token}"}).status_code == 200


def test_apikey_auth(client):
    c, state = client
    state.settings.auth_mode = "APIKEY"
    state.settings.api_key_header = "X-Api-Key"
    state.settings.api_key_value = "k1"
    assert c.get("/api/item").status_code == 401
    assert c.get("/api/item", headers={"X-Api-Key": "k1"}).status_code == 200


@pytest.mark.parametrize("status_code", [401, 403, 429, 500])
def test_forced_http_errors(client, status_code):
    c, state = client
    state.settings.force_status = status_code
    assert c.get("/api/item").status_code == status_code


def test_response_path_is_switchable(client):
    c, state = client
    state.settings.response_path = "result.rows"
    body = c.get("/api/item").json()
    assert isinstance(body["result"]["rows"], list)


def test_override_rows_win_and_can_be_cleared(client):
    c, state = client
    state.overrides["/api/item"] = [{"bizKey": "SIM-OVERRIDE-2"}, {"bizKey": "SIM-OVERRIDE-1"}]
    rows = c.get("/api/item").json()["data"]["list"]
    assert [r["bizKey"] for r in rows] == ["SIM-OVERRIDE-1", "SIM-OVERRIDE-2"]
    state.overrides.clear()
    assert c.get("/api/item").json()["data"]["list"]


def test_empty_result_is_still_success(client):
    c, state = client
    state.overrides["/api/item"] = []
    body = c.get("/api/item").json()
    assert body["data"]["list"] == []
    assert body["code"] == 0


def test_audit_redacts_authorization_and_token(client):
    c, state = client
    state.settings.auth_mode = "TOKEN"
    state.settings.token = "super-secret-token"
    c.get("/api/item", headers={"Authorization": "Bearer super-secret-token"})
    entries = state.audit
    assert entries
    dumped = str(entries)
    assert "super-secret-token" not in dumped
    assert entries[-1]["headers"]["authorization"] == "***REDACTED***"