"""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 from app.trading.worker.models import LeaseTask from app.trading.worker.site_interact import ( CheckoutSummary, SiteInteractor, _extract_error_message, _extract_purchase_fields, _parse_initial_state, ) 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 == {} # ---- 未实现的方法仍然抛 NotImplementedError ---- @pytest.mark.parametrize( "method,args", [ ("enter_checkout", (_make_task(),)), ("submit_order", (_make_task(),)), ("pay", (_make_task(), "ord-1")), ("monitor", (_make_task(), "ord-1")), ], ) async def test_unimplemented_methods_raise(method: str, args: tuple): site = SiteInteractor(auth_session=None, settings=None) # type: ignore[arg-type] fn = getattr(site, method) with pytest.raises(NotImplementedError): await fn(*args) async def test_parse_checkout_unimplemented(): site = SiteInteractor(auth_session=None, settings=None) # type: ignore[arg-type] with pytest.raises(NotImplementedError): await site.parse_checkout("") # ---- add_to_cart / verify_cart 在没启动 Playwright 时应失败 ---- 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 ----