rakuten-api/tests/test_search_url.py

"""搜索/商品 URL 构建与解析的单元测试"""
from urllib.parse import parse_qsl, urlsplit

import pytest

from app.shared.errors import InvalidRequestError
from app.scraping.models.scrape import ItemCondition, SearchRequest, SortOption
from app.scraping.utils.urls import (
    build_item_url,
    build_search_url,
    item_url_parts,
    normalize_search_url,
    split_item_url,
)


def query_of(url: str) -> list[tuple[str, str]]:
    return parse_qsl(urlsplit(url).query, keep_blank_values=True)


def test_keyword_only_uses_default_sort_without_params():
    url = build_search_url(SearchRequest(keyword="nintendo switch"))
    assert url == "https://search.rakuten.co.jp/search/mall/nintendo%20switch/"


def test_genre_without_keyword_uses_placeholder_segment():
    url = build_search_url(SearchRequest(genre_id="565950"))
    assert urlsplit(url).path == "/search/mall/-/565950/"


def test_keyword_and_genre_are_both_path_segments():
    url = build_search_url(SearchRequest(keyword="switch", genre_id="565950"))
    assert urlsplit(url).path == "/search/mall/switch/565950/"


def test_page_one_is_omitted_but_later_pages_are_explicit():
    assert "p=" not in build_search_url(SearchRequest(keyword="a", page=1))
    assert ("p", "3") in query_of(build_search_url(SearchRequest(keyword="a", page=3)))


@pytest.mark.parametrize(
    ("sort", "expected"),
    [
        (SortOption.STANDARD, None),
        (SortOption.PRICE_ASC, "2"),
        (SortOption.PRICE_DESC, "3"),
        (SortOption.NEWEST, "4"),
        (SortOption.REVIEW_COUNT, "5"),
        (SortOption.REVIEW_SCORE, "6"),
        (SortOption.PRICE_WITH_SHIPPING_ASC, "11"),
        (SortOption.PRICE_WITH_SHIPPING_DESC, "12"),
    ],
)
def test_sort_maps_to_site_codes(sort, expected):
    """排序码取自站点前端 bundle 的枚举，改动会直接影响返回顺序"""
    query = dict(query_of(build_search_url(SearchRequest(keyword="a", sort=sort))))
    assert query.get("s") == expected


def test_condition_and_bool_filters_share_repeatable_f_param():
    url = build_search_url(
        SearchRequest(
            keyword="a",
            condition=ItemCondition.USED,
            free_shipping=True,
            include_sold_out=True,
            has_review=True,
            next_day_delivery=True,
            super_deal=True,
        )
    )
    f_values = sorted(value for key, value in query_of(url) if key == "f")
    assert f_values == sorted(["100", "0", "2", "4", "12", "13"])


def test_advanced_filters_are_mapped_to_site_param_names():
    url = build_search_url(
        SearchRequest(
            keyword="a",
            min_price=1000,
            max_price=5000,
            shop_id=272415,
            exclude_keyword="中古",
            title_only=True,
            or_query=True,
            min_review_score=4,
            tags=["1001", "1002"],
        )
    )
    query = dict(query_of(url))
    assert query["min"] == "1000"
    assert query["max"] == "5000"
    assert query["sid"] == "272415"
    assert query["nitem"] == "中古"
    assert query["sf"] == "1"
    assert query["st"] == "O"
    assert query["review"] == "4"
    assert query["tg"] == "1001,1002"


def test_search_request_requires_a_target():
    with pytest.raises(ValueError):
        SearchRequest()


def test_search_request_rejects_inverted_price_range():
    with pytest.raises(ValueError):
        SearchRequest(keyword="a", min_price=5000, max_price=1000)


def test_normalize_search_url_keeps_url_untouched_on_first_page():
    raw = "https://search.rakuten.co.jp/search/mall/switch/?s=2&f=2"
    assert normalize_search_url(raw, 1) == raw


def test_normalize_search_url_overrides_existing_page():
    raw = "https://search.rakuten.co.jp/search/mall/switch/?p=9&s=2"
    query = dict(query_of(normalize_search_url(raw, 4)))
    assert query["p"] == "4"
    assert query["s"] == "2"


def test_normalize_search_url_rejects_foreign_host():
    with pytest.raises(InvalidRequestError):
        normalize_search_url("https://example.com/search/mall/switch/", 1)


def test_build_item_url_joins_shop_and_item_code():
    assert build_item_url("edion", "4902370549263") == "https://item.rakuten.co.jp/edion/4902370549263/"


def test_split_item_url_extracts_shop_and_item_code():
    assert split_item_url("https://item.rakuten.co.jp/edion/4902370549263/?variantId=x") == (
        "edion",
        "4902370549263",
    )


def test_split_item_url_rejects_foreign_host():
    with pytest.raises(InvalidRequestError):
        split_item_url("https://www.rakuten.co.jp/edion/4902370549263/")


def test_item_url_parts_returns_empty_for_ad_redirect_links():
    """广告位链接指向跳转域，解析不出来应静默返回空串而不是抛异常"""
    assert item_url_parts("https://grp07.ias.rakuten.co.jp/redirect_rpp/?s=abc") == ("", "")