content-find-agent-new/tests/test_dual_channel_normalization.py

"""V3-M1D: dual-channel canonical isomorphism + real dispatch."""

from __future__ import annotations

import pytest

from content_agent.errors import ContentAgentError, ErrorCode
from content_agent.integrations.douyin import CrawapiDouyinClient
from content_agent.integrations.kuaishou import (
    CrawapiKuaishouClient,
    _normalize_kuaishou_item,
)
from content_agent.integrations.shipinhao import (
    CrawapiShipinhaoClient,
    _normalize_shipinhao_item,
)
from content_agent.run_service import RunService

_QUERY = {"search_query_id": "q_001", "search_query": "彩虹", "discovery_start_source": "pattern_itemset"}


def _douyin_client():
    return CrawapiDouyinClient(
        base_url="http://crawapi.test",
        keyword_path="/k",
        blogger_path="/b",
        detail_path="/d",
        http_client=object(),
    )


def test_douyin_shipinhao_and_kuaishou_share_required_canonical_keys():
    douyin_item = _douyin_client()._normalize_content_item(
        _QUERY,
        {"aweme_id": "a1", "author": {"sec_uid": "u1", "nickname": "n"}, "video": {"play_addr": {"url_list": ["http://v"]}}},
        1,
        True,
        "12",
    )
    sph_item = _normalize_shipinhao_item(
        _QUERY,
        {"channel_content_id": "c1", "channel_account_id": "acc", "title": "彩虹 #彩虹", "video_url_list": [{"video_url": "http://v"}]},
        1,
        True,
        "12",
    )
    ks_item = _normalize_kuaishou_item(
        _QUERY,
        {
            "channel_content_id": "k1",
            "content_link": "https://www.kuaishou.com/short-video/k1",
            "channel_account_id": "acc",
            "channel_account_name": "快手作者",
            "title": "彩虹 #彩虹",
            "video_url_list": [{"video_url": "http://v"}],
        },
        1,
        False,
        "",
    )
    assert set(douyin_item) == set(sph_item)
    assert set(douyin_item).issubset(set(ks_item))
    assert douyin_item["platform"] == "douyin"
    assert sph_item["platform"] == "shipinhao"
    assert ks_item["platform"] == "kuaishou"
    assert ks_item["platform_content_url"].endswith("/k1")


def test_real_dispatch_builds_registered_platform_clients(monkeypatch):
    monkeypatch.setattr(
        CrawapiDouyinClient, "from_env", classmethod(lambda cls: object.__new__(cls))
    )
    monkeypatch.setattr(
        CrawapiKuaishouClient, "from_env", classmethod(lambda cls: object.__new__(cls))
    )
    monkeypatch.setattr(
        CrawapiShipinhaoClient, "from_env", classmethod(lambda cls: object.__new__(cls))
    )
    service = object.__new__(RunService)

    assert isinstance(service._platform_client("douyin", "real"), CrawapiDouyinClient)
    assert isinstance(service._platform_client("kuaishou", "real"), CrawapiKuaishouClient)
    assert isinstance(service._platform_client("shipinhao", "real"), CrawapiShipinhaoClient)


def test_unsupported_real_platform_raises():
    service = object.__new__(RunService)
    with pytest.raises(ContentAgentError) as exc:
        service._platform_client("bilibili", "real")
    assert exc.value.error_code == ErrorCode.INVALID_REQUEST