q792602257
48c6f2a28f
- POST /api/orders 新增可选 callback_url（仅 http/https，其余 422） - 仅终结类事件各通知一次：terminal report 推入终态（succeeded/failed/ needs_human）、租约过期被 sweep 置 stale；中间态与终结后的监控上报不通知 - 幂等重发不更新既有任务的回调地址；投递 best-effort 单次尝试，失败只记日志 - CallbackNotifier 发后不管（create_task + 在途任务强引用），关停等在途发完； 生产路径按回调地址逐次构造客户端，满足统一出站代理策略（test_proxy.py） - tasks 表加 callback_url 列，GatewayDB.start() 内置迁移兼容既有库 - 新增 RAKUTEN_CALLBACK_TIMEOUT_SECONDS（默认 10）；文档补 §4.8； openapi.json 重新导出（gitignore 未跟踪）；新增 17 条测试，全量 492 通过
357 lines
12 KiB
Python
357 lines
12 KiB
Python
"""终结类事件回调（callback_url）测试，对应 docs/order-gateway.md §4.8
|
|
|
|
三层覆盖：
|
|
|
|
- API 层：提交校验（非法地址 422）、详情透出 callback_url、幂等重发不改地址、
|
|
HTTP 全流程下 terminal report 恰好通知一次（注入记录型假 notifier）。
|
|
- 队列层：中间态不通知、终结通知一次、终结后的后续 report 不重复通知、
|
|
租约过期 sweep 触发 stale 通知、未登记地址不通知。
|
|
- 通知器层：真实发送路径（httpx.MockTransport）——成功送达、上游 5xx、
|
|
连接异常都不抛出、只记日志（best-effort）。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import aiosqlite
|
|
import httpx
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.gateway.callback import CallbackNotifier
|
|
from app.gateway.db import GatewayDB
|
|
from app.gateway.task_queue import TaskQueue
|
|
from app.shared.config import Settings, get_settings
|
|
from app.shared.task_state import OrderState, TaskStatus
|
|
|
|
TOKEN = get_settings().bearer_token
|
|
AUTH = {"Authorization": f"Bearer {TOKEN}"}
|
|
CALLBACK_URL = "https://upstream.example.com/hooks/rakuten-order"
|
|
|
|
|
|
class FakeNotifier:
|
|
"""记录型假通知器：与 CallbackNotifier 同接口，同步记录便于断言"""
|
|
|
|
def __init__(self) -> None:
|
|
self.sent: list[tuple[str, dict]] = []
|
|
|
|
def notify(self, callback_url: str, payload: dict) -> None:
|
|
self.sent.append((callback_url, payload))
|
|
|
|
|
|
# ---- API 层 ----
|
|
|
|
|
|
@pytest.fixture
|
|
def gateway_client(tmp_path: Path, monkeypatch):
|
|
db_path = tmp_path / "gw.db"
|
|
monkeypatch.setenv("RAKUTEN_GATEWAY_DB_PATH", str(db_path))
|
|
monkeypatch.setenv("RAKUTEN_ACCOUNT_DISCOVERY_ENABLED", "false")
|
|
get_settings.cache_clear()
|
|
try:
|
|
from app.gateway.main import create_app
|
|
|
|
app = create_app()
|
|
with TestClient(app) as client:
|
|
yield client
|
|
finally:
|
|
get_settings.cache_clear()
|
|
|
|
|
|
@pytest.fixture
|
|
def fake_notifier(gateway_client):
|
|
"""把容器里 TaskQueue 的通知器换成记录型假实现，用完还原"""
|
|
container = gateway_client.app.state.container
|
|
original = container.task_queue.notifier
|
|
fake = FakeNotifier()
|
|
container.task_queue.notifier = fake
|
|
yield fake
|
|
container.task_queue.notifier = original
|
|
|
|
|
|
def _submit(client, *, task_id: str, callback_url: str | None = CALLBACK_URL):
|
|
payload: dict = {"task_id": task_id, "site": "rakuten", "intent": {}}
|
|
if callback_url is not None:
|
|
payload["callback_url"] = callback_url
|
|
return client.post("/api/orders", json=payload, headers=AUTH)
|
|
|
|
|
|
def test_submit_with_callback_url_roundtrip(gateway_client):
|
|
response = _submit(gateway_client, task_id="t1")
|
|
assert response.status_code == 200
|
|
assert response.json()["data"]["created"] is True
|
|
|
|
detail = gateway_client.get("/api/orders/t1", headers=AUTH).json()["data"]
|
|
assert detail["callback_url"] == CALLBACK_URL
|
|
|
|
|
|
def test_submit_without_callback_url_defaults_to_null(gateway_client):
|
|
_submit(gateway_client, task_id="t1", callback_url=None)
|
|
detail = gateway_client.get("/api/orders/t1", headers=AUTH).json()["data"]
|
|
assert detail["callback_url"] is None
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"bad_url",
|
|
[
|
|
"ftp://upstream.example.com/hook", # 非 http/https scheme
|
|
"not-a-url", # 没有 scheme 与 host
|
|
"https://", # 有 scheme 没 host
|
|
],
|
|
)
|
|
def test_submit_rejects_invalid_callback_url(gateway_client, bad_url):
|
|
response = _submit(gateway_client, task_id="t1", callback_url=bad_url)
|
|
assert response.status_code == 422
|
|
|
|
|
|
def test_idempotent_resubmit_keeps_original_callback_url(gateway_client):
|
|
"""同 task_id 重发带不同 callback_url：不新建、不更新地址"""
|
|
_submit(gateway_client, task_id="t1", callback_url=CALLBACK_URL)
|
|
r2 = _submit(gateway_client, task_id="t1", callback_url="https://other.example.com/hook")
|
|
assert r2.json()["data"]["created"] is False
|
|
|
|
detail = gateway_client.get("/api/orders/t1", headers=AUTH).json()["data"]
|
|
assert detail["callback_url"] == CALLBACK_URL
|
|
|
|
|
|
def test_terminal_report_sends_exactly_one_callback(gateway_client, fake_notifier):
|
|
_submit(gateway_client, task_id="t1")
|
|
gateway_client.get("/api/orders/lease?worker_id=w1&wait=0", headers=AUTH)
|
|
|
|
response = gateway_client.post(
|
|
"/api/orders/t1/report",
|
|
json={
|
|
"worker_id": "w1",
|
|
"state": OrderState.PAID.value,
|
|
"payable_yen": 9800,
|
|
"site_order_id": "ord-1",
|
|
"terminal": True,
|
|
"terminal_status": TaskStatus.SUCCEEDED.value,
|
|
"detail": "付款完成",
|
|
},
|
|
headers=AUTH,
|
|
)
|
|
assert response.status_code == 200
|
|
|
|
assert len(fake_notifier.sent) == 1
|
|
url, payload = fake_notifier.sent[0]
|
|
assert url == CALLBACK_URL
|
|
assert payload["event"] == "terminal"
|
|
assert payload["task_id"] == "t1"
|
|
assert payload["site"] == "rakuten"
|
|
assert payload["status"] == TaskStatus.SUCCEEDED.value
|
|
assert payload["state"] == OrderState.PAID.value
|
|
assert payload["payable_yen"] == 9800
|
|
assert payload["site_order_id"] == "ord-1"
|
|
assert payload["detail"] == "付款完成"
|
|
assert payload["reported_at"]
|
|
|
|
|
|
def test_non_terminal_report_sends_no_callback(gateway_client, fake_notifier):
|
|
_submit(gateway_client, task_id="t1")
|
|
gateway_client.get("/api/orders/lease?worker_id=w1&wait=0", headers=AUTH)
|
|
gateway_client.post(
|
|
"/api/orders/t1/report",
|
|
json={"worker_id": "w1", "state": OrderState.IN_CART.value},
|
|
headers=AUTH,
|
|
)
|
|
assert fake_notifier.sent == []
|
|
|
|
|
|
def test_report_after_terminal_sends_no_more_callbacks(gateway_client, fake_notifier):
|
|
"""任务终结后的后续 report（付款后监控）与幂等重报都不再通知"""
|
|
_submit(gateway_client, task_id="t1")
|
|
gateway_client.get("/api/orders/lease?worker_id=w1&wait=0", headers=AUTH)
|
|
terminal_report = {
|
|
"worker_id": "w1",
|
|
"state": OrderState.PAID.value,
|
|
"terminal": True,
|
|
"terminal_status": TaskStatus.SUCCEEDED.value,
|
|
}
|
|
gateway_client.post("/api/orders/t1/report", json=terminal_report, headers=AUTH)
|
|
assert len(fake_notifier.sent) == 1
|
|
|
|
# 同一 (task_id, state) 幂等重报 + 终结后的监控上报，都不应再触发
|
|
gateway_client.post("/api/orders/t1/report", json=terminal_report, headers=AUTH)
|
|
gateway_client.post(
|
|
"/api/orders/t1/report",
|
|
json={"worker_id": "w1", "state": OrderState.SHIPPED.value},
|
|
headers=AUTH,
|
|
)
|
|
assert len(fake_notifier.sent) == 1
|
|
|
|
|
|
def test_needs_human_inferred_terminal_sends_callback(gateway_client, fake_notifier):
|
|
"""terminal=true 不显式给终态时按 state 推断（非 paid/cancelled → needs_human）"""
|
|
_submit(gateway_client, task_id="t1")
|
|
gateway_client.get("/api/orders/lease?worker_id=w1&wait=0", headers=AUTH)
|
|
gateway_client.post(
|
|
"/api/orders/t1/report",
|
|
json={
|
|
"worker_id": "w1",
|
|
"state": OrderState.AWAITING_PAYMENT.value,
|
|
"terminal": True,
|
|
"detail": "弹了 3DS",
|
|
},
|
|
headers=AUTH,
|
|
)
|
|
assert len(fake_notifier.sent) == 1
|
|
_, payload = fake_notifier.sent[0]
|
|
assert payload["status"] == TaskStatus.NEEDS_HUMAN.value
|
|
|
|
|
|
# ---- 队列层 ----
|
|
|
|
|
|
@pytest.fixture
|
|
async def queue(tmp_path: Path):
|
|
db = GatewayDB(tmp_path / "gw.db")
|
|
await db.start()
|
|
fake = FakeNotifier()
|
|
q = TaskQueue(
|
|
db, lease_ttl_seconds=0, worker_offline_alert_seconds=300, notifier=fake
|
|
)
|
|
yield q, fake
|
|
await db.close()
|
|
|
|
|
|
async def test_sweep_stale_sends_callback(queue):
|
|
"""租约过期被 sweep 置 stale 时触发 stale 通知（上游需要人工 reclaim）"""
|
|
q, fake = queue
|
|
await q.submit(
|
|
task_id="t1", site="rakuten", intent={}, callback_url=CALLBACK_URL
|
|
)
|
|
# lease_ttl=0：领走即过期，下一次 sweep 置 stale
|
|
await q.lease(worker_id="w1", wait=0, site=None, max_wait=60)
|
|
|
|
swept = await q.sweep()
|
|
assert swept == 1
|
|
assert len(fake.sent) == 1
|
|
url, payload = fake.sent[0]
|
|
assert url == CALLBACK_URL
|
|
assert payload["event"] == "stale"
|
|
assert payload["status"] == TaskStatus.STALE.value
|
|
assert payload["task_id"] == "t1"
|
|
|
|
|
|
async def test_sweep_without_callback_url_sends_nothing(queue):
|
|
q, fake = queue
|
|
await q.submit(task_id="t1", site="rakuten", intent={})
|
|
await q.lease(worker_id="w1", wait=0, site=None, max_wait=60)
|
|
|
|
swept = await q.sweep()
|
|
assert swept == 1
|
|
assert fake.sent == []
|
|
|
|
|
|
async def test_terminal_without_callback_url_sends_nothing(queue):
|
|
q, fake = queue
|
|
await q.submit(task_id="t1", site="rakuten", intent={})
|
|
await q.lease(worker_id="w1", wait=0, site=None, max_wait=60)
|
|
await q.report(
|
|
"t1",
|
|
worker_id="w1",
|
|
state=OrderState.PAID.value,
|
|
payable_yen=None,
|
|
pay_deadline=None,
|
|
site_order_id=None,
|
|
evidence_ref=None,
|
|
detail="",
|
|
terminal=True,
|
|
terminal_status=None,
|
|
)
|
|
assert fake.sent == []
|
|
|
|
|
|
# ---- DB 迁移 ----
|
|
|
|
|
|
async def test_start_migrates_tasks_table_without_callback_url(tmp_path: Path):
|
|
"""既有库（无 callback_url 列的老表）启动时补列，原数据不丢"""
|
|
db_path = tmp_path / "old.db"
|
|
conn = await aiosqlite.connect(str(db_path))
|
|
await conn.execute(
|
|
"CREATE TABLE tasks ("
|
|
"task_id TEXT PRIMARY KEY, site TEXT NOT NULL, intent_json TEXT NOT NULL, "
|
|
"status TEXT NOT NULL, lease_owner TEXT, lease_expires_at TEXT, "
|
|
"lease_count INTEGER NOT NULL DEFAULT 0, created_at TEXT NOT NULL, "
|
|
"updated_at TEXT NOT NULL)"
|
|
)
|
|
await conn.execute(
|
|
"INSERT INTO tasks VALUES ('t1', 'rakuten', '{}', 'queued', "
|
|
"NULL, NULL, 0, '2026-08-16T00:00:00Z', '2026-08-16T00:00:00Z')"
|
|
)
|
|
await conn.commit()
|
|
await conn.close()
|
|
|
|
db = GatewayDB(db_path)
|
|
await db.start()
|
|
try:
|
|
task = await db.get_task("t1")
|
|
assert task is not None
|
|
assert task.callback_url is None
|
|
finally:
|
|
await db.close()
|
|
|
|
|
|
# ---- 通知器层（真实发送路径，MockTransport）----
|
|
|
|
|
|
def _bare_settings() -> Settings:
|
|
"""不读 .env 的最小配置，通知器只要代理策略字段"""
|
|
return Settings(_env_file=None)
|
|
|
|
|
|
async def test_notifier_delivers_payload():
|
|
received: list[httpx.Request] = []
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
received.append(request)
|
|
return httpx.Response(200)
|
|
|
|
notifier = CallbackNotifier(
|
|
settings=_bare_settings(),
|
|
timeout_seconds=5,
|
|
client=httpx.AsyncClient(transport=httpx.MockTransport(handler)),
|
|
)
|
|
payload = {"task_id": "t1", "event": "terminal", "status": "succeeded"}
|
|
notifier.notify(CALLBACK_URL, payload)
|
|
await notifier.aclose() # 等在途通知发完
|
|
|
|
assert len(received) == 1
|
|
assert str(received[0].url) == CALLBACK_URL
|
|
assert json.loads(received[0].content) == payload
|
|
|
|
|
|
async def test_notifier_swallows_upstream_5xx(caplog):
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
return httpx.Response(500)
|
|
|
|
notifier = CallbackNotifier(
|
|
settings=_bare_settings(),
|
|
timeout_seconds=5,
|
|
client=httpx.AsyncClient(transport=httpx.MockTransport(handler)),
|
|
)
|
|
with caplog.at_level("WARNING", logger="app.gateway.callback"):
|
|
notifier.notify(CALLBACK_URL, {"task_id": "t1"})
|
|
await notifier.aclose() # 不抛出
|
|
|
|
assert any("非成功状态" in rec.message for rec in caplog.records)
|
|
|
|
|
|
async def test_notifier_swallows_connection_error(caplog):
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
raise httpx.ConnectError("connection refused", request=request)
|
|
|
|
notifier = CallbackNotifier(
|
|
settings=_bare_settings(),
|
|
timeout_seconds=5,
|
|
client=httpx.AsyncClient(transport=httpx.MockTransport(handler)),
|
|
)
|
|
with caplog.at_level("WARNING", logger="app.gateway.callback"):
|
|
notifier.notify(CALLBACK_URL, {"task_id": "t1"})
|
|
await notifier.aclose() # 不抛出
|
|
|
|
assert any("发送失败" in rec.message for rec in caplog.records)
|