q792602257andClaude Opus 4.6
07107094a7
按 docs/order-gateway.md 落地：第三个部署单元 app.gateway（:31109）承担任务队列 + 状态镜像；本地 worker 在 app.trading.worker 内，按 RAKUTEN_ORDER_GATEWAY_URL 决定是否启动。规格 §5 最关键约束已守：租约过期绝不自动重投，恢复只能 reclaim， worker 收到 lease_count>1 时先核对站点订单。 站点交互（加购/下单/付款/订单列表反查）按规格 §10 留接口缝，site_interact.py 全部 NotImplementedError，verify.py 恒返回 unknown——等真实账号实测后再填， 不写猜测的提交逻辑。 310 个测试全绿，覆盖规格 §9 验收清单 12 条；架构测试守住三方互不 import。 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
149 lines
5.2 KiB
Python
149 lines
5.2 KiB
Python
"""本地订单 DB 与证据存储测试
|
|
|
|
覆盖 worker 侧的不变量：本地幂等闸门（has_finished）、状态迁移事件、证据落盘
|
|
路径与 meta.json 字段（规格 §6.1）。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from app.trading.worker.evidence import EvidenceStore
|
|
from app.trading.worker.local_db import LocalDB
|
|
|
|
|
|
@pytest.fixture
|
|
async def local_db(tmp_path: Path) -> LocalDB:
|
|
db = LocalDB(tmp_path / "trading.db")
|
|
await db.start()
|
|
yield db
|
|
await db.close()
|
|
|
|
|
|
# ---- 本地幂等闸门 ----
|
|
|
|
|
|
async def test_has_finished_false_for_unknown_task(local_db: LocalDB):
|
|
assert await local_db.has_finished("no-such") is False
|
|
|
|
|
|
async def test_mark_finished_idempotent(local_db: LocalDB):
|
|
"""ensure_started + mark_finished 后，has_finished 返回 True"""
|
|
await local_db.ensure_started("t1", "rakuten", {"k": "v"})
|
|
assert await local_db.has_finished("t1") is False
|
|
|
|
await local_db.mark_finished("t1", "paid")
|
|
assert await local_db.has_finished("t1") is True
|
|
assert await local_db.final_state("t1") == "paid"
|
|
|
|
|
|
async def test_ensure_started_is_idempotent(local_db: LocalDB):
|
|
"""同任务多次 ensure_started 不覆盖 started_at"""
|
|
await local_db.ensure_started("t1", "rakuten", {"k": "v"})
|
|
row1 = await local_db.get_order("t1")
|
|
await local_db.ensure_started("t1", "rakuten", {"k": "v"})
|
|
row2 = await local_db.get_order("t1")
|
|
assert row1.started_at == row2.started_at
|
|
assert row1.intent == row2.intent == {"k": "v"}
|
|
|
|
|
|
# ---- 状态事件 ----
|
|
|
|
|
|
async def test_record_event_inserts_once_per_state(local_db: LocalDB):
|
|
"""同一 (task_id, state) 重复记录视为同一次——INSERT OR IGNORE"""
|
|
await local_db.ensure_started("t1", "rakuten", {})
|
|
await local_db.record_event("t1", "in_cart", detail="已加购", evidence_ref="t1/01")
|
|
await local_db.record_event("t1", "in_cart", detail="再次加购", evidence_ref="t1/01")
|
|
|
|
async with local_db.conn.execute(
|
|
"SELECT state, detail FROM order_events WHERE task_id = ?", ("t1",)
|
|
) as cur:
|
|
events = await cur.fetchall()
|
|
# 第二次记录被忽略（INSERT OR IGNORE）
|
|
assert len(events) == 1
|
|
assert events[0]["detail"] == "已加购"
|
|
|
|
|
|
async def test_record_event_allows_distinct_states(local_db: LocalDB):
|
|
await local_db.ensure_started("t1", "rakuten", {})
|
|
await local_db.record_event("t1", "in_cart")
|
|
await local_db.record_event("t1", "ordered")
|
|
await local_db.record_event("t1", "paid")
|
|
|
|
async with local_db.conn.execute(
|
|
"SELECT state FROM order_events WHERE task_id = ? ORDER BY recorded_at",
|
|
("t1",),
|
|
) as cur:
|
|
events = await cur.fetchall()
|
|
assert [r["state"] for r in events] == ["in_cart", "ordered", "paid"]
|
|
|
|
|
|
# ---- 证据存储 ----
|
|
|
|
|
|
def test_evidence_write_step_creates_expected_files(tmp_path: Path):
|
|
"""规格 §6.1：每步一个子目录，{序号:02d}-{步骤名}.{html|png|meta.json}"""
|
|
store = EvidenceStore(tmp_path / "evidence")
|
|
rel = store.write_step(
|
|
"po-20260727-0001",
|
|
step_no=1,
|
|
step_name="cart-add",
|
|
html="<html>cart</html>",
|
|
png=b"\x89PNGfake",
|
|
meta={"url": "https://example.com/cart", "method": "POST", "status": 200},
|
|
)
|
|
|
|
assert rel == "po-20260727-0001/01-cart-add"
|
|
step_dir = tmp_path / "evidence" / "po-20260727-0001"
|
|
assert (step_dir / "01-cart-add.html").read_text(encoding="utf-8") == "<html>cart</html>"
|
|
assert (step_dir / "01-cart-add.png").read_bytes() == b"\x89PNGfake"
|
|
|
|
meta = json.loads(
|
|
(step_dir / "01-cart-add.meta.json").read_text(encoding="utf-8")
|
|
)
|
|
assert meta["url"] == "https://example.com/cart"
|
|
assert meta["method"] == "POST"
|
|
assert meta["status"] == 200
|
|
|
|
|
|
def test_evidence_write_step_supports_partial(tmp_path: Path):
|
|
"""只写 meta 不写 html/png 也能正常落盘"""
|
|
store = EvidenceStore(tmp_path / "evidence")
|
|
store.write_step("t1", step_no=2, step_name="confirm", meta={"a": 1})
|
|
step_dir = tmp_path / "evidence" / "t1"
|
|
assert not (step_dir / "02-confirm.html").exists()
|
|
assert not (step_dir / "02-confirm.png").exists()
|
|
assert (step_dir / "02-confirm.meta.json").exists()
|
|
|
|
|
|
def test_evidence_index_records_relative_path(tmp_path: Path):
|
|
"""index_evidence 把相对路径存进 DB，供事后翻查"""
|
|
db = LocalDB(tmp_path / "trading.db")
|
|
# 同步包装：用 asyncio.run 跑一次
|
|
import asyncio
|
|
|
|
async def _run():
|
|
await db.start()
|
|
try:
|
|
await db.ensure_started("t1", "rakuten", {})
|
|
await db.index_evidence("t1", 1, "cart-add", "t1/01-cart-add")
|
|
await db.index_evidence("t1", 2, "confirm", "t1/03-order-confirm")
|
|
|
|
async with db.conn.execute(
|
|
"SELECT step_no, step_name, rel_path FROM evidence_index "
|
|
"WHERE task_id = ? ORDER BY step_no",
|
|
("t1",),
|
|
) as cur:
|
|
rows = await cur.fetchall()
|
|
assert [(r["step_no"], r["step_name"], r["rel_path"]) for r in rows] == [
|
|
(1, "cart-add", "t1/01-cart-add"),
|
|
(2, "confirm", "t1/03-order-confirm"),
|
|
]
|
|
finally:
|
|
await db.close()
|
|
|
|
asyncio.run(_run())
|