"""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_detail_status, _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 ( "" ) # ---- _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("no state here") is None def test_parse_initial_state_returns_none_on_invalid_json(): html = "" 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 = """

「商品 X」は商品情報が変更されました。

最新の商品ページから再度かごに追加してください。

ご注文手続き中の通信は、SSLによりすべて暗号化されます

""" 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 = "

未選択の項目からどれか１つ選んでください。

{json: 'x'}

" 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 = "" 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 个

是从真实页面摘录的进度条结构原样保留， # 只是把每个 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'

' f'

{stage}

' ) return f"

注文番号：{order_id}

{''.join(items)}" 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 == "" # ---- _parse_order_status：2026-08-16 真实详情页样本暴露出面包屑干扰的回归 ---- # 真实详情页除了配送进度条，还有一套面包屑

（class 同样含 CSS modules hash、 # label 也是「ショップ」且紧邻 title--2uGVi）。不锚定 li class 前缀时，findall # 会先抓走面包屑、再把带 -active-- 的进度条第 0 项错位跳过，导致永远判不出当前 # 阶段。修复后锚定 item--3gWCU 前缀，下面用脚本 probe_order_detail.py 落盘的真实 # 片段回归防倒退回退化。 # 摘自 .probe/order_detail/00-detail-default.html 的可复现片段： # 面包屑 li（应被排除）在前，真正的进度条第 0 项（active）夹在中间。 _REAL_STEPPER_HTML = ( '' '

' '

' '' '

' '
..
ショップ
' '
出荷
' '
配達店
' '
配達完了

' f'

注文番号：{_REAL_ORDER_ID}

' '' ) def test_parse_order_status_breadcrumb_sibling_does_not_shadow_active_item0(): """真实详情页有面包屑 li，不得让它替代进度条 item-0 挡在 -active-- 判定前""" snapshot = _parse_order_status(_REAL_STEPPER_HTML, _REAL_ORDER_ID) assert snapshot.found is True assert snapshot.stage_label == "ショップ" assert snapshot.order_state is None # ---- _parse_order_detail_status：2026-08-16 首次拿到详情页真实样本才有的路径 ---- # orderData.shippingList[].deliveryInfo.deliveryStatus 是站点结构化配送状态，比 # CSS 进度条稳。但映射遵循「只映射实测值」，当前只有 CHECKING_ORDER 一个样本。 def _detail_state(*, delivery_status: str | None, title: str | None, order_ok: bool = True) -> dict: """构造详情页 __INITIAL_STATE__ 最小 fixture（真实样本 .probe/order_detail）""" return { "pageType": "ph-detail", "requestParams": {"orderNumber": _REAL_ORDER_ID if order_ok else "other-order"}, "orderData": { "orderSummary": {"orderNumber": _REAL_ORDER_ID if order_ok else "other-order"}, "shippingList": [ { "deliveryInfo": ( {"deliveryStatus": delivery_status, "deliveryStatusTitle": title} if delivery_status else {} ) } ], }, } def test_parse_order_detail_status_uses_delivery_status_when_present(): """结构化的 deliveryStatus 优先于 stepper，stage_label 取站点原文""" snapshot = _parse_order_detail_status( _detail_state(delivery_status="CHECKING_ORDER", title="ご注文確認中"), _REAL_ORDER_ID ) assert snapshot is not None assert snapshot.found is True assert snapshot.delivery_status == "CHECKING_ORDER" assert snapshot.stage_label == "ご注文確認中" # 尚无实测映射的枚举值不猜 order_state（CHECKING_ORDER 对应「ショップ」阶段， # 已报告过 ORDERED/PAID，不需要 monitor 再报，与 stepper 行为一致） assert snapshot.order_state is None def test_parse_order_detail_status_unknown_code_falls_back_to_none(): """没实测过的 deliveryStatus 不接管——交回 stepper 兜底（避免掐掉 SHIPPED/DELIVERED 上报）实测只认识 CHECKING_ORDER；一旦真实订单走到「出荷/配達完了」，其 deliveryStatus 字符串还是未知，此时若结构化路径返回 order_state=None 会把 monitor 的状态上报掐掉。所以未识别的枚举值直接返回 None，由调用方降级到 stepper 的 _ORDER_STAGE_TO_STATE（后者早就能映射出荷/配達完了）。 """ assert ( _parse_order_detail_status( _detail_state(delivery_status="SOME_FUTURE_STATUS", title="出荷待ち"), _REAL_ORDER_ID, ) is None ) def test_parse_order_detail_status_returns_none_wrong_order(): """state 的 orderNumber 对不上所查订单时不得当 found（防 SPA 复用到别的订单）""" assert ( _parse_order_detail_status( _detail_state(delivery_status="CHECKING_ORDER", title="ご注文確認中", order_ok=False), _REAL_ORDER_ID, ) is None ) def test_parse_order_detail_status_returns_none_when_no_order_data(): assert _parse_order_detail_status(None, _REAL_ORDER_ID) is None assert ( _parse_order_detail_status(_detail_state(delivery_status=None, title=None), _REAL_ORDER_ID) is None ) # ---- 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="ログイン", ) 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="パスワードを入力", ) 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"{auth_site.RAKUTEN_LOGGED_OUT_MARKER}", ) 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, "ログイン")]), _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, "ログイン")]) 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, "ログイン")]) 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, "ご注文の反映に10分ほどかかります") ]) 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, "ログイン")]) 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, "改版了")]) 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("没有 state 的页面") 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 = "

お支払い金額 1,980円

" 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 = ( "

お支払い合計 12,345円

" "

ご注文番号 306087-20260813-0863947697

" "

お支払い期限 2026/08/20

" ) 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("

合計 1,980円

") # 标签不在候选列表里 def test_parse_checkout_summary_conflicting_amounts_raises(): html = "

お支払い金額 1,000円

お支払い合計 2,000円

" with pytest.raises(OrderOperationError): _parse_checkout_summary(html) def test_parse_checkout_summary_optional_fields_default_none(): summary = _parse_checkout_summary("

お支払い金額 500円

") 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 = ( '支払い金額' '

' '

297

' '

円

' "

" ) 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 = ( '

お支払い金額 297円

' '注文番号 306087-20260813-0863947697' ) 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 = "", 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="boom") 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") == "boom" 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 ----