"""本地订单 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="cart", 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") == "cart" 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())