q792602257andClaude Opus 5
c03158488b
上游要的不只是网关记的任务状态镜像，还有「已登录账号在站点上的真实订单」，
但账号只在 NAT 后本地机上，只能经网关队列走。新增独立查询通道（§11）：
- gateway 单开 account_queries 表 + QueryStatus 状态机，接口
POST /api/account/queries（幂等）/ lease / {id}/result / {id}
- 不复用下单任务队列：查询是只读，租约过期可安全重投（与下单「绝不自动
重投」相反），且不该被全局并发度 1 堵死、task_reports 是订单镜像不能污染
- 本地交易服务起第二条常驻循环 query_runner，领到即调 SiteInteractor 真读：
order_list 复用已实测的 list_recent_orders（规范化字段 + 站点
orderListData 原文），order_detail 复用 fetch_order_detail（配送阶段 +
页面 __INITIAL_STATE__ 原样透传，结构未经真实样本，不抽字段）
- 账号级串行仍由 SiteInteractor 的锁保证；每次执行套超时按失败回报
- 错误码 6005/6006（查询通道，可重试只读区别于 6001-6004）；/health 暴露
queued_query_count；结果体积上限先丢原始 JSON
openapi.json 重导，docs/order-gateway.md §11、README、.env.example 补全
配置与实测边界。全量测试 404→454 通过。
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
1029 lines
40 KiB
Python
1029 lines
40 KiB
Python
"""site_interact 单元测试
|
|
|
|
重点测模块级纯函数（_parse_initial_state / _extract_purchase_fields /
|
|
_extract_error_message）——这三者覆盖了「从商品页 HTML 抽加购表单」的核心逻辑，
|
|
是 worker 真实跑起来时最关键的转换。
|
|
|
|
SiteInteractor 类的 add_to_cart / verify_cart 涉及 Playwright，不在离线测试覆盖
|
|
范围；只在 test_worker_runner.py 里用桩站点覆盖 runner 与 site 的契约。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
from app.shared.errors import (
|
|
CartOperationError,
|
|
InvalidRequestError,
|
|
NotLoggedInError,
|
|
OrderOperationError,
|
|
)
|
|
from app.trading.core import auth_site
|
|
from app.trading.worker.models import LeaseTask
|
|
from app.shared.task_state import OrderState
|
|
from app.trading.worker.site_interact import (
|
|
CheckoutSummary,
|
|
OrderListEntry,
|
|
OrderListPage,
|
|
OrderStatusSnapshot,
|
|
SiteInteractor,
|
|
_accumulate_order_list_page,
|
|
_extract_error_message,
|
|
_extract_purchase_fields,
|
|
_OrderListAccumulator,
|
|
_parse_checkout_summary,
|
|
_parse_initial_state,
|
|
_parse_order_list,
|
|
_parse_order_status,
|
|
parse_order_datetime,
|
|
)
|
|
|
|
FIXTURES = Path(__file__).parent / "fixtures"
|
|
|
|
|
|
def _make_task(**kwargs: Any) -> LeaseTask:
|
|
defaults = {
|
|
"task_id": "t1",
|
|
"site": "rakuten",
|
|
"intent": {"item_url": "https://item.rakuten.co.jp/shop/x/"},
|
|
}
|
|
defaults.update(kwargs)
|
|
return LeaseTask(**defaults)
|
|
|
|
|
|
def _wrap_state(state: dict[str, Any]) -> str:
|
|
"""构造一段模拟的 item page HTML，把 state 嵌进 __INITIAL_STATE__"""
|
|
return (
|
|
"<html><script>window.__INITIAL_STATE__ = "
|
|
+ json.dumps(state, ensure_ascii=False)
|
|
+ ";window.__OTHER__ = 1;</script></html>"
|
|
)
|
|
|
|
|
|
# ---- _parse_initial_state ----
|
|
|
|
|
|
def test_parse_initial_state_extracts_state_dict():
|
|
state = {"item": {"itemId": 123}, "purchase": {}}
|
|
parsed = _parse_initial_state(_wrap_state(state))
|
|
assert parsed == state
|
|
|
|
|
|
def test_parse_initial_state_returns_none_when_no_marker():
|
|
assert _parse_initial_state("<html>no state here</html>") is None
|
|
|
|
|
|
def test_parse_initial_state_returns_none_on_invalid_json():
|
|
html = "<script>window.__INITIAL_STATE__ = {invalid};window.__END=1;</script>"
|
|
assert _parse_initial_state(html) is None
|
|
|
|
|
|
# ---- _extract_purchase_fields ----
|
|
|
|
|
|
def test_extract_fields_from_multi_inventory_fixture():
|
|
"""item_state.json 是多规格商品，应自动选第一个非售罄 variant"""
|
|
state = json.loads((FIXTURES / "item_state.json").read_text(encoding="utf-8"))
|
|
fields = _extract_purchase_fields(state, intent_override={})
|
|
assert fields["purchase_condition"] == "enabled"
|
|
assert fields["basket_domain"].startswith("https://")
|
|
assert fields["basket_domain"].endswith("/rms/mall/bss/cartadd/set")
|
|
# 多规格：form 应包含自动选的 variant_id
|
|
assert fields["inventory_flag"] == "2"
|
|
assert "variant_id" in fields["form_fields"]
|
|
assert fields["form_fields"]["__event"] == "ES01_003_001"
|
|
assert fields["form_fields"]["inventory_flag"] == "2"
|
|
# quantity_field / variant_field 是约定字段名
|
|
assert fields["quantity_field"] == "units"
|
|
assert fields["variant_field"] == "variant_id"
|
|
|
|
|
|
def test_extract_fields_single_inventory_uses_item_variant_id():
|
|
"""单规格商品应直接用 item.variantId，不需要选 sku.variants"""
|
|
state = {
|
|
"item": {"itemId": 999, "variantId": "v-1"},
|
|
"purchase": {
|
|
"sku": {"inventoryType": "single", "variants": []},
|
|
"sellType": {
|
|
"normalPurchase": {
|
|
"basketDomain": "https://example.co.jp/add",
|
|
"purchaseCondition": "enabled",
|
|
}
|
|
},
|
|
"information": {},
|
|
},
|
|
"shop": {"information": {"shopId": 42}},
|
|
}
|
|
fields = _extract_purchase_fields(state, intent_override={})
|
|
assert fields["inventory_flag"] == "1"
|
|
assert fields["form_fields"]["variant_id"] == "v-1"
|
|
|
|
|
|
def test_extract_fields_intent_overrides_variant_id():
|
|
"""调用方显式给 variant_id 时，跳过自动选"""
|
|
state = json.loads((FIXTURES / "item_state.json").read_text(encoding="utf-8"))
|
|
fields = _extract_purchase_fields(state, intent_override={"variant_id": "my-choice"})
|
|
assert fields["form_fields"]["variant_id"] == "my-choice"
|
|
|
|
|
|
def test_extract_fields_multi_inventory_no_variants_returns_no_variant():
|
|
"""多规格但 variants 空：不强行填 variant_id；调用方靠这个判断不可加购"""
|
|
state = {
|
|
"item": {"itemId": 1},
|
|
"purchase": {
|
|
"sku": {"inventoryType": "multiple", "variants": []},
|
|
"sellType": {"normalPurchase": {"basketDomain": "https://x/add", "purchaseCondition": "enabled"}},
|
|
"information": {},
|
|
},
|
|
"shop": {"information": {"shopId": 1}},
|
|
}
|
|
fields = _extract_purchase_fields(state, intent_override={})
|
|
assert "variant_id" not in fields["form_fields"]
|
|
|
|
|
|
def test_extract_fields_required_options_auto_picks_first_value():
|
|
"""有必填选项时，自动用第一个候选值填 choice"""
|
|
state = {
|
|
"item": {"itemId": 1, "variantId": "v"},
|
|
"purchase": {
|
|
"sku": {"inventoryType": "single"},
|
|
"sellType": {"normalPurchase": {"basketDomain": "https://x/add", "purchaseCondition": "enabled"}},
|
|
"information": {
|
|
"options": [
|
|
{
|
|
"id": 1,
|
|
"name": "サイズ",
|
|
"type": "select",
|
|
"isRequired": True,
|
|
"values": [
|
|
{"id": 10, "name": "S"},
|
|
{"id": 11, "name": "M"},
|
|
],
|
|
}
|
|
]
|
|
},
|
|
},
|
|
"shop": {"information": {"shopId": 1}},
|
|
}
|
|
fields = _extract_purchase_fields(state, intent_override={})
|
|
assert fields["has_required_options"] is True
|
|
assert fields["form_fields"]["choice"] == "サイズ:S"
|
|
|
|
|
|
def test_extract_fields_intent_choice_override_accepts_list_and_str():
|
|
state = {
|
|
"item": {"itemId": 1, "variantId": "v"},
|
|
"purchase": {
|
|
"sku": {"inventoryType": "single"},
|
|
"sellType": {"normalPurchase": {"basketDomain": "https://x/add", "purchaseCondition": "enabled"}},
|
|
"information": {"options": [{"name": "x", "isRequired": True, "values": [{"name": "a"}]}]},
|
|
},
|
|
"shop": {"information": {"shopId": 1}},
|
|
}
|
|
# list 形态
|
|
fields = _extract_purchase_fields(state, intent_override={"choice": ["サイズ:M", "色:赤"]})
|
|
assert fields["form_fields"]["choice"] == "サイズ:M,色:赤"
|
|
# str 形态
|
|
fields = _extract_purchase_fields(state, intent_override={"choice": "custom:value"})
|
|
assert fields["form_fields"]["choice"] == "custom:value"
|
|
|
|
|
|
def test_extract_fields_basket_domain_unescapes_u002f():
|
|
"""JSON encoded 的 / 在 __INITIAL_STATE__ 里可能是 \\u002F，要还原"""
|
|
state = {
|
|
"item": {"itemId": 1, "variantId": "v"},
|
|
"purchase": {
|
|
"sku": {"inventoryType": "single"},
|
|
"sellType": {
|
|
"normalPurchase": {
|
|
"basketDomain": "https:\\u002F\\u002Fx.example\\u002Fadd",
|
|
"purchaseCondition": "enabled",
|
|
}
|
|
},
|
|
"information": {},
|
|
},
|
|
"shop": {"information": {"shopId": 1}},
|
|
}
|
|
fields = _extract_purchase_fields(state, intent_override={})
|
|
assert fields["basket_domain"] == "https://x.example/add"
|
|
|
|
|
|
# ---- _extract_error_message ----
|
|
|
|
|
|
def test_extract_error_message_picks_visible_text():
|
|
body = """
|
|
<html><body>
|
|
<div>「商品 X」は商品情報が変更されました。</div>
|
|
<div>最新の商品ページから再度かごに追加してください。</div>
|
|
<div>ご注文手続き中の通信は、SSLによりすべて暗号化されます</div>
|
|
<div>© Rakuten Group, Inc.</div>
|
|
</body></html>
|
|
"""
|
|
msg = _extract_error_message(body)
|
|
# 应当包含前两条核心提示；SSL 与版权被过滤
|
|
assert "商品情報が変更されました" in msg
|
|
assert "SSL" not in msg
|
|
assert "Rakuten Group" not in msg
|
|
|
|
|
|
def test_extract_error_message_filters_punctuation_heavy_strings():
|
|
body = "<div>未選択の項目からどれか１つ選んでください。</div><div>{json: 'x'}</div>"
|
|
msg = _extract_error_message(body)
|
|
assert "未選択" in msg
|
|
# 含 { 与 ' 的 JSON 片段应被过滤
|
|
assert "json" not in msg
|
|
|
|
|
|
def test_extract_error_message_returns_empty_when_no_meaningful_text():
|
|
body = "<html></html>"
|
|
assert _extract_error_message(body) == ""
|
|
|
|
|
|
# ---- CheckoutSummary 仍是 dataclass ----
|
|
|
|
|
|
def test_checkout_summary_defaults():
|
|
s = CheckoutSummary(payable_yen=1000)
|
|
assert s.payable_yen == 1000
|
|
assert s.site_order_id is None
|
|
assert s.pay_deadline is None
|
|
|
|
|
|
# ---- SiteInteractor 构造（不调 start，无 Playwright 依赖）----
|
|
|
|
|
|
def test_site_interactor_construction_does_not_require_playwright():
|
|
"""构造 SiteInteractor 不应触发 Playwright import；start() 才会"""
|
|
site = SiteInteractor(auth_session=None, settings=None) # type: ignore[arg-type]
|
|
assert site._playwright is None
|
|
assert site._browser is None
|
|
assert site._context is None
|
|
assert site._per_task_state == {}
|
|
|
|
|
|
# ---- _parse_order_status：付款后监控的解析核心，2026-08-13 用真实订单号
|
|
# 306087-20260813-0863947697 的详情页 HTML 验证过（见 _ORDER_STEPPER_ITEM_PATTERN
|
|
# 上方注释）。下面 fixture 里的 4 个 <li> 是从真实页面摘录的进度条结构原样保留，
|
|
# 只是把每个 class 里易变的 CSS modules hash 换成了固定占位，不影响所依赖的
|
|
# `-active--` 中缀。
|
|
|
|
_REAL_ORDER_ID = "306087-20260813-0863947697"
|
|
|
|
|
|
def _stepper_html(*, active_stage: str | None, order_id: str = _REAL_ORDER_ID) -> str:
|
|
"""构造「订单号 + 4 阶段进度条」的最小 fixture，active_stage 指定哪一阶带 -active-- class"""
|
|
stages = ["ショップ", "出荷", "配達店", "配達完了"]
|
|
items = []
|
|
for i, stage in enumerate(stages):
|
|
active_cls = " item-shipping-active--1mu0i" if stage == active_stage else ""
|
|
items.append(
|
|
f'<li class="item--3gWCU item-{i}--hash title-m--3FZT3{active_cls}">'
|
|
f'<div class="title--2uGVi">{stage}</div></li>'
|
|
)
|
|
return f"<div>注文番号：{order_id}</div><ul>{''.join(items)}</ul>"
|
|
|
|
|
|
def test_parse_order_status_not_found_when_order_id_missing():
|
|
snapshot = _parse_order_status(_stepper_html(active_stage="ショップ", order_id="999999-x"), _REAL_ORDER_ID)
|
|
assert snapshot.found is False
|
|
assert snapshot.order_state is None
|
|
|
|
|
|
def test_parse_order_status_shop_stage_has_no_order_state_mapping():
|
|
"""「ショップ」（已接单未发货）阶段没有对应的 OrderState——已经在 ORDERED/PAID 报过了"""
|
|
snapshot = _parse_order_status(_stepper_html(active_stage="ショップ"), _REAL_ORDER_ID)
|
|
assert snapshot.found is True
|
|
assert snapshot.stage_label == "ショップ"
|
|
assert snapshot.order_state is None
|
|
|
|
|
|
def test_parse_order_status_shipped_stage_maps_to_shipped():
|
|
snapshot = _parse_order_status(_stepper_html(active_stage="出荷"), _REAL_ORDER_ID)
|
|
assert snapshot.order_state == OrderState.SHIPPED
|
|
|
|
|
|
def test_parse_order_status_depot_stage_also_maps_to_shipped():
|
|
"""「配達店」（配送网点中转）没有单独状态，归入 SHIPPED"""
|
|
snapshot = _parse_order_status(_stepper_html(active_stage="配達店"), _REAL_ORDER_ID)
|
|
assert snapshot.order_state == OrderState.SHIPPED
|
|
|
|
|
|
def test_parse_order_status_delivered_stage_maps_to_delivered():
|
|
snapshot = _parse_order_status(_stepper_html(active_stage="配達完了"), _REAL_ORDER_ID)
|
|
assert snapshot.order_state == OrderState.DELIVERED
|
|
|
|
|
|
def test_parse_order_status_no_active_marker_returns_found_without_state():
|
|
"""进度条 4 项都没有 -active-- class（页面结构变了/解析不出）：found 但 order_state=None"""
|
|
snapshot = _parse_order_status(_stepper_html(active_stage=None), _REAL_ORDER_ID)
|
|
assert snapshot.found is True
|
|
assert snapshot.order_state is None
|
|
assert snapshot.stage_label is None
|
|
|
|
|
|
def test_order_status_snapshot_defaults():
|
|
s = OrderStatusSnapshot(found=False)
|
|
assert s.stage_label is None
|
|
assert s.order_state is None
|
|
assert s.html == ""
|
|
|
|
|
|
# ---- check_order_status 在没启动 Playwright 时应失败 ----
|
|
|
|
|
|
async def test_check_order_status_without_start_raises():
|
|
site = SiteInteractor(auth_session=None, settings=None) # type: ignore[arg-type]
|
|
with pytest.raises((AttributeError, TypeError)):
|
|
await site.check_order_status(_REAL_ORDER_ID)
|
|
|
|
|
|
# ---- auth_site.looks_logged_out：业务页上的单边「像是掉登录」判据 ----
|
|
|
|
|
|
def test_looks_logged_out_detects_sso_redirect():
|
|
"""订单页落地到 SSO 域 → 判为像掉登录"""
|
|
assert auth_site.looks_logged_out(
|
|
"rakuten",
|
|
final_url="https://login.account.rakuten.com/sso/authorize?client_id=x",
|
|
body="<html>ログイン</html>",
|
|
)
|
|
|
|
|
|
def test_looks_logged_out_ignores_session_upgrade():
|
|
"""session upgrade 是「已登录但要求复核密码」的站点风控，不是掉登录
|
|
|
|
误判会把风控拦截当 cookie 过期，白跑一次重登还掩盖真实原因。
|
|
"""
|
|
assert not auth_site.looks_logged_out(
|
|
"rakuten",
|
|
final_url="https://login.account.rakuten.com/sign_in/session/upgrade?client_id=x",
|
|
body="<html>パスワードを入力</html>",
|
|
)
|
|
|
|
|
|
def test_looks_logged_out_detects_legacy_marker_on_business_page():
|
|
"""业务页正文出现旧版未登录 marker（SSR 落地 HTML）→ 判为像掉登录"""
|
|
assert auth_site.looks_logged_out(
|
|
"rakuten",
|
|
final_url="https://order.my.rakuten.co.jp/purchase-history/order-list",
|
|
body=f"<html>{auth_site.RAKUTEN_LOGGED_OUT_MARKER}</html>",
|
|
)
|
|
|
|
|
|
def test_looks_logged_out_false_on_normal_order_page():
|
|
"""正常订单页没有任何未登录信号 → False（注意这**不**等于「确认登录着」）"""
|
|
assert not auth_site.looks_logged_out(
|
|
"rakuten",
|
|
final_url="https://order.my.rakuten.co.jp/purchase-history/order-list",
|
|
body=_stepper_html(active_stage="出荷"),
|
|
)
|
|
|
|
|
|
def test_looks_logged_out_rejects_unknown_site():
|
|
with pytest.raises(ValueError):
|
|
auth_site.looks_logged_out("rakuma", final_url="https://x/", body="")
|
|
|
|
|
|
# ---- 只读操作的「执行中掉登录 → 重登一次 → 重跑」----
|
|
#
|
|
# 用最小替身顶掉 Playwright（_FakeContext/_FakeOrderPage）与 AuthSession
|
|
# （_FakeAuthSession），测的是 SiteInteractor 自己的重试编排：什么时候判掉登录、
|
|
# 重登几次、重登失败怎么收场。页面解析由上面那批 _parse_* 纯函数测试覆盖。
|
|
|
|
|
|
class _FakeOrderPage:
|
|
"""只实现 goto/content/url/close 的 page 替身；每次 goto 按脚本换一份响应"""
|
|
|
|
def __init__(self, responses: list[tuple[str, str]]):
|
|
# responses: [(final_url, html), ...]，按 goto 次序消费，用尽后重复最后一份
|
|
self._responses = responses
|
|
self.goto_urls: list[str] = []
|
|
self.url = ""
|
|
self._html = ""
|
|
self.closed = False
|
|
|
|
async def goto(self, url: str, **kwargs: Any) -> None:
|
|
self.goto_urls.append(url)
|
|
idx = min(len(self.goto_urls) - 1, len(self._responses) - 1)
|
|
self.url, self._html = self._responses[idx]
|
|
|
|
async def wait_for_timeout(self, ms: int) -> None:
|
|
return None
|
|
|
|
async def content(self) -> str:
|
|
return self._html
|
|
|
|
async def close(self) -> None:
|
|
self.closed = True
|
|
|
|
|
|
class _FakeContext:
|
|
"""new_page() 按脚本发页面；页面用尽则复用最后一个"""
|
|
|
|
def __init__(self, pages: list[_FakeOrderPage]):
|
|
self._pages = pages
|
|
self.handed_out: list[_FakeOrderPage] = []
|
|
|
|
async def new_page(self) -> _FakeOrderPage:
|
|
page = self._pages[min(len(self.handed_out), len(self._pages) - 1)]
|
|
self.handed_out.append(page)
|
|
return page
|
|
|
|
|
|
class _FakeAuthSession:
|
|
"""记录 require_logged_in / try_relogin 调用次数的 AuthSession 替身"""
|
|
|
|
def __init__(self, *, relogin_ok: bool = True):
|
|
self.relogin_ok = relogin_ok
|
|
self.require_calls = 0
|
|
self.relogin_calls = 0
|
|
|
|
async def require_logged_in(self, site: str) -> None:
|
|
self.require_calls += 1
|
|
|
|
async def try_relogin(self, site: str) -> bool:
|
|
self.relogin_calls += 1
|
|
return self.relogin_ok
|
|
|
|
|
|
_SSO_URL = "https://login.account.rakuten.com/sso/authorize?client_id=x"
|
|
_ORDER_DETAIL_LANDED_URL = "https://order.my.rakuten.co.jp/purchase-history/?order_number=x"
|
|
_ORDER_LIST_LANDED_URL = "https://order.my.rakuten.co.jp/purchase-history/order-list"
|
|
|
|
|
|
def _build_site(tmp_path: Path, context: _FakeContext, auth: _FakeAuthSession) -> SiteInteractor:
|
|
"""装配一个不依赖 Playwright 的 SiteInteractor
|
|
|
|
auth_state_dir 指向空目录：_refresh_context_if_stale 见不到 storage_state
|
|
文件就直接返回，不会试图重建 context（重建需要真实 browser）。
|
|
"""
|
|
from app.shared.config import Settings
|
|
|
|
site = SiteInteractor(
|
|
auth_session=auth, # type: ignore[arg-type]
|
|
settings=Settings(auth_state_dir=str(tmp_path / "auth"), evidence_dir=str(tmp_path)),
|
|
)
|
|
site._context = context # type: ignore[assignment]
|
|
return site
|
|
|
|
|
|
async def test_check_order_status_relogins_and_retries_when_kicked_to_sso(tmp_path):
|
|
"""详情页被踢到 SSO → 重登一次 → 重跑拿到真实快照"""
|
|
pages = [
|
|
_FakeOrderPage([(_SSO_URL, "<html>ログイン</html>")]),
|
|
_FakeOrderPage([(_ORDER_DETAIL_LANDED_URL, _stepper_html(active_stage="出荷"))]),
|
|
]
|
|
auth = _FakeAuthSession(relogin_ok=True)
|
|
site = _build_site(tmp_path, _FakeContext(pages), auth)
|
|
|
|
snapshot = await site.check_order_status(_REAL_ORDER_ID)
|
|
|
|
assert snapshot.found is True
|
|
assert snapshot.order_state == OrderState.SHIPPED
|
|
assert auth.relogin_calls == 1
|
|
assert auth.require_calls == 2 # 每次 attempt 前都过一遍前置检查
|
|
assert all(p.closed for p in pages) # 两次 attempt 的 page 都关掉了
|
|
|
|
|
|
async def test_check_order_status_raises_not_logged_in_when_relogin_fails(tmp_path):
|
|
"""重登失败 → 抛 NotLoggedInError，而不是把「被踢走」伪装成 found=False"""
|
|
page = _FakeOrderPage([(_SSO_URL, "<html>ログイン</html>")])
|
|
auth = _FakeAuthSession(relogin_ok=False)
|
|
site = _build_site(tmp_path, _FakeContext([page]), auth)
|
|
|
|
with pytest.raises(NotLoggedInError):
|
|
await site.check_order_status(_REAL_ORDER_ID)
|
|
assert auth.relogin_calls == 1
|
|
|
|
|
|
async def test_check_order_status_raises_not_logged_in_when_still_kicked_after_relogin(tmp_path):
|
|
"""重登「成功」了但页面还是被踢走 → 只重试一次就抛错，不无限重登"""
|
|
page = _FakeOrderPage([(_SSO_URL, "<html>ログイン</html>")])
|
|
auth = _FakeAuthSession(relogin_ok=True)
|
|
site = _build_site(tmp_path, _FakeContext([page]), auth)
|
|
|
|
with pytest.raises(NotLoggedInError):
|
|
await site.check_order_status(_REAL_ORDER_ID)
|
|
assert auth.relogin_calls == 1
|
|
assert len(page.goto_urls) == 2 # 只跑了两轮
|
|
|
|
|
|
async def test_check_order_status_order_not_yet_visible_is_not_treated_as_logged_out(tmp_path):
|
|
"""订单号还没反映到详情页（站点自己说要等 10 分钟）不是掉登录：不重登，返回 found=False"""
|
|
page = _FakeOrderPage([
|
|
(_ORDER_DETAIL_LANDED_URL, "<html>ご注文の反映に10分ほどかかります</html>")
|
|
])
|
|
auth = _FakeAuthSession(relogin_ok=True)
|
|
site = _build_site(tmp_path, _FakeContext([page]), auth)
|
|
|
|
snapshot = await site.check_order_status(_REAL_ORDER_ID)
|
|
|
|
assert snapshot.found is False
|
|
assert auth.relogin_calls == 0
|
|
|
|
|
|
async def test_list_recent_orders_relogins_and_retries_from_first_page(tmp_path):
|
|
"""列表页第一页拿不到结构化数据且像掉登录 → 重登后从 page 1 重跑"""
|
|
logged_out_page = _FakeOrderPage([(_SSO_URL, "<html>ログイン</html>")])
|
|
good_html = _wrap_state(_real_order_list_state())
|
|
good_page = _FakeOrderPage([(_ORDER_LIST_LANDED_URL, good_html)])
|
|
auth = _FakeAuthSession(relogin_ok=True)
|
|
site = _build_site(tmp_path, _FakeContext([logged_out_page, good_page]), auth)
|
|
|
|
window = await site.list_recent_orders(since=datetime(2026, 8, 1, tzinfo=timezone.utc))
|
|
|
|
assert [e.order_number for e in window.entries] == [_REAL_ORDER_ID]
|
|
assert window.window_fully_covered is True
|
|
assert auth.relogin_calls == 1
|
|
# 重跑确实是从第一页开始，不是接着上次的页码
|
|
assert good_page.goto_urls[0].endswith("/order-list")
|
|
|
|
|
|
async def test_list_recent_orders_page_structure_change_is_not_treated_as_logged_out(tmp_path):
|
|
"""第一页拿不到 ph-list 但没有任何未登录信号（页面改版）→ 不重登，按原路径转「没覆盖完」"""
|
|
page = _FakeOrderPage([(_ORDER_LIST_LANDED_URL, "<html><body>改版了</body></html>")])
|
|
auth = _FakeAuthSession(relogin_ok=True)
|
|
site = _build_site(tmp_path, _FakeContext([page]), auth)
|
|
|
|
window = await site.list_recent_orders(since=datetime(2026, 8, 1, tzinfo=timezone.utc))
|
|
|
|
assert window.entries == []
|
|
assert window.window_fully_covered is False
|
|
assert auth.relogin_calls == 0
|
|
|
|
|
|
async def test_read_retry_does_not_swallow_navigation_failure(tmp_path):
|
|
"""导航失败仍是 OrderOperationError，不会被重登重试路径吃掉"""
|
|
|
|
class _FailingPage(_FakeOrderPage):
|
|
async def goto(self, url: str, **kwargs: Any) -> None:
|
|
raise RuntimeError("net::ERR_TIMED_OUT")
|
|
|
|
page = _FailingPage([(_ORDER_LIST_LANDED_URL, "")])
|
|
auth = _FakeAuthSession(relogin_ok=True)
|
|
site = _build_site(tmp_path, _FakeContext([page]), auth)
|
|
|
|
with pytest.raises(OrderOperationError):
|
|
await site.list_recent_orders(since=datetime(2026, 8, 1, tzinfo=timezone.utc))
|
|
assert auth.relogin_calls == 0
|
|
|
|
|
|
# ---- _parse_order_list：verify_on_site 恢复核对用，2026-08-13 用真实账号跑
|
|
# order.my.rakuten.co.jp/purchase-history/order-list 验证过这套 __INITIAL_STATE__
|
|
# 结构（pageType="ph-list" 时 orderListData 直接是结构化 JSON，不需要正则抠 DOM）。
|
|
# 下面 fixture 里的字段名/嵌套结构与真实响应一致，只是精简掉了大量与匹配逻辑
|
|
# 无关的字段（推荐组件、通知、会员信息等）。
|
|
|
|
|
|
def _real_order_list_state(
|
|
*,
|
|
orders_found: int = 1,
|
|
page_size: int = 25,
|
|
orders: list[dict] | None = None,
|
|
page_type: str = "ph-list",
|
|
) -> dict:
|
|
if orders is None:
|
|
orders = [
|
|
{
|
|
"orderNumber": _REAL_ORDER_ID,
|
|
"orderDate": "2026-08-13T09:47:28.000Z",
|
|
"shopId": 306087,
|
|
"shopName": "BACKYARD FAMILY インテリアタウン",
|
|
"items": [
|
|
{
|
|
"itemId": 10422789,
|
|
"itemName": "ウォールフック 粘着フック",
|
|
"itemUrl": "https://item.rakuten.co.jp/moccasin/ds001iwrgesaaa2/?variantId=ds001iwrgesaaa2-1a-2a",
|
|
"itemPrice": 297,
|
|
}
|
|
],
|
|
}
|
|
]
|
|
return {
|
|
"pageType": page_type,
|
|
"orderListData": {
|
|
"ordersFound": orders_found,
|
|
"pageSize": page_size,
|
|
"orderList": orders,
|
|
},
|
|
}
|
|
|
|
|
|
def test_parse_order_list_extracts_entries_from_real_shape():
|
|
html = _wrap_state(_real_order_list_state())
|
|
page = _parse_order_list(html)
|
|
assert page.orders_found == 1
|
|
assert page.page_size == 25
|
|
assert len(page.entries) == 1
|
|
entry = page.entries[0]
|
|
assert entry.order_number == _REAL_ORDER_ID
|
|
assert entry.order_date == "2026-08-13T09:47:28.000Z"
|
|
assert entry.shop_id == 306087
|
|
assert len(entry.items) == 1
|
|
assert entry.items[0].item_url == (
|
|
"https://item.rakuten.co.jp/moccasin/ds001iwrgesaaa2/?variantId=ds001iwrgesaaa2-1a-2a"
|
|
)
|
|
|
|
|
|
def test_parse_order_list_multiple_items_in_one_order():
|
|
orders = [
|
|
{
|
|
"orderNumber": "111-20260101-0000000001",
|
|
"orderDate": "2026-01-01T00:00:00.000Z",
|
|
"items": [
|
|
{"itemUrl": "https://item.rakuten.co.jp/shop/a/", "itemName": "A"},
|
|
{"itemUrl": "https://item.rakuten.co.jp/shop/b/", "itemName": "B"},
|
|
],
|
|
}
|
|
]
|
|
html = _wrap_state(_real_order_list_state(orders=orders))
|
|
page = _parse_order_list(html)
|
|
assert len(page.entries[0].items) == 2
|
|
|
|
|
|
def test_parse_order_list_wrong_page_type_returns_empty_and_no_orders_found():
|
|
"""pageType 不是 ph-list（比如命中了 detail_page_view 的错误分支）：拿不到结构化数据"""
|
|
html = _wrap_state(_real_order_list_state(page_type="ph-error"))
|
|
page = _parse_order_list(html)
|
|
assert page.entries == []
|
|
assert page.orders_found is None
|
|
|
|
|
|
def test_parse_order_list_no_initial_state_returns_empty():
|
|
page = _parse_order_list("<html>没有 state 的页面</html>")
|
|
assert page.entries == []
|
|
assert page.orders_found is None
|
|
|
|
|
|
def test_parse_order_list_zero_orders_still_has_orders_found():
|
|
"""账号没有任何订单：orderList 空，但 orders_found=0（不是 None）——区分「真的没有」与「解析不出」"""
|
|
html = _wrap_state(_real_order_list_state(orders_found=0, orders=[]))
|
|
page = _parse_order_list(html)
|
|
assert page.entries == []
|
|
assert page.orders_found == 0
|
|
|
|
|
|
# ---- parse_order_datetime ----
|
|
|
|
|
|
def test_parse_order_datetime_parses_real_format():
|
|
dt = parse_order_datetime("2026-08-13T09:47:28.000Z")
|
|
assert dt == datetime(2026, 8, 13, 9, 47, 28, tzinfo=timezone.utc)
|
|
|
|
|
|
def test_parse_order_datetime_none_and_invalid():
|
|
assert parse_order_datetime(None) is None
|
|
assert parse_order_datetime("") is None
|
|
assert parse_order_datetime("不是日期") is None
|
|
|
|
|
|
# ---- _accumulate_order_list_page：list_recent_orders 翻页决策的纯函数核心，
|
|
# 2026-08-13 只有「账号 1 笔订单、1 页」的真实数据支撑第一个分支，其余分支
|
|
# （多页、命中窗口边界、拿不到结构化数据）目前只验证逻辑正确，未经真实多页数据跑过 ----
|
|
|
|
|
|
def _page(entries: list[OrderListEntry], *, orders_found: int | None) -> OrderListPage:
|
|
return OrderListPage(entries=entries, orders_found=orders_found, page_size=25)
|
|
|
|
|
|
def _entry(order_number: str, order_date: str) -> OrderListEntry:
|
|
return OrderListEntry(order_number=order_number, order_date=order_date, items=[])
|
|
|
|
|
|
_SINCE = datetime(2026, 8, 1, tzinfo=timezone.utc)
|
|
|
|
|
|
def test_accumulate_single_page_all_within_window_and_matches_orders_found():
|
|
acc = _OrderListAccumulator()
|
|
page = _page([_entry("o1", "2026-08-10T00:00:00Z")], orders_found=1)
|
|
new_acc, stop = _accumulate_order_list_page(acc, page, since=_SINCE)
|
|
assert stop is True
|
|
assert new_acc.gave_up is False
|
|
assert [e.order_number for e in new_acc.entries] == ["o1"]
|
|
|
|
|
|
def test_accumulate_stops_when_oldest_entry_older_than_since():
|
|
"""翻到比 since 更早的订单：窗口内数据已经看全，停止翻页且视为覆盖完整"""
|
|
acc = _OrderListAccumulator(entries=[_entry("o1", "2026-08-10T00:00:00Z")], total_found=5, is_first_page=False)
|
|
page = _page([_entry("o2", "2026-07-01T00:00:00Z")], orders_found=5)
|
|
new_acc, stop = _accumulate_order_list_page(acc, page, since=_SINCE)
|
|
assert stop is True
|
|
assert new_acc.gave_up is False
|
|
assert [e.order_number for e in new_acc.entries] == ["o1", "o2"]
|
|
|
|
|
|
def test_accumulate_continues_when_more_pages_remain():
|
|
"""orders_found 还没凑够、最老的一条也没早于 since：应该继续翻下一页"""
|
|
acc = _OrderListAccumulator()
|
|
page = _page([_entry("o1", "2026-08-10T00:00:00Z")], orders_found=3)
|
|
new_acc, stop = _accumulate_order_list_page(acc, page, since=_SINCE)
|
|
assert stop is False
|
|
assert new_acc.total_found == 3
|
|
|
|
|
|
def test_accumulate_empty_page_means_fully_covered():
|
|
"""翻到空页（没有更多订单了）：不是错误，视为窗口已覆盖完"""
|
|
acc = _OrderListAccumulator(total_found=1, is_first_page=False)
|
|
page = _page([], orders_found=1)
|
|
new_acc, stop = _accumulate_order_list_page(acc, page, since=_SINCE)
|
|
assert stop is True
|
|
assert new_acc.gave_up is False
|
|
|
|
|
|
def test_accumulate_first_page_missing_orders_found_gives_up():
|
|
"""第一页就拿不到结构化数据（orders_found=None）：停止翻页，但 gave_up=True，不算覆盖完"""
|
|
acc = _OrderListAccumulator()
|
|
page = _page([], orders_found=None)
|
|
new_acc, stop = _accumulate_order_list_page(acc, page, since=_SINCE)
|
|
assert stop is True
|
|
assert new_acc.gave_up is True
|
|
|
|
|
|
# ---- list_recent_orders 在没启动 Playwright 时应失败 ----
|
|
|
|
|
|
async def test_list_recent_orders_without_start_raises():
|
|
site = SiteInteractor(auth_session=None, settings=None) # type: ignore[arg-type]
|
|
with pytest.raises((AttributeError, TypeError)):
|
|
await site.list_recent_orders(since=_SINCE)
|
|
|
|
|
|
# ---- 账号只读查询通道用到的两条只读路径（docs/order-gateway.md §11）----
|
|
#
|
|
# 查询接口不新写解析，复用的就是上面这两条已实测路径；这里补的是「站点原始
|
|
# JSON 有没有被完整带出来」和「翻页上限有没有生效」——这两点是查询接口独有的，
|
|
# 恢复核对那条老路径不关心。
|
|
|
|
|
|
def test_parse_order_list_keeps_raw_order_list_data():
|
|
"""规范化字段之外，站点 orderListData 原文要原样留着供上游取用"""
|
|
html = _wrap_state(_real_order_list_state())
|
|
page = _parse_order_list(html)
|
|
assert page.raw is not None
|
|
assert page.raw["ordersFound"] == 1
|
|
# 规范化模型里没有的字段也在（这正是「原样透传」的意义）
|
|
assert page.raw["orderList"][0]["shopName"] == "BACKYARD FAMILY インテリアタウン"
|
|
|
|
|
|
def test_parse_order_list_raw_is_none_when_page_type_wrong():
|
|
"""不是 ph-list（改版 / 掉登录 / act 分支不对）：raw 为 None，不给上游半截数据"""
|
|
html = _wrap_state(_real_order_list_state(page_type="ph-detail"))
|
|
assert _parse_order_list(html).raw is None
|
|
|
|
|
|
def test_accumulate_collects_raw_pages_in_order():
|
|
"""多页翻页时，每页的原始 JSON 按顺序累积"""
|
|
acc = _OrderListAccumulator()
|
|
page1 = OrderListPage(
|
|
entries=[_entry("o1", "2026-08-10T00:00:00Z")], orders_found=3, raw={"page": 1}
|
|
)
|
|
acc, stop = _accumulate_order_list_page(acc, page1, since=_SINCE)
|
|
assert stop is False
|
|
page2 = OrderListPage(
|
|
entries=[_entry("o2", "2026-07-01T00:00:00Z")], orders_found=3, raw={"page": 2}
|
|
)
|
|
acc, stop = _accumulate_order_list_page(acc, page2, since=_SINCE)
|
|
assert stop is True
|
|
assert acc.raw_pages == [{"page": 1}, {"page": 2}]
|
|
|
|
|
|
async def test_list_recent_orders_respects_max_pages(tmp_path):
|
|
"""max_pages 是硬上限：翻到上限就停，且必须如实报 window_fully_covered=False"""
|
|
# 每页都还有更多订单（orders_found 远大于已取回数），正常会一直翻下去
|
|
state = _real_order_list_state(orders_found=99)
|
|
page = _FakeOrderPage([(_ORDER_LIST_LANDED_URL, _wrap_state(state))])
|
|
site = _build_site(tmp_path, _FakeContext([page]), _FakeAuthSession())
|
|
|
|
window = await site.list_recent_orders(since=_SINCE, max_pages=2)
|
|
|
|
assert len(page.goto_urls) == 2
|
|
assert page.goto_urls[1].endswith("?page=2")
|
|
assert window.window_fully_covered is False
|
|
assert len(window.raw_pages) == 2
|
|
|
|
|
|
async def test_list_recent_orders_window_carries_raw_pages(tmp_path):
|
|
html = _wrap_state(_real_order_list_state())
|
|
page = _FakeOrderPage([(_ORDER_LIST_LANDED_URL, html)])
|
|
site = _build_site(tmp_path, _FakeContext([page]), _FakeAuthSession())
|
|
|
|
window = await site.list_recent_orders(since=_SINCE)
|
|
|
|
assert window.window_fully_covered is True
|
|
assert [raw["ordersFound"] for raw in window.raw_pages] == [1]
|
|
|
|
|
|
async def test_fetch_order_detail_returns_status_and_raw_state(tmp_path):
|
|
"""详情页：已实测的配送阶段照常解析，页面原始状态原样带出"""
|
|
html = _wrap_state({"pageType": "ph-detail", "whatever": {"a": 1}}) + _stepper_html(
|
|
active_stage="出荷"
|
|
)
|
|
page = _FakeOrderPage([(_ORDER_DETAIL_LANDED_URL, html)])
|
|
site = _build_site(tmp_path, _FakeContext([page]), _FakeAuthSession())
|
|
|
|
detail = await site.fetch_order_detail(_REAL_ORDER_ID)
|
|
|
|
assert detail.status.found is True
|
|
assert detail.status.order_state == OrderState.SHIPPED
|
|
assert detail.raw == {"pageType": "ph-detail", "whatever": {"a": 1}}
|
|
|
|
|
|
async def test_fetch_order_detail_raw_is_none_without_inline_state(tmp_path):
|
|
"""页面没有内联状态时 raw=None——不编造，也不因此把这次读取判成失败"""
|
|
page = _FakeOrderPage([(_ORDER_DETAIL_LANDED_URL, _stepper_html(active_stage="出荷"))])
|
|
site = _build_site(tmp_path, _FakeContext([page]), _FakeAuthSession())
|
|
|
|
detail = await site.fetch_order_detail(_REAL_ORDER_ID)
|
|
|
|
assert detail.raw is None
|
|
assert detail.status.found is True
|
|
|
|
|
|
async def test_check_order_status_still_returns_only_status(tmp_path):
|
|
"""付款后监控那条老路径不受影响：仍然拿到 OrderStatusSnapshot"""
|
|
page = _FakeOrderPage([(_ORDER_DETAIL_LANDED_URL, _stepper_html(active_stage="配達完了"))])
|
|
site = _build_site(tmp_path, _FakeContext([page]), _FakeAuthSession())
|
|
|
|
snapshot = await site.check_order_status(_REAL_ORDER_ID)
|
|
|
|
assert isinstance(snapshot, OrderStatusSnapshot)
|
|
assert snapshot.order_state == OrderState.DELIVERED
|
|
|
|
|
|
|
|
# ---- submit_order / pay：没有 enter_checkout 留存的确认页会话时应报错，不静默成功 ----
|
|
|
|
|
|
async def test_submit_order_without_checkout_page_raises():
|
|
site = SiteInteractor(auth_session=None, settings=None) # type: ignore[arg-type]
|
|
with pytest.raises(OrderOperationError):
|
|
await site.submit_order(_make_task())
|
|
|
|
|
|
async def test_pay_without_checkout_page_raises():
|
|
site = SiteInteractor(auth_session=None, settings=None) # type: ignore[arg-type]
|
|
with pytest.raises(OrderOperationError):
|
|
await site.pay(_make_task(), "ord-1")
|
|
|
|
|
|
# ---- parse_checkout / _parse_checkout_summary：金额解析已用真实确认页 HTML 验证
|
|
# （2026-08-13，data/evidence/checkout-research-20260811/i-final-state.html），
|
|
# 订单号/付款期限仍未验证。下面简单 fixture 测的是防御性行为（找不到/矛盾就报错）。
|
|
|
|
|
|
async def test_parse_checkout_delegates_to_pure_function():
|
|
site = SiteInteractor(auth_session=None, settings=None) # type: ignore[arg-type]
|
|
html = "<div>お支払い金額 1,980円</div>"
|
|
summary = await site.parse_checkout(html)
|
|
assert summary.payable_yen == 1980
|
|
|
|
|
|
def test_parse_checkout_summary_extracts_amount_order_id_deadline():
|
|
"""订单号 fixture 用真实格式（三段数字用「-」连接，见 _ORDER_ID_PATTERN 注释），
|
|
2026-08-13 之前这里用的是纯猜测的 "AB-123456" 字母前缀格式，拿到真实订单号后
|
|
已确认不是这个形状。"""
|
|
html = (
|
|
"<div>お支払い合計 12,345円</div>"
|
|
"<div>ご注文番号 306087-20260813-0863947697</div>"
|
|
"<div>お支払い期限 2026/08/20</div>"
|
|
)
|
|
summary = _parse_checkout_summary(html)
|
|
assert summary.payable_yen == 12345
|
|
assert summary.site_order_id == "306087-20260813-0863947697"
|
|
assert summary.pay_deadline == "2026/08/20"
|
|
|
|
|
|
def test_parse_checkout_summary_missing_amount_raises():
|
|
"""找不到「标签+金额」的明确匹配就报错，不落回猜数字——这个值直接喂给金额守卫"""
|
|
with pytest.raises(OrderOperationError):
|
|
_parse_checkout_summary("<div>合計 1,980円</div>") # 标签不在候选列表里
|
|
|
|
|
|
def test_parse_checkout_summary_conflicting_amounts_raises():
|
|
html = "<div>お支払い金額 1,000円</div><div>お支払い合計 2,000円</div>"
|
|
with pytest.raises(OrderOperationError):
|
|
_parse_checkout_summary(html)
|
|
|
|
|
|
def test_parse_checkout_summary_optional_fields_default_none():
|
|
summary = _parse_checkout_summary("<div>お支払い金額 500円</div>")
|
|
assert summary.payable_yen == 500
|
|
assert summary.site_order_id is None
|
|
assert summary.pay_deadline is None
|
|
|
|
|
|
def test_parse_checkout_summary_matches_real_page_dom_shape():
|
|
"""真实确认页金额结构：标签纯「支払い金額」（无「お」前缀），数字与「円」
|
|
分别在独立标签里、中间隔着大段 class 属性，不是「标签+数字+円」紧邻文案
|
|
（2026-08-13 从真实页面简化摘录的结构，见 _AMOUNT_TAG_BOUNDED_PATTERN 注释）"""
|
|
html = (
|
|
'<span class="label--x">支払い金額</span></div>'
|
|
'<div class="number-display--x">'
|
|
'<div class="value--x">297</div>'
|
|
'<span class="suffix--x"><div class="text-display--x">円</div></span>'
|
|
"</div>"
|
|
)
|
|
summary = _parse_checkout_summary(html)
|
|
assert summary.payable_yen == 297
|
|
|
|
|
|
def test_parse_checkout_summary_matches_real_order_id_dom_shape():
|
|
"""真实下单完成页订单号结构：标签和号码之间是 ` ` 实体而不是普通空白，
|
|
号码是三段数字用「-」连接（2026-08-13 真实点击「注文を確定する」后从
|
|
完成页摘录的结构，见 _ORDER_ID_PATTERN 注释）。旧版正则的分隔符只认
|
|
非单词字符，会卡在 ` ` 里的 n/b/s/p 上、完全连不到号码。"""
|
|
html = (
|
|
'<div>お支払い金額 297円</div>'
|
|
'<span class="label--x">注文番号 306087-20260813-0863947697</span>'
|
|
)
|
|
summary = _parse_checkout_summary(html)
|
|
assert summary.site_order_id == "306087-20260813-0863947697"
|
|
|
|
|
|
# ---- enter_checkout / add_to_cart / verify_cart 在没启动 Playwright 时应失败 ----
|
|
|
|
|
|
async def test_enter_checkout_without_start_raises():
|
|
"""没调 start() 就调 enter_checkout，auth_session=None，应在 require_logged_in 处抛错
|
|
|
|
与 add_to_cart 同理：测的是「未启动时不应该静默成功」，不是 CheckoutBlockedError
|
|
这条业务分支（那条分支需要真实 Playwright page，不在离线单测范围）。
|
|
"""
|
|
site = SiteInteractor(auth_session=None, settings=None) # type: ignore[arg-type]
|
|
with pytest.raises((AttributeError, TypeError)):
|
|
await site.enter_checkout(_make_task())
|
|
|
|
|
|
async def test_add_to_cart_without_start_raises():
|
|
"""没调 start() 就调 add_to_cart，_context 是 None，应抛错"""
|
|
site = SiteInteractor(auth_session=None, settings=None) # type: ignore[arg-type]
|
|
# require_logged_in 会因为 auth_session=None 抛 AttributeError，比 _context None 更早
|
|
# 这里测的是「未启动时不应该静默成功」
|
|
with pytest.raises((AttributeError, TypeError)):
|
|
await site.add_to_cart(_make_task())
|
|
|
|
|
|
# ---- _dump_debug_snapshot：失败时落调试快照，本身出错不能污染原始异常 ----
|
|
|
|
|
|
class _FakePage:
|
|
"""不依赖 Playwright 的最小 page 替身，只实现 _dump_debug_snapshot 用到的接口"""
|
|
|
|
def __init__(self, *, html: str = "<html></html>", url: str = "https://example.test/x"):
|
|
self.url = url
|
|
self._html = html
|
|
self.screenshot_calls: list[str] = []
|
|
|
|
async def content(self) -> str:
|
|
return self._html
|
|
|
|
async def screenshot(self, *, path: str, full_page: bool = True) -> None:
|
|
self.screenshot_calls.append(path)
|
|
Path(path).write_bytes(b"fake-png")
|
|
|
|
|
|
class _BrokenPage(_FakePage):
|
|
"""content() 抛错，模拟页面已关闭等场景"""
|
|
|
|
async def content(self) -> str:
|
|
raise RuntimeError("page already closed")
|
|
|
|
|
|
async def test_dump_debug_snapshot_writes_html_and_png(tmp_path):
|
|
from app.shared.config import Settings
|
|
|
|
settings = Settings(evidence_dir=str(tmp_path))
|
|
site = SiteInteractor(auth_session=None, settings=settings) # type: ignore[arg-type]
|
|
page = _FakePage(html="<html>boom</html>")
|
|
|
|
await site._dump_debug_snapshot(page, task_id="t1", label="enter_checkout-error")
|
|
|
|
written = list((tmp_path / "t1").glob("debug-*"))
|
|
htmls = [p for p in written if p.suffix == ".html"]
|
|
pngs = [p for p in written if p.suffix == ".png"]
|
|
assert len(htmls) == 1
|
|
assert len(pngs) == 1
|
|
assert htmls[0].read_text(encoding="utf-8") == "<html>boom</html>"
|
|
|
|
|
|
async def test_dump_debug_snapshot_swallows_its_own_errors(tmp_path):
|
|
"""page 已经关闭、content() 抛错时，快照方法本身不能再抛，否则会掩盖原始异常"""
|
|
from app.shared.config import Settings
|
|
|
|
settings = Settings(evidence_dir=str(tmp_path))
|
|
site = SiteInteractor(auth_session=None, settings=settings) # type: ignore[arg-type]
|
|
page = _BrokenPage()
|
|
|
|
await site._dump_debug_snapshot(page, task_id="t1", label="pay-blocked")
|
|
|
|
assert list((tmp_path / "t1").glob("debug-*")) == []
|
|
|
|
|
|
# ---- helper ----
|