MediaCrawler/tests/test_qrlogin.py

# -*- coding: utf-8 -*-
"""Tests for CDP-driven QR login and the login-state check.

Two things are worth pinning down here, because both silently produce a
logged-out crawler when they regress:

* the QR must be read from the browser's **default** context, since a new context
  is an incognito-like profile whose cookies the crawler never sees;
* "am I logged in?" must be answered by the browser's profile, **not** by the
  in-memory scan session -- that session dies on any restart, so a successful scan
  would otherwise look like nothing happened.
"""

import json
from unittest.mock import AsyncMock, MagicMock

import pytest

from api.monitor import qrlogin


@pytest.fixture(autouse=True)
def _reset_module_state():
    """Each test starts from "nothing on screen" and leaves it that way."""
    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)


def _probe_result(logged_in: bool, nickname=None, known: bool = True) -> str:
    return json.dumps({"known": known, "loggedIn": logged_in, "nickname": nickname})


def _fake_stack(logged_in: bool = False, nickname=None, qr: str = "data:image/png;base64,AAAA"):
    """A Playwright/Chrome stand-in wired the way the real one behaves."""
    page = MagicMock()
    page.url = "https://www.xiaohongshu.com/explore"
    page.is_closed = MagicMock(return_value=False)
    page.goto = AsyncMock()
    page.close = AsyncMock()
    page.evaluate = AsyncMock(return_value=_probe_result(logged_in, nickname))

    context = MagicMock()
    context.pages = []
    context.new_page = AsyncMock(return_value=page)

    browser = MagicMock()
    browser.contexts = [context]
    # Reaching for a fresh context is the bug this guards against, so make it blow
    # up loudly rather than quietly handing back a throwaway 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"):
    monkeypatch.setattr(qrlogin, "async_playwright", lambda: manager)
    monkeypatch.setattr(qrlogin.utils, "find_login_qrcode", AsyncMock(return_value=qr))


@pytest.mark.asyncio
async def test_idle_reports_the_browsers_login_state(monkeypatch):
    manager, _pw, _browser, _context, _page, _qr = _fake_stack(logged_in=True, nickname="小明")
    _patch(monkeypatch, manager)

    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():
    """Only xhs is wired; anything else must fail loudly, not show a dead button."""
    with pytest.raises(ValueError):
        await qrlogin.start("dy")


@pytest.mark.asyncio
async def test_start_returns_the_qr_and_reuses_the_default_context(monkeypatch):
    manager, _pw, browser, context, _page, _qr = _fake_stack()
    _patch(monkeypatch, manager)

    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):
    """An open xhs tab is adopted, so restarts do not litter the browser."""
    manager, _pw, _browser, context, page, _qr = _fake_stack()
    context.pages = [page]
    _patch(monkeypatch, manager)

    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):
    """No QR on the page plus a signed-in profile is success, not failure."""
    manager, _pw, _browser, _context, _page, _qr = _fake_stack(
        logged_in=True, nickname="老王"
    )
    _patch(monkeypatch, manager, qr="")

    snapshot = await qrlogin.start(qrlogin.PLATFORM_XHS)

    assert snapshot["status"] == qrlogin.STATUS_SUCCESS
    assert snapshot["nickname"] == "老王"


@pytest.mark.asyncio
async def test_no_qr_and_not_signed_in_is_an_error(monkeypatch):
    manager, _pw, _browser, _context, _page, _qr = _fake_stack(logged_in=False)
    _patch(monkeypatch, manager, qr="")

    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, _pw, _browser, _context, page, _qr = _fake_stack(logged_in=False)
    _patch(monkeypatch, manager)
    await qrlogin.start(qrlogin.PLATFORM_XHS)

    # The operator scans: the page now reports a signed-in profile.
    page.evaluate = AsyncMock(return_value=_probe_result(True, "小红"))
    qrlogin._state_cache = None  # the 5s cache would otherwise hide the change

    snapshot = await qrlogin.status()

    assert snapshot["status"] == qrlogin.STATUS_SUCCESS
    assert snapshot["nickname"] == "小红"
    assert "登录成功" in snapshot["message"]


@pytest.mark.asyncio
async def test_session_expires(monkeypatch):
    manager, _pw, _browser, _context, _page, _qr = _fake_stack()
    _patch(monkeypatch, manager)
    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_page_cannot_answer(monkeypatch):
    """An unreachable browser must say so, not quietly report "not logged in"."""
    manager, _pw, _browser, _context, page, _qr = _fake_stack()
    page.evaluate = 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(logged_in=True)
    _patch(monkeypatch, manager)

    await qrlogin.check_login_state()
    await qrlogin.check_login_state()

    page.evaluate.assert_awaited_once()


@pytest.mark.asyncio
async def test_force_bypasses_the_cache(monkeypatch):
    manager, _pw, _browser, _context, page, _qr = _fake_stack(logged_in=True)
    _patch(monkeypatch, manager)

    await qrlogin.check_login_state()
    await qrlogin.check_login_state(force=True)

    assert page.evaluate.await_count == 2
    page.goto.assert_awaited()  # force reloads before reading


@pytest.mark.asyncio
async def test_cancel_resets_without_closing_the_operators_tab(monkeypatch):
    manager, _pw, _browser, _context, page, _qr = _fake_stack()
    _patch(monkeypatch, manager)
    await qrlogin.start(qrlogin.PLATFORM_XHS)

    snapshot = await qrlogin.cancel()

    assert snapshot["status"] == qrlogin.STATUS_IDLE
    # The tab belongs to the operator's browser and is reused, not closed.
    page.close.assert_not_called()


@pytest.mark.asyncio
async def test_shutdown_does_not_close_the_tab(monkeypatch):
    manager, playwright, _browser, _context, page, _qr = _fake_stack()
    _patch(monkeypatch, manager)
    await qrlogin.start(qrlogin.PLATFORM_XHS)

    await qrlogin.shutdown()

    page.close.assert_not_called()
    playwright.stop.assert_awaited()