"""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 的契约。 clear_cart 与 _dump_debug_snapshot 用不依赖 Playwright 的 fake page 覆盖了方法自身逻辑（见文件尾部对应小节），真实站点行为仍由 scripts/ 真账号脚本验证。 """ from __future__ import annotations import asyncio import json from pathlib import Path from typing import Any import pytest from datetime import datetime, timezone from app.shared.errors import ( BrowserDeadError, 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, _DELETE_BUTTON_SELECTOR, _extract_error_message, _extract_purchase_fields, _normalize_intent_items, _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 ( "" ) # ---- 下单意图兼容 ---- def test_normalize_intent_items_accepts_legacy_single_item(): assert _normalize_intent_items({ "item_url": "https://item.rakuten.co.jp/shop/x/", "quantity": 2, }) == [{ "item_url": "https://item.rakuten.co.jp/shop/x/", "quantity": 2, "variant_id": None, "choice": None, }] def test_normalize_intent_items_accepts_multiple_items_and_string_urls(): assert _normalize_intent_items({ "items": [ {"item_url": "https://item.rakuten.co.jp/shop/x/", "quantity": 2}, "https://item.rakuten.co.jp/shop/y/", ] }) == [ { "item_url": "https://item.rakuten.co.jp/shop/x/", "quantity": 2, "variant_id": None, "choice": None, }, { "item_url": "https://item.rakuten.co.jp/shop/y/", "quantity": 1, "variant_id": None, "choice": None, }, ] def test_normalize_intent_items_rejects_empty_items(): with pytest.raises(InvalidRequestError): _normalize_intent_items({"items": []}) async def test_add_to_cart_processes_all_items_and_keeps_legacy_state(): site = SiteInteractor.__new__(SiteInteractor) site._lock = asyncio.Lock() site._per_task_state = {} calls: list[dict[str, Any]] = [] async def fake_add(**kwargs): calls.append(kwargs) index = len(calls) return { "item_id": str(index), "shop_bid": "shop", "basket_domain": "https://basket", "response_html": f"html-{index}", "screenshot": b"", } site._add_to_cart_with_fields = fake_add snapshot = await site.add_to_cart(_make_task(intent={ "items": [ {"item_url": "https://item.rakuten.co.jp/shop/x/"}, {"item_url": "https://item.rakuten.co.jp/shop/y/", "quantity": 3}, ] })) assert [call["item_url"] for call in calls] == [ "https://item.rakuten.co.jp/shop/x/", "https://item.rakuten.co.jp/shop/y/", ] assert calls[1]["quantity"] == 3 assert site._per_task_state["t1"]["item_id"] == "1" assert site._per_task_state["t1"]["item_ids"] == ["1", "2"] assert snapshot.html == "html-2" # ---- _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_required_options_skips_placeholder_value(): """回归：必填 select 的 values[0] 是「選択してください」占位项，不能当答案提交真实商品页（.probe/checkout/03-add-fail-2.html 等 3 份样本）里必填 select 的第一个候选恒为 id=0 的占位项，真实取值从 id=200 起。旧实现直接取 values[0]，等于把「請選擇」填进 choice 提交上去。 """ state = { "item": {"itemId": 1, "variantId": "v"}, "purchase": { "sku": {"inventoryType": "single"}, "sellType": {"normalPurchase": {"basketDomain": "https://x/add", "purchaseCondition": "enabled"}}, "information": { "options": [ { "id": 100, "name": "この商品は「トライタン製／プラスチック」です", "type": "select", "isRequired": True, "values": [ {"id": 0, "name": "選択してください"}, {"id": 200, "name": "確認した"}, ], } ] }, }, "shop": {"information": {"shopId": 1}}, } fields = _extract_purchase_fields(state, intent_override={}) assert fields["has_required_options"] is True assert fields["form_fields"]["choice"] == "この商品は「トライタン製／プラスチック」です:確認した" assert fields["unfillable_required_options"] == [] def test_extract_fields_required_free_text_option_cannot_be_auto_filled(): """必填自由文本项（type=text，无候选值）自动填不了，要点名报出来""" state = { "item": {"itemId": 1, "variantId": "v"}, "purchase": { "sku": {"inventoryType": "single"}, "sellType": {"normalPurchase": {"basketDomain": "https://x/add", "purchaseCondition": "enabled"}}, "information": { "options": [ {"id": 101, "name": "【お名前】", "type": "text", "isRequired": True}, ] }, }, "shop": {"information": {"shopId": 1}}, } fields = _extract_purchase_fields(state, intent_override={}) assert fields["has_required_options"] is True # 填不出来就不填，也不拿占位/空值凑数 assert "choice" not in fields["form_fields"] assert fields["unfillable_required_options"] == ["【お名前】"] def test_extract_fields_skips_optional_options_when_auto_filling(): """非必填项不替上游做业务决定（如「置き配を希望する」），只自动填必填项""" state = { "item": {"itemId": 1, "variantId": "v"}, "purchase": { "sku": {"inventoryType": "single"}, "sellType": {"normalPurchase": {"basketDomain": "https://x/add", "purchaseCondition": "enabled"}}, "information": { "options": [ { "id": 100, "name": "必須確認", "type": "select", "isRequired": True, "values": [{"id": 0, "name": "選択してください"}, {"id": 200, "name": "了解"}], }, { "id": 101, "name": "「置き配」希望について", "type": "select", "isRequired": False, "values": [{"id": 200, "name": "置き配を希望しない"}], }, ] }, }, "shop": {"information": {"shopId": 1}}, } fields = _extract_purchase_fields(state, intent_override={}) assert fields["form_fields"]["choice"] == "必須確認:了解" 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 screenshot(self, *, full_page: bool = True) -> bytes: return b"fake-png" 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" class _FakeBrowser: """Browser 替身：只实现 _ensure_browser_alive 用到的 is_connected() 没有它，`_ensure_context_ready` 会判定「浏览器没起来」并尝试真的 launch 一次 Chromium。connected 可以翻成 False 来模拟掉线。 """ def __init__(self, connected: bool = True): self.connected = connected def is_connected(self) -> bool: return self.connected def _build_site(tmp_path: Path, context: _FakeContext, auth: _FakeAuthSession) -> SiteInteractor: """装配一个不依赖 Playwright 的 SiteInteractor auth_state_dir 指向空目录：_ensure_context_ready 见不到 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._browser = _FakeBrowser() # type: ignore[assignment] 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(): """浏览器活着但没有留存确认页 → OrderOperationError（不是掉线那类错误）""" site = SiteInteractor(auth_session=None, settings=None) # type: ignore[arg-type] site._browser = _FakeBrowser() # type: ignore[assignment] 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] site._browser = _FakeBrowser() # type: ignore[assignment] 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-*")) == [] # ---- _query_cart_count：空车（status=101）与获取失败（其他 status）必须区分 ---- # # 响应 body 均为 2026-08-16 真账号实测原样（scripts/probe_cart_count.py + # .probe/cart_count/）：空车是合法响应，当失败抛错会让 clear_cart 末尾校验拿 -1、 # 开单前清理被 runner 误判为「未清空」。 class _FakeCartCountRequest: """APIRequestContext 替身：回固定 body，记录入参""" def __init__(self, body: str): self._body = body self.calls: list[tuple[str, dict[str, str]]] = [] async def get(self, url: str, headers: dict[str, str] | None = None) -> Any: self.calls.append((url, headers or {})) class _Resp: def __init__(self, text: str): self._text = text async def text(self) -> str: return self._text return _Resp(self._body) class _FakeCartCountContext: def __init__(self, body: str): self.request = _FakeCartCountRequest(body) def _build_cart_count_site(body: str) -> SiteInteractor: site = SiteInteractor(auth_session=None, settings=None) # type: ignore[arg-type] site._context = _FakeCartCountContext(body) # type: ignore[assignment] return site async def test_query_cart_count_non_empty_returns_count(): site = _build_cart_count_site( 'callBack({"status":"100","message":"success.","count":"3"})' ) raw_status, count = await site._query_cart_count() assert (raw_status, count) == ("100", 3) async def test_query_cart_count_empty_cart_is_not_a_failure(): """status=101 + count="" 是空车（合法响应），返回 ("101", 0) 而不是抛错""" site = _build_cart_count_site( 'callBack({"status":"101","message":"value not found.","count":""})' ) raw_status, count = await site._query_cart_count() assert (raw_status, count) == ("101", 0) async def test_query_cart_count_rejected_request_raises(): """status=300（referer/sid 被站点拒绝）才是「获取失败」""" site = _build_cart_count_site( 'callBack({"status":"300","message":"referer is empty value.","count":""})' ) with pytest.raises(CartOperationError): await site._query_cart_count() async def test_query_cart_count_unparseable_body_raises(): site = _build_cart_count_site("not jsonp") with pytest.raises(CartOperationError): await site._query_cart_count() # ---- clear_cart：空车 / 有商品两个场景的核心逻辑（fake page，不依赖 Playwright）---- # # 真实站点行为（SPA 是否点一个少一个、确认 modal 是否存在）由真账号脚本验证 # （scripts/verify_cart_empty.py、scripts/verify_cart_clear.py）；这里覆盖的是 # 方法自身的逻辑：空车立即结束、有商品时循环点「削除」直到按钮消失、最终页面 # HTML 进返回值、HTML 抓取失败不掩盖清理结果。 class _FakeDeleteButton: """「削除」按钮替身：还有剩余按钮时 wait_for 成功，click 消耗一个（模拟 SPA 点完一个就把该项从 DOM 摘掉）""" def __init__(self, page: "_FakeClearCartPage"): self._page = page async def wait_for(self, *, state: str, timeout: int) -> None: if self._page.delete_buttons_left <= 0: raise RuntimeError("模拟 Playwright 等待超时：没有删除按钮了") async def click(self) -> None: self._page.delete_buttons_left -= 1 class _FakeMissingButton: """确认 modal 按钮替身：永远不出现（站点是否弹 modal 未实测，按不存在处理）""" async def wait_for(self, *, state: str, timeout: int) -> None: raise RuntimeError("模拟 Playwright 等待超时：无确认 modal") async def click(self) -> None: raise AssertionError("没有可见的确认按钮，不应触发点击") class _FakeLocator: def __init__(self, button: Any): self.first = button class _FakeClearCartPage: """cart 页替身：实现 clear_cart 用到的全部 page 接口""" def __init__(self, *, delete_buttons: int, html: str): self.delete_buttons_left = delete_buttons self._html = html self.closed = False def locator(self, selector: str) -> _FakeLocator: if selector == _DELETE_BUTTON_SELECTOR: return _FakeLocator(_FakeDeleteButton(self)) return _FakeLocator(_FakeMissingButton()) async def goto(self, url: str, *, wait_until: str, timeout: int) -> None: return None async def wait_for_function(self, *args: Any, **kwargs: Any) -> None: return None async def wait_for_load_state(self, *args: Any, **kwargs: Any) -> None: return None async def wait_for_timeout(self, *args: Any, **kwargs: Any) -> None: return None async def content(self) -> str: return self._html async def screenshot(self, *, full_page: bool = True) -> bytes: return b"fake-png" async def close(self) -> None: self.closed = True class _FakeClearCartContext: """BrowserContext 替身：new_page 返回 fake cart 页，request 回固定 count body""" def __init__(self, *, page: _FakeClearCartPage, count_body: str): self.page = page self.request = _FakeCartCountRequest(count_body) async def new_page(self) -> _FakeClearCartPage: return self.page class _FakeLoggedInAuth: """AuthSession 替身：require_logged_in 直接放行""" async def require_logged_in(self, site: str) -> None: return None def _build_clear_cart_site(*, delete_buttons: int, count_body: str, html: str) -> SiteInteractor: """构造 clear_cart 可离线跑起来的 SiteInteractor（fake page + fake count API）""" site = SiteInteractor(auth_session=_FakeLoggedInAuth(), settings=None) # type: ignore[arg-type] site._browser = _FakeBrowser() # type: ignore[assignment] site._context = _FakeClearCartContext( page=_FakeClearCartPage(delete_buttons=delete_buttons, html=html), count_body=count_body, ) # type: ignore[assignment] # settings=None，走不了真实的 mtime 检查；探活那半段由 _FakeBrowser 覆盖 async def _noop_refresh() -> None: return None site._ensure_context_ready = _noop_refresh # type: ignore[assignment] return site _EMPTY_COUNT_BODY = 'callBack({"status":"101","message":"value not found.","count":""})' async def test_clear_cart_empty_cart_returns_success_without_clicks(): """场景 1：购物车没有商品 —— 找不到「削除」按钮立即结束，按清理成功返回空车时 count API 返回 status=101（合法空车），cart_count 必须是 0 而不是 -1（2026-08-16 修复过的误判）；removed_count=0，最终页面 HTML 与截图照常带回。 """ site = _build_clear_cart_site( delete_buttons=0, count_body=_EMPTY_COUNT_BODY, html="empty cart", ) result = await site.clear_cart() assert result == { "removed_count": 0, "cart_count": 0, "html": "empty cart", "screenshot": b"fake-png", } assert site._context.page.closed is True # 页面必须关，不能泄漏 async def test_clear_cart_with_items_clicks_until_no_button_left(): """场景 2：购物车有商品 —— 循环点第一个「削除」，点一个少一个，直到按钮消失每次循环都重新查 selector（避免索引漂移），3 件商品应点 3 次；清空后 count API 回 status=101 → cart_count=0，最终页面 HTML 与截图进返回值。 """ site = _build_clear_cart_site( delete_buttons=3, count_body=_EMPTY_COUNT_BODY, html="cleared cart", ) result = await site.clear_cart() assert result["removed_count"] == 3 assert result["cart_count"] == 0 assert result["html"] == "cleared cart" assert result["screenshot"] == b"fake-png" assert site._context.page.closed is True async def test_clear_cart_html_capture_failure_keeps_clear_result(): """最终页面 HTML / 截图抓取失败（如页面异常）不掩盖清理结果本身：两者记空""" site = _build_clear_cart_site( delete_buttons=0, count_body=_EMPTY_COUNT_BODY, html="", ) async def _broken_content() -> str: raise RuntimeError("page already closed") async def _broken_screenshot(*, full_page: bool = True) -> bytes: raise RuntimeError("page already closed") site._context.page.content = _broken_content # type: ignore[assignment] site._context.page.screenshot = _broken_screenshot # type: ignore[assignment] result = await site.clear_cart() assert result == {"removed_count": 0, "cart_count": 0, "html": "", "screenshot": b""} # ---- helper ---- # ---- 浏览器掉线：任务边界就地重建，中途掉线转 BrowserDeadError ---- # # 真实的 Chromium 崩溃（容器 /dev/shm 不足、OOM kill、seccomp 挡 sandbox）没法在 # 离线测试里制造，这里用 _FakeBrowser.connected 翻成 False 模拟「is_connected() # 返回 False」这个唯一的可观测信号，覆盖的是 SiteInteractor 对该信号的反应。 def _build_dead_browser_site(tmp_path: Path) -> SiteInteractor: """浏览器已掉线的 SiteInteractor：_browser 在位但 is_connected() 为 False""" from app.shared.config import Settings site = SiteInteractor( auth_session=_FakeLoggedInAuth(), # type: ignore[arg-type] settings=Settings(auth_state_dir=str(tmp_path / "auth"), evidence_dir=str(tmp_path)), ) site._browser = _FakeBrowser(connected=False) # type: ignore[assignment] return site def test_browser_alive_reflects_is_connected(tmp_path): """browser_alive：没起过 / 已掉线 / 正常，三种情况分别为 False/False/True""" site = SiteInteractor(auth_session=None, settings=None) # type: ignore[arg-type] assert site.browser_alive is False # 还没 start() site._browser = _FakeBrowser(connected=False) # type: ignore[assignment] assert site.browser_alive is False site._browser = _FakeBrowser(connected=True) # type: ignore[assignment] assert site.browser_alive is True def test_browser_alive_survives_is_connected_raising(): """is_connected() 自己抛错（驱动已经没了）也要当掉线处理，不能把异常漏出去""" class _ExplodingBrowser: def is_connected(self) -> bool: raise RuntimeError("driver connection closed") site = SiteInteractor(auth_session=None, settings=None) # type: ignore[arg-type] site._browser = _ExplodingBrowser() # type: ignore[assignment] assert site.browser_alive is False def test_browser_status_distinguishes_not_started_from_dead(): """/health 要能分清「还没启动」和「起过但死了」——只有后者算 degraded""" site = SiteInteractor(auth_session=None, settings=None) # type: ignore[arg-type] not_started = site.browser_status() assert not_started["started"] is False assert not_started["alive"] is False site._browser = _FakeBrowser(connected=False) # type: ignore[assignment] dead = site.browser_status() assert dead["started"] is True assert dead["alive"] is False site._browser = _FakeBrowser(connected=True) # type: ignore[assignment] assert site.browser_status() == { "started": True, "alive": True, "pending_checkout_tasks": [], "detail": "连接正常", } async def test_ensure_browser_alive_relaunches_on_task_boundary(tmp_path): """任务边界探到掉线 → 重放一遍启动流程，调用方无感知""" site = _build_dead_browser_site(tmp_path) relaunched = [] async def _fake_launch() -> None: relaunched.append(True) site._browser = _FakeBrowser(connected=True) # type: ignore[assignment] site._launch = _fake_launch # type: ignore[assignment] await site._ensure_browser_alive() assert relaunched == [True] assert site.browser_alive is True async def test_ensure_browser_alive_is_noop_when_connected(tmp_path): """浏览器好好的就不能重建——重建一次要几秒，还会丢掉当前 context 的 cookie""" site = _build_dead_browser_site(tmp_path) site._browser = _FakeBrowser(connected=True) # type: ignore[assignment] async def _fail_launch() -> None: raise AssertionError("浏览器连接正常时不应重建") site._launch = _fail_launch # type: ignore[assignment] await site._ensure_browser_alive() async def test_ensure_browser_alive_drops_orphaned_checkout_pages(tmp_path): """重建前丢掉残留的确认页：那些 Page 已随浏览器一起没了，留着必然报错""" site = _build_dead_browser_site(tmp_path) site._checkout_pages["t-stuck"] = object() # type: ignore[assignment] async def _fake_launch() -> None: site._browser = _FakeBrowser(connected=True) # type: ignore[assignment] site._launch = _fake_launch # type: ignore[assignment] await site._ensure_browser_alive() assert site._checkout_pages == {} async def test_ensure_browser_alive_raises_browser_dead_when_relaunch_fails(tmp_path): """重建也起不来 → BrowserDeadError（AppError，runner 接得住），并把半成品收干净""" site = _build_dead_browser_site(tmp_path) async def _broken_launch() -> None: raise RuntimeError("chromium 起不来") site._launch = _broken_launch # type: ignore[assignment] with pytest.raises(BrowserDeadError): await site._ensure_browser_alive() assert site._browser is None # 已 teardown，下次调用从头再试 async def test_new_page_wraps_target_closed_into_app_error(): """中途掉线的裸 TargetClosedError 必须被包成 AppError 这是整条链最关键的一环：不包的话异常会穿过 runner 的 except AppError，落到主循环那个只记日志的兜底里——任务一次都不上报，网关要干等租约过期。 """ class _DeadContext: async def new_page(self): raise RuntimeError("Target page, context or browser has been closed") site = SiteInteractor(auth_session=None, settings=None) # type: ignore[arg-type] site._context = _DeadContext() # type: ignore[assignment] with pytest.raises(BrowserDeadError) as excinfo: await site._new_page() from app.shared.errors import AppError assert isinstance(excinfo.value, AppError) assert excinfo.value.err_code == 5006 async def test_request_keeps_original_error_when_browser_is_alive(): """浏览器活着时的请求失败是正常业务失败，不能被误标成掉线""" class _FlakyRequest: async def get(self, url: str, **kwargs): raise RuntimeError("站点 502") class _Context: request = _FlakyRequest() site = SiteInteractor(auth_session=None, settings=None) # type: ignore[arg-type] site._browser = _FakeBrowser(connected=True) # type: ignore[assignment] site._context = _Context() # type: ignore[assignment] with pytest.raises(RuntimeError, match="站点 502"): await site._request("get", "https://example.test/") async def test_request_converts_to_browser_dead_when_browser_is_gone(): """同样的请求失败，浏览器确实没了时才改写成 BrowserDeadError""" class _FlakyRequest: async def get(self, url: str, **kwargs): raise RuntimeError("Target closed") class _Context: request = _FlakyRequest() site = SiteInteractor(auth_session=None, settings=None) # type: ignore[arg-type] site._browser = _FakeBrowser(connected=False) # type: ignore[assignment] site._context = _Context() # type: ignore[assignment] with pytest.raises(BrowserDeadError): await site._request("get", "https://example.test/") async def test_submit_order_reports_browser_dead_before_blaming_selectors(tmp_path): """浏览器死时 submit_order 要报掉线，而不是「找不到确认按钮」留存的 Page 是个死对象（不是 None），不先探活的话 selector 轮询会全部超时，最后把「浏览器崩了」误诊成「站点改版了」——这两个结论的处置方式完全不同。 """ site = _build_dead_browser_site(tmp_path) site._checkout_pages["t1"] = object() # type: ignore[assignment] with pytest.raises(BrowserDeadError): await site.submit_order(_make_task(task_id="t1")) async def test_pay_reports_browser_dead(tmp_path): """pay 之前 submit_order 已经真的下过单了，掉线必须当场抛出去交人工""" site = _build_dead_browser_site(tmp_path) site._checkout_pages["t1"] = object() # type: ignore[assignment] with pytest.raises(BrowserDeadError): await site.pay(_make_task(task_id="t1"), "ord-1")