浅念
90 lines
2.6 KiB
Python
90 lines
2.6 KiB
Python
"""Unit tests for the hasLast/end cursor pagination helper in
|
|
:mod:`scripts.scrape_via_page` (WeChat msg/chats deep-history paging).
|
|
|
|
No network/browser — the fetch callable is faked; these verify the
|
|
cursor protocol: page N+1 uses ``data.end - 1`` as ``end`` until
|
|
``hasLast`` is falsy, with guards for malformed ``end`` and a page cap.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from scripts.scrape_via_page import paginate_chats
|
|
|
|
|
|
def _page(messages, has_last, end=1000):
|
|
return {"data": {"messages": messages, "hasLast": has_last, "end": end}}
|
|
|
|
|
|
async def _noop_sleep(_seconds: float) -> None:
|
|
return None
|
|
|
|
|
|
async def test_multi_page_uses_end_minus_one_cursor():
|
|
calls: list[int] = []
|
|
|
|
async def fetch(end_ms: int):
|
|
calls.append(end_ms)
|
|
if len(calls) == 1:
|
|
return _page([{"id": 1}], has_last=True, end=1000)
|
|
return _page([{"id": 2}], has_last=False, end=500)
|
|
|
|
datas = [d async for d in paginate_chats(fetch, end_ms=2000, sleep=_noop_sleep)]
|
|
|
|
assert [len(d["messages"]) for d in datas] == [1, 1]
|
|
assert calls == [2000, 999] # 下一页游标 = end - 1
|
|
|
|
|
|
async def test_stops_when_has_last_falsy():
|
|
async def fetch(end_ms: int):
|
|
return _page([{"id": 1}], has_last=False)
|
|
|
|
datas = [d async for d in paginate_chats(fetch, end_ms=2000, sleep=_noop_sleep)]
|
|
|
|
assert len(datas) == 1
|
|
|
|
|
|
async def test_stops_on_non_dict_data():
|
|
calls = []
|
|
|
|
async def fetch(end_ms: int):
|
|
calls.append(end_ms)
|
|
return {"message": "调用成功", "data": None}
|
|
|
|
datas = [d async for d in paginate_chats(fetch, end_ms=2000, sleep=_noop_sleep)]
|
|
|
|
assert datas == []
|
|
assert len(calls) == 1
|
|
|
|
|
|
async def test_stops_when_end_not_numeric():
|
|
async def fetch(end_ms: int):
|
|
return _page([{"id": 1}], has_last=True, end=None)
|
|
|
|
datas = [d async for d in paginate_chats(fetch, end_ms=2000, sleep=_noop_sleep)]
|
|
|
|
assert len(datas) == 1 # hasLast 真但游标缺失 → 不再无限请求
|
|
|
|
|
|
async def test_page_cap_prevents_infinite_loop():
|
|
calls = []
|
|
|
|
async def fetch(end_ms: int):
|
|
calls.append(end_ms)
|
|
return _page([{"id": len(calls)}], has_last=True, end=end_ms - 1)
|
|
|
|
datas = [d async for d in paginate_chats(fetch, end_ms=2000, max_pages=5, sleep=_noop_sleep)]
|
|
|
|
assert len(datas) == 5
|
|
assert len(calls) == 5
|
|
|
|
|
|
async def test_empty_messages_list_still_pages():
|
|
async def fetch(end_ms: int):
|
|
if end_ms > 1500:
|
|
return _page(None, has_last=True, end=1500)
|
|
return _page([{"id": 1}], has_last=False)
|
|
|
|
datas = [d async for d in paginate_chats(fetch, end_ms=2000, sleep=_noop_sleep)]
|
|
|
|
assert len(datas) == 2
|