multi-platform-media-manage/server/python/test_xiaohongshu_paging_logic.py

import asyncio
import sys
from pathlib import Path

CURRENT_DIR = Path(__file__).parent.resolve()
if str(CURRENT_DIR) not in sys.path:
    sys.path.insert(0, str(CURRENT_DIR))

from platforms.xiaohongshu import XiaohongshuPublisher


async def main():
    publisher = XiaohongshuPublisher(headless=True)

    async def fake_fetch_notes_page(p: int):
        if str(p) == "0":
            return {
                "success": True,
                "data": {
                    "notes": [{"id": "1"}, {"id": "2"}],
                    "tags": [{"id": "special.note_time_desc", "notes_count": 3}],
                    "page": 1,
                },
            }
        if str(p) == "1":
            return {
                "success": True,
                "data": {
                    "notes": [{"id": "3"}],
                    "tags": [{"id": "special.note_time_desc", "notes_count": 3}],
                    "page": -1,
                },
            }
        return {"success": True, "data": {"notes": []}}

    class DummyPage:
        url = "https://creator.xiaohongshu.com/new/note-manager"

        async def goto(self, *_args, **_kwargs):
            return None

        async def evaluate(self, _script, p):
            return await fake_fetch_notes_page(p)

    publisher.page = DummyPage()
    publisher.context = object()
    publisher.browser = object()

    async def dummy_init_browser():
        return None

    async def dummy_set_cookies(_cookies):
        return None

    publisher.init_browser = dummy_init_browser  # type: ignore
    publisher.set_cookies = dummy_set_cookies  # type: ignore
    publisher.parse_cookies = lambda _c: []  # type: ignore
    publisher.close_browser = lambda: asyncio.sleep(0)  # type: ignore

    r0 = await publisher.get_works("[]", page=0, page_size=20)
    assert r0.success
    assert r0.total == 3
    assert r0.has_more is True
    assert r0.next_page == 1

    r1 = await publisher.get_works("[]", page=1, page_size=20)
    assert r1.success
    assert r1.total == 3
    assert r1.has_more is False
    assert r1.next_page == -1

    print("test_xiaohongshu_paging_logic passed")

    r_all = await publisher.get_all_works("[]")
    assert r_all.success
    assert r_all.total == 3
    assert len(r_all.works) == 3
    assert r_all.has_more is False
    print("test_xiaohongshu_auto_paging passed")


if __name__ == "__main__":
    asyncio.run(main())