"""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, OrderOperationError 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 ( "" ) # ---- _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によりすべて暗号化されます
© Rakuten Group, Inc.
""" 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}
    " 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) # ---- _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) # ---- 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 ----