"""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 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,
OrderStatusSnapshot,
SiteInteractor,
_extract_error_message,
_extract_purchase_fields,
_parse_checkout_summary,
_parse_initial_state,
_parse_order_status,
)
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)
# ---- 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 = (
'支払い金額'
'"
)
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())
# ---- helper ----