cmautobuy/client/test/test_pdd_collect_service.py

"""PDD 商品采集与 XML 解析测试，不连接真机。"""

from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
import unittest
import xml.etree.ElementTree as ET

from src.pdd_collect_service import (
    PddCollectError,
    PddCollectService,
    parse_goods_page,
    parse_quantity,
    parse_spec_panel,
)
from src.pdd_device_service import PddDeviceService


FIXTURES = Path(__file__).parent / "fixtures"


@dataclass(frozen=True)
class FakeTask:
    goods_url: str
    goods_id: str | None = None


class FakeCollectDevice:
    def __init__(self, home_xml: str, spec_xml: str):
        self.home_xml = home_xml
        self.spec_xml = spec_xml
        self.panel_open = False
        self.clicks = []
        self.swipes = []

    def app_current(self):
        return {"package": "com.xunmeng.pinduoduo"}

    def app_start(self, _package):
        raise AssertionError("PDD 已经在前台，不应重复启动")

    def app_wait(self, _package, timeout=10):
        return 123

    def open_url(self, url):
        self.opened_url = url

    def dump_hierarchy(self):
        return self.spec_xml if self.panel_open else self.home_xml

    def click(self, x, y):
        self.clicks.append((x, y))
        if y > 1600:
            self.panel_open = True

    def swipe(self, *args, **kwargs):
        self.swipes.append((args, kwargs))


class LoadingDevice(FakeCollectDevice):
    def dump_hierarchy(self):
        return '<hierarchy><node text="正在加载" /></hierarchy>'


def keep_only_one_sku(xml_data: str) -> str:
    """从脱敏固件中删除蓝色和 L，只保留一个组合。"""

    root = ET.fromstring(xml_data)
    for parent in root.iter():
        for child in list(parent):
            own_label = (
                child.get("text") or child.get("content-desc") or ""
            ).strip()
            labels = " ".join(
                (node.get("text") or node.get("content-desc") or "").strip()
                for node in child.iter("node")
            )
            if own_label == "蓝色" or labels.strip() == "L":
                parent.remove(child)
    return ET.tostring(root, encoding="unicode")


class PddCollectParserTest(unittest.TestCase):
    @classmethod
    def setUpClass(cls):
        cls.home_xml = (FIXTURES / "pdd_goods_page.xml").read_text(encoding="utf-8")
        cls.spec_xml = (FIXTURES / "pdd_spec_panel.xml").read_text(encoding="utf-8")

    def test_quantity_keeps_raw_value_and_approximate_flag(self):
        result = parse_quantity("已拼1.2万+件")
        self.assertEqual(result.value, 12000)
        self.assertEqual(result.raw, "已拼1.2万+件")
        self.assertTrue(result.approximate)

    def test_parse_goods_page(self):
        result = parse_goods_page(self.home_xml)
        self.assertEqual(result.title, "测试纯棉短袖商品")
        self.assertEqual(result.shop_name, "测试服饰旗舰店")
        self.assertEqual(result.sales.value, 12000)
        self.assertTrue(result.sales.approximate)
        self.assertEqual(result.reviews.value, 2356)

    def test_parse_spec_panel_uses_generic_dimensions_and_cents(self):
        result = parse_spec_panel(self.spec_xml)
        self.assertEqual(result.price_cent, 1000)
        self.assertEqual([item.key for item in result.dimensions], ["color", "size"])
        self.assertEqual(
            [value.text for value in result.dimensions[0].values],
            ["红色", "蓝色"],
        )
        self.assertFalse(result.dimensions[0].values[1].available)
        self.assertEqual(
            [value.text for value in result.dimensions[1].values],
            ["M", "L"],
        )

    def test_login_and_invalid_xml_have_different_error_codes(self):
        with self.assertRaises(PddCollectError) as login:
            parse_goods_page('<hierarchy><node text="手机号登录" /></hierarchy>')
        with self.assertRaises(PddCollectError) as invalid:
            parse_goods_page("<hierarchy>")
        self.assertEqual(login.exception.code, "PDD_PAGE_LOGIN_REQUIRED")
        self.assertEqual(invalid.exception.code, "PDD_DATA_XML_INVALID")

    def test_captcha_and_incomplete_spec_have_specific_codes(self):
        with self.assertRaises(PddCollectError) as captcha:
            parse_goods_page('<hierarchy><node text="请完成验证" /></hierarchy>')
        with self.assertRaises(PddCollectError) as incomplete:
            parse_spec_panel('<hierarchy><node text="普通页面" /></hierarchy>')
        self.assertEqual(captcha.exception.code, "PDD_PAGE_CAPTCHA")
        self.assertEqual(incomplete.exception.code, "PDD_DATA_SPEC_INCOMPLETE")

    def test_collect_returns_versioned_pdd_data_without_real_device(self):
        # 收窄到一个可用组合，便于验证完整流程而不模拟真实页面切换。
        one_sku_xml = keep_only_one_sku(self.spec_xml)
        device = FakeCollectDevice(self.home_xml, one_sku_xml)
        service = PddCollectService(
            PddDeviceService(lambda _serial: device),
            "USB-001",
            "client-001",
            sleeper=lambda _seconds: None,
            now=lambda: datetime(2026, 8, 7, 8, 0, tzinfo=timezone.utc),
            max_page_swipes=1,
            max_spec_swipes=1,
        )

        result = service.collect(
            FakeTask("https://mobile.yangkeduo.com/goods.html?goods_id=123")
        )
        data = result.to_pdd_data()

        self.assertEqual(data["schema_version"], 1)
        self.assertEqual(data["goods"]["goods_id"], "123")
        self.assertEqual(data["metrics"]["sales"]["value"], 12000)
        self.assertEqual(data["dimensions"][0]["key"], "color")
        self.assertEqual(data["skus"][0]["options"], {"color": "红色", "size": "M"})
        self.assertEqual(data["skus"][0]["price_cent"], 1000)
        self.assertEqual(data["source"]["device_address"], "USB-001")
        self.assertIsNone(data["purchase"])

    def test_missing_goods_id_is_structured_error(self):
        service = PddCollectService(
            PddDeviceService(lambda _serial: object()),
            "USB-001",
            "client-001",
        )
        with self.assertRaises(PddCollectError) as raised:
            service.collect(FakeTask("https://example.com/goods.html"))
        self.assertEqual(raised.exception.code, "PDD_DATA_GOODS_ID_MISSING")

    def test_page_timeout_has_specific_error_code(self):
        device = LoadingDevice(self.home_xml, self.spec_xml)
        ticks = iter((0.0, 2.0))
        service = PddCollectService(
            PddDeviceService(lambda _serial: device),
            "USB-001",
            "client-001",
            sleeper=lambda _seconds: None,
            monotonic=lambda: next(ticks),
            page_timeout=1.0,
        )

        with self.assertRaises(PddCollectError) as raised:
            service.collect(
                FakeTask("https://mobile.yangkeduo.com/goods.html?goods_id=123")
            )
        self.assertEqual(raised.exception.code, "PDD_PAGE_TIMEOUT")

    def test_incomplete_goods_data_does_not_return_partial_success(self):
        incomplete_home = self.home_xml.replace(
            '<node class="android.widget.TextView" text="店铺：测试服饰旗舰店" '
            'bounds="[30,1050][600,1110]" visible-to-user="true" '
            'enabled="true" />',
            "",
        )
        device = FakeCollectDevice(incomplete_home, self.spec_xml)
        service = PddCollectService(
            PddDeviceService(lambda _serial: device),
            "USB-001",
            "client-001",
            sleeper=lambda _seconds: None,
            max_page_swipes=0,
        )

        with self.assertRaises(PddCollectError) as raised:
            service.collect(
                FakeTask("https://mobile.yangkeduo.com/goods.html?goods_id=123")
            )
        self.assertEqual(raised.exception.code, "PDD_DATA_SHOP_MISSING")


if __name__ == "__main__":
    unittest.main()