浅念
fff4f27e98
- email/test 与 push/test 复用 LoginLimiter 滑动窗模式: (kind, tenant_slug) 键每租户 15 分钟 5 次真实出站,超限 429——guard 在端点最前(鉴权后), 被限流的租户连 400 预检也不触发;record 只落在通过全部预检真正出站 的那次(校验失败/未配置不消耗预算,两端口预算按 kind 互不挤占) - conftest autouse 复位限流器(跨测试隔离);新增三测: 第 6 次 429+ 不同租户独立+email/push 互不挤占、校验失败不耗预算 - 回归: 后端 630 passed（627+3） Co-Authored-By: Claude Code <noreply@anthropic.com>
275 lines
11 KiB
Python
275 lines
11 KiB
Python
"""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_rate_limited_after_5(super_client, monkeypatch):
|
|
"""限流(Phase 4 Task 5B): 每租户 15 分钟 5 次探测——前 5 次 200,
|
|
第 6 次 429;email 与 push 预算互不挤占;不同租户独立。"""
|
|
_create(super_client, slug="demo")
|
|
_create(super_client, slug="other", profile="auto")
|
|
assert _put_push_group(super_client, "demo", PUSH_GROUP).status_code == 200
|
|
calls = _install_fake_client(monkeypatch, _ok)
|
|
|
|
for i in range(5):
|
|
r = _post_test(super_client, "demo")
|
|
assert r.status_code == 200, f"第 {i + 1} 次不应限流: {r.text}"
|
|
r = _post_test(super_client, "demo")
|
|
assert r.status_code == 429
|
|
assert "频繁" in r.json()["detail"]
|
|
assert len(_post_calls(calls)) == 5 # 第 6 次不再出站
|
|
|
|
# email 预算独立(键含 kind): 同租户 email/test 不受 push 拉满影响
|
|
# (不配置 SMTP → 400 业务分支,证明 guard 放行)
|
|
monkeypatch.setattr(settings, "SMTP_HOST", "")
|
|
monkeypatch.setattr(settings, "SMTP_USER", "")
|
|
r = super_client.post(f"/api/platform/tenants/demo/email/test",
|
|
headers=_auth(super_client), json={"to": "a@b.com"})
|
|
assert r.status_code == 400
|
|
|
|
# 不同租户独立预算
|
|
assert _put_push_group(super_client, "other", {
|
|
**PUSH_GROUP, "prefix": "other-p"}).status_code == 200
|
|
r = _post_test(super_client, "other")
|
|
assert r.status_code == 200
|
|
|
|
|
|
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
|