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

"""auth-probe 路由测试：入站实体授权探测与过滤（进程内，无网络）。"""
from __future__ import annotations

import time

import pytest

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

import app as sim_app  # noqa: E402
from catalog import sample_adapters as sa  # noqa: E402

_CATALOG = {"objects": [
    {"objectCode": "ITEM", "apiInbound": {"supported": True,
                                           "entityCode": "MDM_ITEM",
                                           "contractVersion": "v1"}},
    {"objectCode": "CUSTOMER", "apiInbound": {"supported": True,
                                              "entityCode": "MDM_CUSTOMER",
                                              "contractVersion": "v1"}},
    {"objectCode": "NO_INBOUND", "apiInbound": {"supported": False,
                                                "entityCode": "MDM_X"}},
]}


class _FakeClient:
    def op_schema(self, entity, contract_version=None):
        return (403 if entity == "MDM_CUSTOMER" else 200), {"message": "ok"}, {}


def _reset_cache(monkeypatch):
    monkeypatch.setitem(sim_app._AUTH_PROBE_CACHE, "at", 0.0)
    monkeypatch.setitem(sim_app._AUTH_PROBE_CACHE, "data", None)


def _patch_reachable(monkeypatch):
    monkeypatch.setattr(sim_app.aidop_probe, "check_reachable",
                        lambda *a, **k: {"reachable": True})


def test_auth_probe_filters_unauthorized(monkeypatch):
    monkeypatch.setattr(sim_app, "_build_inbound_client", lambda: _FakeClient())
    monkeypatch.setattr(sa, "load_catalog", lambda: _CATALOG)
    _patch_reachable(monkeypatch)
    _reset_cache(monkeypatch)
    r = TestClient(sim_app.app).get("/api/inbound/auth-probe", params={"refresh": 1})
    assert r.status_code == 200
    body = r.json()
    assert body["cached"] is False
    assert body["entities"] == {"MDM_ITEM": True, "MDM_CUSTOMER": False}


def test_auth_probe_unreachable_marks_all_false_without_probing(monkeypatch):
    monkeypatch.setattr(sim_app, "_build_inbound_client",
                        lambda: (_ for _ in ()).throw(AssertionError("should not call")))
    monkeypatch.setattr(sa, "load_catalog", lambda: _CATALOG)
    monkeypatch.setattr(sim_app.aidop_probe, "check_reachable",
                        lambda *a, **k: {"reachable": False, "error": "down"})
    _reset_cache(monkeypatch)
    r = TestClient(sim_app.app).get("/api/inbound/auth-probe", params={"refresh": 1})
    assert r.status_code == 200
    body = r.json()
    assert body["entities"] == {"MDM_ITEM": False, "MDM_CUSTOMER": False}
    assert any("不可达" in e for e in body["probeErrors"])


def test_auth_probe_missing_credentials_400(monkeypatch):
    monkeypatch.setattr(sim_app, "_env", lambda name: "")
    _reset_cache(monkeypatch)
    r = TestClient(sim_app.app).get("/api/inbound/auth-probe", params={"refresh": 1})
    assert r.status_code == 400
    assert "AIDOP_SIM_INBOUND_ACCESS_KEY" in r.json()["detail"]


def test_auth_probe_serves_cache_without_client(monkeypatch):
    monkeypatch.setitem(sim_app._AUTH_PROBE_CACHE, "at", time.monotonic())
    monkeypatch.setitem(sim_app._AUTH_PROBE_CACHE, "data",
                        {"entities": {"MDM_ITEM": True}, "probedAt": "x", "probeErrors": []})
    monkeypatch.setattr(sim_app, "_build_inbound_client",
                        lambda: (_ for _ in ()).throw(AssertionError("should not call")))
    r = TestClient(sim_app.app).get("/api/inbound/auth-probe")
    assert r.status_code == 200
    assert r.json()["cached"] is True