ai-kefu-admin/tests/test_push_test_endpoint.py

"""POST /api/platform/tenants/{slug}/push/test(Phase 3 Task 7c)——推送连通性探测。

端点红线: **不发任何线索数据**——探测 body = {"messages": []}(空批量,
与 lead_push_service._do_post 同端点同头;UniService 是批量结构,空列表
对方案应为无操作)。测试断言钉死这一形态。

响应契约: {reachable, auth_ok(可空), status, reason, error}——
- 传输层异常 → reachable=false,reason 分类 dns/timeout/connect;
- HTTP 有响应 → reachable=true + status;2xx → auth_ok=true,
  401/403 → false,其余(4xx/5xx)→ null(到达但状态异常,鉴权不定);
- 单次探测不重试(与 _run_push 的重试矩阵无关)。

测试纪律: monkeypatch config_routes.httpx.AsyncClient(假客户端),绝不真连。
"""
import socket

import httpx

from app.engine.config import settings
from app.platform.routes import config_routes
from tests.test_tenant_admin_api import _auth, _bearer, _create, _tenant_admin_token

CONFIG = "/api/platform/tenants/{slug}/config"
PUSH_TEST = "/api/platform/tenants/{slug}/push/test"

PUSH_GROUP = {
    "enabled": True,
    "url": "https://push.example/open/v1/inquiry/push",
    "api_key": "k-t1",
    "prefix": "demo-p",
    "timeout": 5.5,
    "max_attempts": 3,
    "retry_backoff_base": 1,
}


def _install_fake_client(monkeypatch, handler):
    """config_routes.httpx.AsyncClient 桩: handler(call) → httpx.Response,
    抛异常照抛。返回调用记录(init 构造参数 + 每次 post 的 url/json/headers)。"""
    calls = []

    class FakeClient:
        def __init__(self, **kwargs):
            calls.append({"init": kwargs})

        async def __aenter__(self):
            return self

        async def __aexit__(self, *exc_info):
            return False

        async def post(self, url, json=None, headers=None):
            call = {"url": url, "json": json, "headers": dict(headers or {})}
            calls.append(call)
            return handler(call)

    monkeypatch.setattr(config_routes.httpx, "AsyncClient", FakeClient)
    return calls


def _ok(call):
    return httpx.Response(200, json={"success": True, "data": {
        "accepted": 0, "duplicated": 0, "failed": []}})


def _put_push_group(client, slug, payload):
    return client.put(f"{CONFIG.format(slug=slug)}/push",
                      headers=_auth(client), json=payload)


def _post_test(client, slug, headers=None):
    return client.post(PUSH_TEST.format(slug=slug), headers=headers or _auth(client))


def _post_calls(calls):
    return [c for c in calls if "url" in c]


def test_push_test_probe_shape_and_ok(super_client, monkeypatch):
    """正常路径: 200 → reachable+auth_ok;请求形态逐项钉死——同 _do_post 的
    端点/X-Api-Key 头/超时/follow_redirects,body 恒为空 messages(红线)。"""
    _create(super_client, slug="demo")
    assert _put_push_group(super_client, "demo", PUSH_GROUP).status_code == 200
    calls = _install_fake_client(monkeypatch, _ok)

    r = _post_test(super_client, "demo")
    assert r.status_code == 200, r.text
    assert r.json() == {"slug": "demo", "reachable": True, "auth_ok": True,
                        "status": 200, "reason": None, "error": None}
    assert calls[0]["init"] == {"timeout": 5.5, "follow_redirects": True}
    post = _post_calls(calls)[0]
    assert post["url"] == "https://push.example/open/v1/inquiry/push"
    assert post["json"] == {"messages": []}          # 红线: 空批量,零线索数据
    assert post["headers"] == {"X-Api-Key": "k-t1"}  # 与 _do_post 同头


def test_push_test_auth_rejected(super_client, monkeypatch):
    """可达但鉴权失败: 401/403 → auth_ok=false(前端三态展示的第二态)。"""
    _create(super_client, slug="demo")
    assert _put_push_group(super_client, "demo", PUSH_GROUP).status_code == 200
    for status in (401, 403):
        _install_fake_client(
            monkeypatch, lambda call, s=status: httpx.Response(s))
        r = _post_test(super_client, "demo")
        assert r.status_code == 200
        body = r.json()
        assert body["reachable"] is True and body["status"] == status
        assert body["auth_ok"] is False


def test_push_test_5xx_reachable_auth_unknown_single_shot(super_client, monkeypatch):
    """5xx: 可达但状态异常(auth_ok=null);探测单次发起,不重试。"""
    _create(super_client, slug="demo")
    assert _put_push_group(super_client, "demo", PUSH_GROUP).status_code == 200
    calls = _install_fake_client(
        monkeypatch, lambda call: httpx.Response(502))

    r = _post_test(super_client, "demo")
    assert r.status_code == 200
    body = r.json()
    assert body["reachable"] is True and body["status"] == 502
    assert body["auth_ok"] is None
    assert len(_post_calls(calls)) == 1             # 无重试矩阵


def test_push_test_unreachable_dns(super_client, monkeypatch):
    """DNS 失败(异常 cause 链含 gaierror)→ reachable=false + reason=dns。"""
    _create(super_client, slug="demo")
    assert _put_push_group(super_client, "demo", PUSH_GROUP).status_code == 200

    def dns_fail(call):
        raise httpx.ConnectError("Failed to resolve.") from socket.gaierror(
            -2, "Name or service not known")

    _install_fake_client(monkeypatch, dns_fail)
    r = _post_test(super_client, "demo")
    assert r.status_code == 200
    body = r.json()
    assert body["reachable"] is False and body["auth_ok"] is None
    assert body["status"] is None
    assert body["reason"] == "dns" and body["error"]


def test_push_test_unreachable_connect(super_client, monkeypatch):
    """连接拒绝/不可达(无 gaierror cause 的 ConnectError)→ reason=connect。"""
    _create(super_client, slug="demo")
    assert _put_push_group(super_client, "demo", PUSH_GROUP).status_code == 200

    def refused(call):
        raise httpx.ConnectError("Connection refused")

    _install_fake_client(monkeypatch, refused)
    r = _post_test(super_client, "demo")
    body = r.json()
    assert body["reachable"] is False
    assert body["reason"] == "connect"


def test_push_test_unreachable_dns_windows_text(super_client, monkeypatch):
    """Windows 实测形态(冒烟 Task 7c): getaddrinfo 失败被 httpx/anyio 包成
    裸 ConnectError("[Errno 11001] getaddrinfo failed"),cause 链不带
    gaierror——文案匹配兜底,仍须归 dns 而非 connect。"""
    _create(super_client, slug="demo")
    assert _put_push_group(super_client, "demo", PUSH_GROUP).status_code == 200

    def win_dns_fail(call):
        raise httpx.ConnectError("[Errno 11001] getaddrinfo failed")

    _install_fake_client(monkeypatch, win_dns_fail)
    r = _post_test(super_client, "demo")
    body = r.json()
    assert body["reachable"] is False
    assert body["reason"] == "dns"
    assert "getaddrinfo" in body["error"]


def test_push_test_unreachable_timeout(super_client, monkeypatch):
    """连接超时(TimeoutException 族)→ reason=timeout。"""
    _create(super_client, slug="demo")
    assert _put_push_group(super_client, "demo", PUSH_GROUP).status_code == 200

    def timed_out(call):
        raise httpx.ConnectTimeout("timed out")

    _install_fake_client(monkeypatch, timed_out)
    r = _post_test(super_client, "demo")
    body = r.json()
    assert body["reachable"] is False
    assert body["reason"] == "timeout"


def test_push_test_url_unconfigured_400(super_client, monkeypatch):
    """URL 回退后仍空(租户 url 空槽 + env PUSH_API_URL 清空)→ 400。"""
    monkeypatch.setattr(settings, "PUSH_API_URL", "")
    _create(super_client, slug="demo")
    assert _put_push_group(super_client, "demo", {"prefix": "demo-p"}).status_code == 200

    r = _post_test(super_client, "demo")
    assert r.status_code == 400
    detail = r.json()["detail"]
    assert "URL" in detail and "未配置" in detail


def test_push_test_env_fallback(super_client, monkeypatch):
    """租户 push 组空槽 → env PUSH_* 回退(探测用的就是部署级平台默认)。"""
    monkeypatch.setattr(settings, "PUSH_API_URL", "https://env-push.example/v1")
    monkeypatch.setattr(settings, "PUSH_API_KEY", "env-key")
    monkeypatch.setattr(settings, "PUSH_TIMEOUT_SECONDS", 8.0)
    _create(super_client, slug="demo")
    assert _put_push_group(super_client, "demo", {"prefix": "demo-p"}).status_code == 200
    calls = _install_fake_client(monkeypatch, _ok)

    r = _post_test(super_client, "demo")
    assert r.status_code == 200
    assert r.json()["reachable"] is True
    assert calls[0]["init"]["timeout"] == 8.0
    post = _post_calls(calls)[0]
    assert post["url"] == "https://env-push.example/v1"
    assert post["headers"] == {"X-Api-Key": "env-key"}


def test_push_test_unknown_tenant_404(super_client, monkeypatch):
    _install_fake_client(monkeypatch, _ok)
    r = _post_test(super_client, "ghost")
    assert r.status_code == 404


def test_push_test_auth_matrix(super_client, monkeypatch):
    """require_tenant(slug): 本租户管理员放行;跨租户 403;未认证 401。"""
    _create(super_client, slug="jc2", admin="jc2-admin", password="jc2-pass-1")
    _create(super_client, slug="au2", profile="auto",
            admin="au2-admin", password="au2-pass-1")
    assert _put_push_group(super_client, "jc2", PUSH_GROUP).status_code == 200
    token = _tenant_admin_token(super_client, "jc2-admin", "jc2-pass-1")
    _install_fake_client(monkeypatch, _ok)

    # 本租户管理员放行
    assert _post_test(super_client, "jc2", headers=_bearer(token)).status_code == 200
    # 别的租户 → 403
    assert _post_test(super_client, "au2", headers=_bearer(token)).status_code == 403
    # 未认证 → 401
    r = super_client.post(PUSH_TEST.format(slug="jc2"))
    assert r.status_code == 401