q792602257andClaude Opus 5
f6c3976c0a
新增 §12 定时下派通道：网关自己定期派 order_list / order_detail 查询，把账号
真实订单沉淀进新表 account_orders（回写网关），上游可直接查 GET /api/account/orders
拿到账号里实际有哪些订单，不必自己记 order_number。
- collector.py: OrderDiscoveryCollector 常驻后台任务（与 sweep 并列）。每隔
account_discovery_interval_seconds 派 order_list，收割结果后把「编目里没有的
订单」upsert 进编目，再逐笔派 order_detail 沉淀 delivery_status/order_state。
状态无痕：不新增编排跟踪表，仅内存 _pending + _detail_dispatched_this_day；
detail query_id 按天分段（discover-detail-<order>-<日期>），同天幂等防重复派。
边界：collector 只消费 worker 结果的规范化字段，绝不解析 raw/raw_pages。
- db.py: account_orders 表 + AccountOrderRow + upsert/single/list/stale 访问；
upsert 以 order_number 为主键，重复采集只刷新，列表重扫不抹详情节点。
- 路由: GET /api/account/orders（列编目）、GET /api/account/orders/{n}(6005)、
POST /api/account/discovery/trigger（手动立即派一轮 list）。
- config: RAKUTEN_ACCOUNT_DISCOVERY_ENABLED / _INTERVAL_SECONDS / _MAX_PAGES /
RAKUTEN_ACCOUNT_DETAIL_REFRESH_SECONDS。
- 既有网关测试夹具统一关 discovery（会启动即派扫描污染查询队列语义），
新表测试在 tests/test_gateway_discovery.py（13 用例，真实 QueryQueue+DB 装配）。
472 tests passed.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
311 lines
9.7 KiB
Python
311 lines
9.7 KiB
Python
"""网关 HTTP API 测试
|
|
|
|
覆盖规格 §8 的错误码与 §9 验收清单里 HTTP 层能验证的条目：鉴权、响应信封、
|
|
6xxx 错误映射、(task_id, state) 重复 report 不产生第二条、reclaim 错误条件。
|
|
|
|
长轮询与并发抢任务的时序在 test_gateway_leasing.py 单独覆盖。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.shared.config import get_settings
|
|
from app.shared.task_state import OrderState, TaskStatus
|
|
|
|
TOKEN = get_settings().bearer_token
|
|
AUTH = {"Authorization": f"Bearer {TOKEN}"}
|
|
|
|
|
|
@pytest.fixture
|
|
def gateway_client(tmp_path: Path, monkeypatch):
|
|
"""起一个独立 DB 的网关应用
|
|
|
|
- monkeypatch 设置 RAKUTEN_GATEWAY_DB_PATH
|
|
- get_settings.cache_clear() 让新 env 生效
|
|
- TestClient 进入时触发 lifespan（建表、起 sweep 后台任务）
|
|
"""
|
|
db_path = tmp_path / "gw.db"
|
|
monkeypatch.setenv("RAKUTEN_GATEWAY_DB_PATH", str(db_path))
|
|
# 定时下派通道（§12）会在一启动就派 order_list 扫描，污染查询队列语义测试，
|
|
# 这类「队列/租赁/鉴权」用例统一关掉它（它有自己的专门测试文件）。
|
|
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()
|
|
|
|
|
|
# ---- 鉴权 ----
|
|
|
|
|
|
def test_health_needs_no_token(gateway_client):
|
|
response = gateway_client.get("/health")
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["success"] is True
|
|
assert body["data"]["status"] in ("ok", "degraded")
|
|
assert body["data"]["queued_count"] == 0
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
"path,method",
|
|
[
|
|
("/api/orders", "POST"),
|
|
("/api/orders/lease", "GET"),
|
|
("/api/orders/t1", "GET"),
|
|
],
|
|
)
|
|
def test_endpoints_reject_missing_token(gateway_client, path, method):
|
|
response = gateway_client.request(method, path)
|
|
assert response.status_code == 401
|
|
assert response.json()["code"] == 1001
|
|
|
|
|
|
def test_endpoints_reject_wrong_token(gateway_client):
|
|
response = gateway_client.post(
|
|
"/api/orders",
|
|
json={"site": "rakuten", "intent": {}},
|
|
headers={"Authorization": "Bearer wrong"},
|
|
)
|
|
assert response.status_code == 401
|
|
|
|
|
|
# ---- 提交与幂等 ----
|
|
|
|
|
|
def test_submit_returns_task_id_and_status(gateway_client):
|
|
response = gateway_client.post(
|
|
"/api/orders",
|
|
json={"task_id": "t1", "site": "rakuten", "intent": {"k": "v"}},
|
|
headers=AUTH,
|
|
)
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["success"] is True
|
|
assert body["code"] == 0
|
|
assert body["data"]["task_id"] == "t1"
|
|
assert body["data"]["status"] == TaskStatus.QUEUED.value
|
|
assert body["data"]["created"] is True
|
|
|
|
|
|
def test_submit_with_same_task_id_is_idempotent(gateway_client):
|
|
payload = {"task_id": "t1", "site": "rakuten", "intent": {}}
|
|
r1 = gateway_client.post("/api/orders", json=payload, headers=AUTH).json()
|
|
r2 = gateway_client.post("/api/orders", json=payload, headers=AUTH).json()
|
|
assert r1["data"]["created"] is True
|
|
assert r2["data"]["created"] is False
|
|
assert r1["data"]["task_id"] == r2["data"]["task_id"]
|
|
|
|
|
|
# ---- 错误码 6xxx 映射 ----
|
|
|
|
|
|
def test_get_unknown_task_returns_6001(gateway_client):
|
|
response = gateway_client.get("/api/orders/no-such-task", headers=AUTH)
|
|
assert response.status_code == 404
|
|
body = response.json()
|
|
assert body["success"] is False
|
|
assert body["code"] == 6001
|
|
|
|
|
|
def test_report_wrong_worker_returns_6002(gateway_client):
|
|
# 提交并 lease 给 w1
|
|
gateway_client.post(
|
|
"/api/orders",
|
|
json={"task_id": "t1", "site": "rakuten", "intent": {}},
|
|
headers=AUTH,
|
|
)
|
|
gateway_client.get("/api/orders/lease?worker_id=w1&wait=0", headers=AUTH)
|
|
|
|
# w2 试图 report
|
|
response = gateway_client.post(
|
|
"/api/orders/t1/report",
|
|
json={"worker_id": "w2", "state": OrderState.IN_CART.value},
|
|
headers=AUTH,
|
|
)
|
|
assert response.status_code == 409
|
|
assert response.json()["code"] == 6002
|
|
|
|
|
|
def test_reclaim_non_stale_returns_6003(gateway_client):
|
|
gateway_client.post(
|
|
"/api/orders",
|
|
json={"task_id": "t1", "site": "rakuten", "intent": {}},
|
|
headers=AUTH,
|
|
)
|
|
response = gateway_client.post(
|
|
"/api/orders/t1/reclaim",
|
|
json={"worker_id": "w1"},
|
|
headers=AUTH,
|
|
)
|
|
assert response.status_code == 409
|
|
assert response.json()["code"] == 6003
|
|
|
|
|
|
def test_renew_unknown_task_returns_6001(gateway_client):
|
|
response = gateway_client.post(
|
|
"/api/orders/no-such/renew",
|
|
json={"worker_id": "w1"},
|
|
headers=AUTH,
|
|
)
|
|
assert response.status_code == 404
|
|
assert response.json()["code"] == 6001
|
|
|
|
|
|
# ---- report 幂等：同一 (task_id, state) 不产生第二条 ----
|
|
|
|
|
|
def test_report_same_state_does_not_duplicate(gateway_client):
|
|
gateway_client.post(
|
|
"/api/orders",
|
|
json={"task_id": "t1", "site": "rakuten", "intent": {}},
|
|
headers=AUTH,
|
|
)
|
|
gateway_client.get("/api/orders/lease?worker_id=w1&wait=0", headers=AUTH)
|
|
|
|
report_payload = {
|
|
"worker_id": "w1",
|
|
"state": OrderState.IN_CART.value,
|
|
"payable_yen": 9800,
|
|
"evidence_ref": "t1/01-cart-add",
|
|
"detail": "已加购",
|
|
}
|
|
r1 = gateway_client.post("/api/orders/t1/report", json=report_payload, headers=AUTH).json()
|
|
r2 = gateway_client.post("/api/orders/t1/report", json=report_payload, headers=AUTH).json()
|
|
|
|
assert r1["data"]["recorded"] is True
|
|
assert r2["data"]["recorded"] is False
|
|
|
|
detail = gateway_client.get("/api/orders/t1", headers=AUTH).json()["data"]
|
|
assert len(detail["reports"]) == 1
|
|
|
|
|
|
def test_report_terminal_releases_lease(gateway_client):
|
|
gateway_client.post(
|
|
"/api/orders",
|
|
json={"task_id": "t1", "site": "rakuten", "intent": {}},
|
|
headers=AUTH,
|
|
)
|
|
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
|
|
|
|
detail = gateway_client.get("/api/orders/t1", headers=AUTH).json()["data"]
|
|
assert detail["status"] == TaskStatus.SUCCEEDED.value
|
|
assert detail["lease_owner"] is None
|
|
|
|
|
|
# ---- 列表 ----
|
|
|
|
|
|
def test_list_orders_filters_by_status(gateway_client):
|
|
gateway_client.post(
|
|
"/api/orders",
|
|
json={"task_id": "t1", "site": "rakuten", "intent": {}},
|
|
headers=AUTH,
|
|
)
|
|
gateway_client.post(
|
|
"/api/orders",
|
|
json={"task_id": "t2", "site": "rakuten", "intent": {}},
|
|
headers=AUTH,
|
|
)
|
|
|
|
response = gateway_client.get("/api/orders?status=queued", headers=AUTH).json()
|
|
assert response["data"]["total"] == 2
|
|
assert {item["task_id"] for item in response["data"]["items"]} == {"t1", "t2"}
|
|
|
|
filtered = gateway_client.get(
|
|
"/api/orders?status=leased", headers=AUTH
|
|
).json()
|
|
assert filtered["data"]["total"] == 0
|
|
|
|
|
|
def test_list_orders_filters_by_site(gateway_client):
|
|
gateway_client.post(
|
|
"/api/orders",
|
|
json={"task_id": "t1", "site": "rakuten", "intent": {}},
|
|
headers=AUTH,
|
|
)
|
|
gateway_client.post(
|
|
"/api/orders",
|
|
json={"task_id": "t2", "site": "rakuma", "intent": {}},
|
|
headers=AUTH,
|
|
)
|
|
|
|
response = gateway_client.get("/api/orders?site=rakuma", headers=AUTH).json()
|
|
assert response["data"]["total"] == 1
|
|
assert response["data"]["items"][0]["site"] == "rakuma"
|
|
|
|
|
|
def test_get_task_detail_includes_full_report_history(gateway_client):
|
|
gateway_client.post(
|
|
"/api/orders",
|
|
json={"task_id": "t1", "site": "rakuten", "intent": {}},
|
|
headers=AUTH,
|
|
)
|
|
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, "detail": "step1"},
|
|
headers=AUTH,
|
|
)
|
|
gateway_client.post(
|
|
"/api/orders/t1/report",
|
|
json={"worker_id": "w1", "state": OrderState.ORDERED.value, "detail": "step2"},
|
|
headers=AUTH,
|
|
)
|
|
|
|
detail = gateway_client.get("/api/orders/t1", headers=AUTH).json()["data"]
|
|
assert [r["state"] for r in detail["reports"]] == [
|
|
OrderState.IN_CART.value,
|
|
OrderState.ORDERED.value,
|
|
]
|
|
assert detail["latest_state"] == OrderState.ORDERED.value
|
|
|
|
|
|
# ---- lease 立即返回空（无任务时）----
|
|
|
|
|
|
def test_lease_returns_null_when_queue_empty(gateway_client):
|
|
"""wait=0 + 无任务时立即返回 data: null"""
|
|
response = gateway_client.get("/api/orders/lease?worker_id=w1&wait=0", headers=AUTH)
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["success"] is True
|
|
assert body["data"] is None
|
|
|
|
|
|
def test_lease_returns_null_when_active_task_blocks(gateway_client):
|
|
"""有 leased 任务时，lease 立即返回 data: null（并发度 1）"""
|
|
gateway_client.post(
|
|
"/api/orders",
|
|
json={"task_id": "t1", "site": "rakuten", "intent": {}},
|
|
headers=AUTH,
|
|
)
|
|
gateway_client.get("/api/orders/lease?worker_id=w1&wait=0", headers=AUTH)
|
|
|
|
response = gateway_client.get("/api/orders/lease?worker_id=w2&wait=0", headers=AUTH)
|
|
assert response.status_code == 200
|
|
assert response.json()["data"] is None
|