butubb
44cbe8e2aa
顺序反了：原先先开页面、读二维码，读完才发现「已经登录、没有二维码」。而读二维码内部 会 wait_for_selector 等满 30 秒才放弃 —— 用户点一下按钮要干等半分钟，还白开一个标签页。 实测日志里就是 `Page.wait_for_selector: Timeout 30000ms exceeded`。 改成先问登录状态（那是一次接口调用，很快），已登录就直接返回，根本不碰页面。 新增测试守住这个顺序：已登录时 context.new_page 不得被调用。
269 lines
9.5 KiB
Python
269 lines
9.5 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""监控侧扫码登录与登录态检测。
|
||
|
||
两个要点在这里被钉死：
|
||
|
||
* 二维码必须从浏览器**默认 context** 里读 —— 新建 context 是无痕式的 profile，
|
||
扫了也白扫，爬虫读不到那份 cookie；
|
||
* 「登录了吗」不能用页面里的 `window.__INITIAL_STATE__`。那是**页面加载那一刻的快照**：
|
||
浏览器本来就登录着时它是对的，但扫码是加载**之后**才登录的，快照不会翻转，
|
||
于是扫完码界面会一直停在二维码上。判据改成拿 cookie 问后台接口。
|
||
"""
|
||
|
||
from unittest.mock import AsyncMock, MagicMock
|
||
|
||
import pytest
|
||
|
||
from api.creator.client import CreatorApiError
|
||
from api.monitor import qrlogin
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _reset_module_state():
|
||
for attribute in ("_current", "_page", "_playwright", "_state_cache"):
|
||
setattr(qrlogin, attribute, None)
|
||
yield
|
||
for attribute in ("_current", "_page", "_playwright", "_state_cache"):
|
||
setattr(qrlogin, attribute, None)
|
||
|
||
|
||
XHS_COOKIES = [
|
||
{"name": "a1", "value": "an-a1-value"},
|
||
{"name": "web_session", "value": "a-session"},
|
||
]
|
||
|
||
|
||
def _fake_stack(cookies=None, qr="data:image/png;base64,AAAA"):
|
||
"""Chrome/Playwright 替身，行为与真实的一致。"""
|
||
page = MagicMock()
|
||
page.url = "https://www.xiaohongshu.com/explore"
|
||
page.is_closed = MagicMock(return_value=False)
|
||
page.goto = AsyncMock()
|
||
page.close = AsyncMock()
|
||
|
||
context = MagicMock()
|
||
context.pages = []
|
||
context.cookies = AsyncMock(return_value=list(cookies if cookies is not None else XHS_COOKIES))
|
||
context.new_page = AsyncMock(return_value=page)
|
||
|
||
browser = MagicMock()
|
||
browser.contexts = [context]
|
||
# 去新建 context 正是这里要防的 bug，所以让它直接炸，而不是悄悄返回一个无痕 profile。
|
||
browser.new_context = AsyncMock(
|
||
side_effect=AssertionError("must reuse browser.contexts[0], not a new context")
|
||
)
|
||
|
||
playwright = MagicMock()
|
||
playwright.chromium.connect_over_cdp = AsyncMock(return_value=browser)
|
||
playwright.stop = AsyncMock()
|
||
|
||
manager = MagicMock()
|
||
manager.start = AsyncMock(return_value=playwright)
|
||
|
||
return manager, playwright, browser, context, page, qr
|
||
|
||
|
||
def _patch(monkeypatch, manager, qr="data:image/png;base64,AAAA", resolver=None):
|
||
"""``resolver(cookie)`` 返回账号信息 dict，或抛 CreatorApiError。"""
|
||
if resolver is None:
|
||
resolver = lambda _cookie: {"user_id": "u1", "nickname": "小明"} # noqa: E731
|
||
|
||
class _Client:
|
||
def __init__(self, cookie, **kwargs):
|
||
self.cookie = cookie
|
||
|
||
async def fetch_user_info(self):
|
||
return resolver(self.cookie)
|
||
|
||
monkeypatch.setattr(qrlogin, "async_playwright", lambda: manager)
|
||
monkeypatch.setattr(qrlogin, "CreatorClient", _Client)
|
||
monkeypatch.setattr(qrlogin.utils, "find_login_qrcode", AsyncMock(return_value=qr))
|
||
|
||
|
||
def _signed_out(_cookie):
|
||
raise CreatorApiError("登录态无效或已过期", status=401)
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_idle_reports_the_browsers_login_state(monkeypatch):
|
||
manager, *_ = _fake_stack()
|
||
_patch(monkeypatch, manager, resolver=lambda _c: {"user_id": "u1", "nickname": "小明"})
|
||
|
||
snapshot = await qrlogin.status()
|
||
|
||
assert snapshot["status"] == qrlogin.STATUS_IDLE
|
||
assert snapshot["logged_in"] is True
|
||
assert snapshot["nickname"] == "小明"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_unwired_platform_is_rejected():
|
||
"""只有小红书接了扫码；别的平台必须直接报错，而不是给个按不动的按钮。"""
|
||
with pytest.raises(ValueError):
|
||
await qrlogin.start("dy")
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_start_reads_the_qr_from_the_default_context(monkeypatch):
|
||
manager, _pw, browser, context, _page, _qr = _fake_stack()
|
||
_patch(monkeypatch, manager, resolver=_signed_out)
|
||
|
||
snapshot = await qrlogin.start(qrlogin.PLATFORM_XHS)
|
||
|
||
assert snapshot["status"] == qrlogin.STATUS_WAITING
|
||
assert snapshot["image"] == "data:image/png;base64,AAAA"
|
||
context.new_page.assert_awaited_once()
|
||
browser.new_context.assert_not_called()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_start_does_not_open_a_second_tab(monkeypatch):
|
||
"""已有的 xhs 标签页会被认领，所以重启不会在浏览器里堆孤儿页。"""
|
||
manager, _pw, _browser, context, page, _qr = _fake_stack()
|
||
context.pages = [page]
|
||
_patch(monkeypatch, manager, resolver=_signed_out)
|
||
|
||
await qrlogin.start(qrlogin.PLATFORM_XHS)
|
||
|
||
context.new_page.assert_not_called()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_an_already_signed_in_profile_needs_no_scan(monkeypatch):
|
||
"""没二维码但 profile 已登录 —— 这是成功，不是失败。"""
|
||
manager, *_rest = _fake_stack()
|
||
_patch(monkeypatch, manager, qr="", resolver=lambda _c: {"user_id": "u9", "nickname": "老王"})
|
||
|
||
snapshot = await qrlogin.start(qrlogin.PLATFORM_XHS)
|
||
|
||
assert snapshot["status"] == qrlogin.STATUS_SUCCESS
|
||
assert snapshot["nickname"] == "老王"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_an_already_signed_in_profile_never_opens_a_page(monkeypatch):
|
||
"""已登录时**根本不该去开页面**。
|
||
|
||
读二维码内部会 wait_for_selector 等满 30 秒才放弃，而已经登录时页面上没有二维码 ——
|
||
顺序反了的话，用户点一下按钮要干等半分钟，还白开一个标签页。
|
||
"""
|
||
manager, _pw, _browser, context, _page, _qr = _fake_stack()
|
||
_patch(monkeypatch, manager, resolver=lambda _c: {"user_id": "u1", "nickname": "小明"})
|
||
|
||
await qrlogin.start(qrlogin.PLATFORM_XHS)
|
||
|
||
context.new_page.assert_not_called()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_no_qr_and_not_signed_in_is_an_error(monkeypatch):
|
||
manager, *_rest = _fake_stack()
|
||
_patch(monkeypatch, manager, qr="", resolver=_signed_out)
|
||
|
||
snapshot = await qrlogin.start(qrlogin.PLATFORM_XHS)
|
||
|
||
assert snapshot["status"] == qrlogin.STATUS_ERROR
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_a_completed_scan_flips_the_session_to_success(monkeypatch):
|
||
"""**这个判据是重点**：扫码是页面加载之后才发生的，所以不能用页面快照来判断。"""
|
||
manager, *_rest = _fake_stack()
|
||
signed_in = {"value": False}
|
||
|
||
def resolver(cookie):
|
||
assert "a1=an-a1-value" in cookie # 判据必须真的用 cookie 去问
|
||
if not signed_in["value"]:
|
||
raise CreatorApiError("登录态无效或已过期", status=401)
|
||
return {"user_id": "u1", "nickname": "小红"}
|
||
|
||
_patch(monkeypatch, manager, resolver=resolver)
|
||
await qrlogin.start(qrlogin.PLATFORM_XHS)
|
||
assert qrlogin._current.status == qrlogin.STATUS_WAITING
|
||
|
||
# 操作者扫了码
|
||
signed_in["value"] = True
|
||
qrlogin._state_cache = None # 5 秒缓存否则会遮住这次变化
|
||
|
||
snapshot = await qrlogin.status()
|
||
|
||
assert snapshot["status"] == qrlogin.STATUS_SUCCESS
|
||
assert snapshot["nickname"] == "小红"
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_the_successful_session_hands_over_a_cookie(monkeypatch):
|
||
"""扫码不该只写浏览器 profile —— 还要能把 cookie 交出来存库，
|
||
否则关掉 CDP 就断了。"""
|
||
manager, *_rest = _fake_stack()
|
||
_patch(monkeypatch, manager, resolver=lambda _c: {"user_id": "u1", "nickname": "小明"})
|
||
|
||
await qrlogin.start(qrlogin.PLATFORM_XHS)
|
||
cookie = await qrlogin.take_cookie()
|
||
|
||
assert cookie is not None
|
||
assert "a1=an-a1-value" in cookie
|
||
# 只能取一次，否则每次轮询都会重复写库
|
||
assert await qrlogin.take_cookie() is None
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_session_expires(monkeypatch):
|
||
manager, *_rest = _fake_stack()
|
||
_patch(monkeypatch, manager, resolver=_signed_out)
|
||
await qrlogin.start(qrlogin.PLATFORM_XHS)
|
||
|
||
qrlogin._current.started_at -= qrlogin.QR_TTL_SECONDS + 1
|
||
snapshot = await qrlogin.status()
|
||
|
||
assert snapshot["status"] == qrlogin.STATUS_EXPIRED
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_check_login_state_reports_when_the_browser_cannot_answer(monkeypatch):
|
||
"""连不上浏览器时要说出来，不能悄悄报成「未登录」。"""
|
||
manager, _pw, _browser, context, _page, _qr = _fake_stack()
|
||
context.cookies = AsyncMock(side_effect=RuntimeError("Target closed"))
|
||
_patch(monkeypatch, manager)
|
||
|
||
state = await qrlogin.check_login_state()
|
||
|
||
assert state["known"] is False
|
||
assert state["logged_in"] is False
|
||
assert "Target closed" in state["error"]
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_check_login_state_is_cached(monkeypatch):
|
||
manager, _pw, _browser, context, _page, _qr = _fake_stack()
|
||
_patch(monkeypatch, manager, resolver=lambda _c: {"user_id": "u1", "nickname": "x"})
|
||
|
||
await qrlogin.check_login_state()
|
||
await qrlogin.check_login_state()
|
||
|
||
context.cookies.assert_awaited_once()
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_force_bypasses_the_cache(monkeypatch):
|
||
manager, _pw, _browser, context, _page, _qr = _fake_stack()
|
||
_patch(monkeypatch, manager, resolver=lambda _c: {"user_id": "u1", "nickname": "x"})
|
||
|
||
await qrlogin.check_login_state()
|
||
await qrlogin.check_login_state(force=True)
|
||
|
||
assert context.cookies.await_count == 2
|
||
|
||
|
||
@pytest.mark.asyncio
|
||
async def test_cancel_keeps_the_operators_tab(monkeypatch):
|
||
"""与运营模块不同：那里的上下文是临时的、用完即弃；这里的标签页属于操作者的浏览器。"""
|
||
manager, _pw, _browser, _context, page, _qr = _fake_stack()
|
||
_patch(monkeypatch, manager, resolver=_signed_out)
|
||
await qrlogin.start(qrlogin.PLATFORM_XHS)
|
||
|
||
snapshot = await qrlogin.cancel()
|
||
|
||
assert snapshot["status"] == qrlogin.STATUS_IDLE
|
||
page.close.assert_not_called()
|