Files
MediaCrawler/tests/test_creator_login.py
T
butubb bef0a4fbde
Deploy VitePress site to Pages / build (push) Canceled after 0s
Deploy VitePress site to Pages / Deploy (push) Canceled after 0s
fix(creator): 扫码成功却一直停在二维码上 + 运营改为左右布局
【扫码不完成】根因是判据本身。原先读页面里的 window.__INITIAL_STATE__,而那是
页面加载那一刻的快照:监控那边的同一探针能用,是因为那台浏览器的页面加载时就已经
登录了;而扫码是加载之后才登录的 —— SPA 内部确实登进去了,但初始快照不会翻转,
于是检测永远等不到。

改成拿 cookie 直接问创作者后台 /api/galaxy/user/info 我是谁。实测这个判据很干净:
游客也会拿到 a1(所以签名算得出来),但接口直接回 401 无登录信息;只有真正登录了
才返回 user_id。所以「有 a1」什么都证明不了,后台认了才算。
顺带按 5 秒节流 —— 前端每 2 秒问一次,没必要每次都打后台接口。

【弹窗不关】成功后不自动关闭,停在二维码上会让人以为没成功。现在显示账号卡片与原话
提示,1.6 秒后自动关闭并提供一个「完成」按钮。

【已完成结果会残留】take_cookie 取走 cookie 就拆会话,而在飞的轮询会看到 _current 为空
回报 idle,把已显示的成功能擦掉。现在把结果记在模块里重复返回,关闭弹窗时清掉 ——
否则下次打开会立刻显示上次的成功。

【布局】按用户要求改成左右两栏(左账号列表、右数据面板),与监控统一,取消二级菜单。
未选过时默认选中第一个,右栏不会一开始就是空的。

测试:tests/test_creator_login.py 新增 7 例,含「游客会话永不完成」「接口不打满每次轮询」
「临时上下文用完必须关掉」。
2026-10-07 16:40:24 +08:00

161 lines
4.7 KiB
Python

# -*- coding: utf-8 -*-
"""运营账号扫码登录的完成判据。
这里守的是一个具体的故障:判据原先读页面里的 `window.__INITIAL_STATE__`,而那是
**页面加载那一刻的快照** —— 扫码是加载之后才登录的,快照不会翻转,于是登录明明
成功了,界面却永远停在二维码上。
现在改成拿 cookie 去问创作者后台"我是谁"。重要的是**游客也有 a1**(所以签名算得
出来),所以"有 a1"什么都不能证明,只有后台认了才算数。
"""
from unittest.mock import AsyncMock, MagicMock
import pytest
from api.creator import login as creator_login
from api.creator.client import CreatorApiError
class _FakeContext:
def __init__(self, cookies):
self._cookies = cookies
self.closed = False
async def cookies(self):
return self._cookies
async def close(self):
self.closed = True
class _FakePage:
def __init__(self):
self.closed = False
async def close(self):
self.closed = True
def _session(cookies):
context = _FakeContext(cookies)
page = _FakePage()
session = creator_login.AccountLoginSession(context, page)
# 跳过节流,让每次 refresh 都真的去问一次。
session._last_login_check = 0.0
return session, context, page
def _patch_client(monkeypatch, behaviour):
"""behaviour(cookie) -> dict 或抛异常。"""
class _Client:
def __init__(self, cookie, **kwargs):
self.cookie = cookie
async def fetch_user_info(self):
return behaviour(self.cookie)
monkeypatch.setattr(creator_login, "CreatorClient", _Client)
GUEST_COOKIES = [
{"name": "a1", "value": "guest-a1"},
{"name": "web_session", "value": "guest-session"},
]
@pytest.mark.asyncio
async def test_a_guest_session_never_completes(monkeypatch):
"""游客也有 a1,但后台回 401 —— 必须继续等,不能当成登录成功。"""
def behaviour(_cookie):
raise CreatorApiError("登录态无效或已过期", status=401)
_patch_client(monkeypatch, behaviour)
session, _context, _page = _session(GUEST_COOKIES)
await session.refresh()
assert session.status == creator_login.STATUS_WAITING
assert session.cookie == ""
@pytest.mark.asyncio
async def test_a_recognised_identity_completes_the_login(monkeypatch):
_patch_client(
monkeypatch,
lambda _cookie: {"user_id": "u123", "nickname": "小明", "red_id": "1"},
)
session, _context, _page = _session(GUEST_COOKIES)
await session.refresh()
assert session.status == creator_login.STATUS_SUCCESS
assert "小明" in session.message
assert "a1=guest-a1" in session.cookie
assert session.account["user_id"] == "u123"
@pytest.mark.asyncio
async def test_an_empty_identity_does_not_complete(monkeypatch):
"""接口返回 200 但没有账号标识 —— 同样不能算成功。"""
_patch_client(monkeypatch, lambda _cookie: {"user_id": "", "nickname": ""})
session, _context, _page = _session(GUEST_COOKIES)
await session.refresh()
assert session.status == creator_login.STATUS_WAITING
@pytest.mark.asyncio
async def test_the_api_is_not_hit_on_every_poll(monkeypatch):
"""前端每 2 秒轮询一次,但每次轮询都打一次后台接口是浪费。"""
calls = []
def behaviour(cookie):
calls.append(cookie)
raise CreatorApiError("还没登录", status=401)
_patch_client(monkeypatch, behaviour)
session, _context, _page = _session(GUEST_COOKIES)
await session.refresh()
# 第一次之后 _last_login_check 已经是"现在",后面两次应落在节流窗口内。
await session.refresh()
await session.refresh()
assert len(calls) == 1
@pytest.mark.asyncio
async def test_expiry_beats_a_successful_scan(monkeypatch):
_patch_client(monkeypatch, lambda _cookie: {"user_id": "u1", "nickname": "x"})
session, _context, _page = _session(GUEST_COOKIES)
session.started_at -= creator_login.QR_TTL_SECONDS + 1
await session.refresh()
assert session.status == creator_login.STATUS_EXPIRED
@pytest.mark.asyncio
async def test_a_closed_window_is_reported(monkeypatch):
session, context, _page = _session(GUEST_COOKIES)
context.cookies = AsyncMock(side_effect=RuntimeError("Target closed"))
await session.refresh()
assert session.status == creator_login.STATUS_ERROR
@pytest.mark.asyncio
async def test_closing_discards_the_temporary_context():
"""临时上下文是这个设计的关键 —— 用完必须关掉,否则会挂在操作者的 Chrome 里。"""
session, context, page = _session(GUEST_COOKIES)
await session.close()
assert context.closed is True
assert page.closed is True