Create-knowledge-find-decode/tests/test_platform_adapters.py

from __future__ import annotations

from types import SimpleNamespace

from acquisition.platforms.douyin import DouyinAdapter
from acquisition.platforms import douyin as douyin_module
from acquisition.platforms import weixin as weixin_module
from acquisition.platforms import xiaohongshu as xhs_module
from acquisition.platforms.weixin import WeixinAdapter
from acquisition.platforms.xiaohongshu import XiaohongshuAdapter
from core.models import Post


def test_xiaohongshu_adapter_maps_search_and_detail(monkeypatch):
    settings = SimpleNamespace(search_content_type="图文")

    def fake_search(query, *, content_type, limit, settings, rate_limiter):
        assert query == "脚本 开头"
        assert content_type == "图文"
        assert limit == 3
        return [
            {
                "id": "xhs-1",
                "url": "https://xhs.test/1",
                "title": "脚本开头",
                "nick_name": "作者 A",
                "cover_url": "https://img.test/cover.jpg",
            }
        ]

    def fake_fetch(source_id, *, settings, rate_limiter):
        assert source_id == "xhs-1"
        return Post(
            id="xhs_xhs-1",
            platform="xiaohongshu",
            url="https://xhs.test/detail/1",
            content_id="xhs-1",
            title="详情标题",
            content_type="图文",
            body_text="详情正文",
            image_urls=["https://img.test/1.jpg"],
            author_name="详情作者",
            raw={"source": "mock"},
        )

    monkeypatch.setattr(xhs_module, "search_xiaohongshu", fake_search)
    monkeypatch.setattr(xhs_module, "fetch_post_detail", fake_fetch)

    adapter = XiaohongshuAdapter()
    candidate = adapter.search("脚本 开头", settings=settings, limit=3, rate_limiter=None)[0]
    item = adapter.fetch_detail(candidate, settings=settings, rate_limiter=None)

    assert candidate.source_id == "xhs-1"
    assert candidate.author == "作者 A"
    assert item.title == "详情标题"
    assert item.image_urls == ["https://img.test/1.jpg"]
    assert item.raw["source"] == "mock"


def test_weixin_adapter_hashes_url_and_reads_detail(monkeypatch):
    settings = SimpleNamespace()

    def fake_search(query, *, limit, settings, rate_limiter):
        assert query == "公众号 选题"
        assert limit == 2
        return [
            {
                "url": "https://mp.weixin.qq.com/s/abc",
                "title": "公众号选题",
                "nick_name": "公众号",
                "cover_url": "https://img.test/wx.jpg",
            }
        ]

    def fake_detail(url, *, settings, rate_limiter):
        assert url == "https://mp.weixin.qq.com/s/abc"
        return "公众号正文", ["https://img.test/wx-1.jpg"]

    monkeypatch.setattr(weixin_module, "search_weixin", fake_search)
    monkeypatch.setattr(weixin_module, "fetch_weixin_detail", fake_detail)

    adapter = WeixinAdapter()
    candidate = adapter.search("公众号 选题", settings=settings, limit=2, rate_limiter=None)[0]
    item = adapter.fetch_detail(candidate, settings=settings, rate_limiter=None)

    assert candidate.platform == "weixin"
    assert len(candidate.source_id) == 16
    assert item.body_text == "公众号正文"
    assert item.image_urls == ["https://img.test/wx-1.jpg"]


def test_douyin_adapter_maps_video_search_and_detail(monkeypatch):
    settings = SimpleNamespace()

    def fake_search(query, *, platform, content_type, limit, settings, rate_limiter):
        assert query == "短视频 讲法"
        assert platform == "douyin"
        assert content_type == "视频"
        assert limit == 1
        return ["dy-1"]

    def fake_fetch(source_id, *, settings, rate_limiter):
        assert source_id == "dy-1"
        return Post(
            id="dy_dy-1",
            platform="douyin",
            url="https://douyin.test/1",
            content_id="dy-1",
            title="短视频讲法",
            content_type="video",
            body_text="视频文案",
            video_urls=["https://video.test/1.mp4"],
            raw={"source": "mock"},
        )

    monkeypatch.setattr(douyin_module, "search_keyword", fake_search)
    monkeypatch.setattr(douyin_module, "fetch_post_detail", fake_fetch)

    adapter = DouyinAdapter()
    candidate = adapter.search("短视频 讲法", settings=settings, limit=1, rate_limiter=None)[0]
    item = adapter.fetch_detail(candidate, settings=settings, rate_limiter=None)

    assert candidate.source_id == "dy-1"
    assert item.content_type == "video"
    assert item.video_urls == ["https://video.test/1.mp4"]