q792602257
f31e127ba7
- site_interact 新增 PageSnapshot{html, screenshot} 证据载体：
add_to_cart 截商品页、verify_cart 截 cart 页、enter_checkout 截落地
确认页、pay 截付款检查后的完成页；截图全部 best-effort，失败记
warning 留空，不掩盖动作结果
- submit_order 改返回 SubmitOutcome{site_order_id, evidence}：完成页
HTML+截图首次随 step 4 落盘（此前只有 meta）
- fetch_order_detail 详情页截图挂在 OrderStatusSnapshot.screenshot，
监控步骤 write_step 带 png；只读查询通道按字段挑出，不受影响
- runner._run_step 认 PageSnapshot（旧 str 契约保留兼容），step 0/3/4
显式传 png
- 测试：桩同步新契约；新增每步三件套齐全的全流程断言；clear_cart
场景单测补截图断言
- 真账号复验 verify_cart_clear.py：两场景截图均合法 PNG，落盘
.probe/cart_clear/ 并目检为空车页渲染
101 lines
4.5 KiB
Python
101 lines
4.5 KiB
Python
"""真账号验证：clear_cart 两个场景——空车返回成功 / 有商品正确清除
|
|
|
|
场景 1（空车）：clear_cart 应 removed_count=0、cart_count=0（status=101 是合法
|
|
空车，不是获取失败），并带回清理后的页面 HTML 与整页截图（runner step 0 落
|
|
证据用）。
|
|
场景 2（有商品）：先加购 1 件商品，clear_cart 应 removed_count>=1、cart_count=0，
|
|
末尾再用 cart_status 独立复核确实是空车。
|
|
|
|
两个场景的截图会写到 .probe/cart_clear/ 供人工查看。
|
|
|
|
前置：登录态有效（.auth/ 下有 storage_state）；浏览器必须非无头（站点风控要求，
|
|
会弹出真实浏览器窗口）。商品 URL 与其他探针（probe_new_card 等）一致。
|
|
|
|
用法：
|
|
.venv/Scripts/python.exe scripts/verify_cart_clear.py
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
|
|
|
|
from app.shared.config import get_settings # noqa: E402
|
|
from app.trading.services.auth_session import AuthSession # noqa: E402
|
|
from app.trading.worker.site_interact import SiteInteractor # noqa: E402
|
|
|
|
ITEM_URL = "https://item.rakuten.co.jp/moccasin/ds001iwrgesaaa2/"
|
|
|
|
PROBE_DIR = Path(__file__).resolve().parent.parent / ".probe" / "cart_clear"
|
|
_PNG_MAGIC = b"\x89PNG\r\n\x1a\n"
|
|
|
|
|
|
def _check_png(data: bytes, label: str) -> None:
|
|
assert data[:8] == _PNG_MAGIC, f"{label} 截图不是 PNG（{len(data)}B）"
|
|
|
|
|
|
async def run() -> int:
|
|
settings = get_settings()
|
|
auth = AuthSession(settings)
|
|
await auth.start()
|
|
site = SiteInteractor(auth_session=auth, settings=settings)
|
|
await site.start()
|
|
PROBE_DIR.mkdir(parents=True, exist_ok=True)
|
|
try:
|
|
# 前置：先清一次，保证从空车起步（顺便清掉上次验证可能残留的商品）
|
|
pre = await site.clear_cart()
|
|
print(f"前置 clear -> removed={pre['removed_count']} cart_count={pre['cart_count']}")
|
|
assert pre["cart_count"] == 0, f"前置清理后应为空车，实际 {pre['cart_count']}"
|
|
|
|
# 场景 1：空车 —— 应直接返回清理成功，不做任何点击
|
|
empty = await site.clear_cart()
|
|
print(
|
|
f"场景1 空车 -> removed={empty['removed_count']} "
|
|
f"cart_count={empty['cart_count']} html={len(empty['html'])}B "
|
|
f"png={len(empty['screenshot'])}B"
|
|
)
|
|
assert empty["removed_count"] == 0, f"空车不应有点击，实际 {empty['removed_count']}"
|
|
assert empty["cart_count"] == 0, f"空车 cart_count 应为 0，实际 {empty['cart_count']}"
|
|
assert empty["html"], "空车也应带回页面 HTML（runner 落证据用）"
|
|
_check_png(empty["screenshot"], "场景1 空车")
|
|
(PROBE_DIR / "scenario1-empty.png").write_bytes(empty["screenshot"])
|
|
|
|
# 场景 2：有商品 —— 先加购 1 件，再清，应真删掉
|
|
added = await site.add_to_cart_payload(item_url=ITEM_URL, quantity=1)
|
|
print(f"加购 -> item_id={added['item_id']} cart_count={added['cart_count']}")
|
|
assert added["cart_count"] >= 1, "加购后购物车应有商品"
|
|
_check_png(added["screenshot"], "加购后商品页")
|
|
|
|
cleared = await site.clear_cart()
|
|
print(
|
|
f"场景2 有商品 -> removed={cleared['removed_count']} "
|
|
f"cart_count={cleared['cart_count']} html={len(cleared['html'])}B "
|
|
f"png={len(cleared['screenshot'])}B"
|
|
)
|
|
assert cleared["removed_count"] >= 1, (
|
|
f"有商品时至少点 1 次「削除」，实际 {cleared['removed_count']}"
|
|
)
|
|
assert cleared["cart_count"] == 0, f"清理后应为空车，实际 {cleared['cart_count']}"
|
|
assert cleared["html"], "清理后应带回页面 HTML"
|
|
_check_png(cleared["screenshot"], "场景2 清理后")
|
|
(PROBE_DIR / "scenario2-cleared.png").write_bytes(cleared["screenshot"])
|
|
|
|
# 复核：独立的 status 查询确认确实是空车（不只信 clear 末尾的一次校验）
|
|
status = await site.cart_status()
|
|
print(f"复核 status -> {status}")
|
|
assert status["count"] == 0 and status["raw_status"] == "101", (
|
|
f"复核应为空车（count=0, raw_status=101），实际 {status}"
|
|
)
|
|
|
|
print(f"\n验证通过：空车正确返回成功；有商品时正确清除并复核为空车。截图见 {PROBE_DIR}")
|
|
return 0
|
|
finally:
|
|
await site.close()
|
|
await auth.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(asyncio.run(run()))
|