MediaCrawler/tests/test_qrlogin.py

# -*- coding: utf-8 -*-
"""Tests for CDP-driven QR login.

The behaviour worth pinning down is not "does it call Playwright" but the two
decisions that silently produce a logged-out crawler if they regress:

* the QR must be read from the browser's **default** context, because a new
  context is an incognito-like profile whose cookies the crawler never sees;
* success must be decided by ``web_session`` *changing*, not merely existing --
  an account already signed in has a value before the operator ever scans.
"""

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."""
    qrlogin._current = None
    yield
    qrlogin._current = None


def _fake_stack(qr_image: str = "data:image/png;base64,AAAA", cookies=None):
    """A Playwright/Chrome stand-in wired the way the real one behaves."""
    context = MagicMock()
    context.cookies = AsyncMock(return_value=list(cookies or []))
    page = MagicMock()
    page.goto = AsyncMock()
    page.close = AsyncMock()
    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 returning 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


def _patch(monkeypatch, manager, qr_image="data:image/png;base64,AAAA"):
    monkeypatch.setattr(qrlogin, "async_playwright", lambda: manager)
    monkeypatch.setattr(
        qrlogin.utils, "find_login_qrcode", AsyncMock(return_value=qr_image)
    )


@pytest.mark.asyncio
async def test_idle_before_any_session():
    snapshot = await qrlogin.status()

    assert snapshot["status"] == qrlogin.STATUS_IDLE
    assert snapshot["image"] == ""


@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, playwright, browser, context, _page = _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()
    playwright.chromium.connect_over_cdp.assert_awaited_once()


@pytest.mark.asyncio
async def test_success_requires_the_cookie_to_change():
    session = qrlogin.QrLoginSession(
        qrlogin.PLATFORM_XHS,
        MagicMock(),
        _page_with_cookies([{"name": "web_session", "value": "before"}]),
        baseline="before",
    )

    await session.refresh()
    assert session.status == qrlogin.STATUS_WAITING

    session._page = _page_with_cookies([{"name": "web_session", "value": "after"}])
    await session.refresh()
    assert session.status == qrlogin.STATUS_SUCCESS


@pytest.mark.asyncio
async def test_an_already_signed_in_profile_needs_no_scan(monkeypatch):
    """No QR on the page plus a session cookie means someone is already logged in."""
    manager, _playwright, _browser, _context, _page = _fake_stack(
        qr_image="", cookies=[{"name": "web_session", "value": "existing"}]
    )
    _patch(monkeypatch, manager, qr_image="")

    snapshot = await qrlogin.start(qrlogin.PLATFORM_XHS)

    assert snapshot["status"] == qrlogin.STATUS_SUCCESS
    assert "登录" in snapshot["message"]


@pytest.mark.asyncio
async def test_no_qr_and_no_session_is_an_error(monkeypatch):
    manager, _playwright, _browser, _context, _page = _fake_stack(qr_image="", cookies=[])
    _patch(monkeypatch, manager, qr_image="")

    snapshot = await qrlogin.start(qrlogin.PLATFORM_XHS)

    assert snapshot["status"] == qrlogin.STATUS_ERROR


@pytest.mark.asyncio
async def test_session_expires(monkeypatch):
    session = qrlogin.QrLoginSession(
        qrlogin.PLATFORM_XHS, MagicMock(), _page_with_cookies([]), baseline=""
    )
    session.started_at -= qrlogin.QR_TTL_SECONDS + 1

    await session.refresh()

    assert session.status == qrlogin.STATUS_EXPIRED


@pytest.mark.asyncio
async def test_closed_tab_is_reported_rather_than_crashing():
    page = MagicMock()
    page.context.cookies = AsyncMock(side_effect=RuntimeError("Target closed"))
    session = qrlogin.QrLoginSession(qrlogin.PLATFORM_XHS, MagicMock(), page, baseline="")

    await session.refresh()

    assert session.status == qrlogin.STATUS_ERROR


@pytest.mark.asyncio
async def test_unreachable_browser_is_reported(monkeypatch):
    """A server whose Chrome is not listening must say so, not 500 anonymously."""
    playwright = MagicMock()
    playwright.chromium.connect_over_cdp = AsyncMock(
        side_effect=OSError("Connection refused")
    )
    playwright.stop = AsyncMock()
    manager = MagicMock()
    manager.start = AsyncMock(return_value=playwright)
    monkeypatch.setattr(qrlogin, "async_playwright", lambda: manager)

    with pytest.raises(RuntimeError) as excinfo:
        await qrlogin.start(qrlogin.PLATFORM_XHS)

    assert "连接浏览器失败" in str(excinfo.value)
    playwright.stop.assert_awaited_once()


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

    snapshot = await qrlogin.cancel()

    assert snapshot["status"] == qrlogin.STATUS_IDLE
    page.close.assert_awaited_once()
    assert (await qrlogin.status())["status"] == qrlogin.STATUS_IDLE


def _page_with_cookies(cookies):
    page = MagicMock()
    page.context.cookies = AsyncMock(return_value=list(cookies))
    return page