Merge branch 'fix/pyright-env-and-type-guards' into dev

2026-10-08 12:34:11 +08:00
parent 224b147735 a3c3a46b9c
commit 88422f2f05
31 changed files with 599 additions and 96 deletions
@@ -0,0 +1,120 @@
+"""虎牙「短信接码一键登录」端点：调用契约与成功/失败两条分支。
+
+背景（这是本文件的**存在理由**，别删）：该端点曾引用一个从来不存在的
+``core.huya.auto_register.auto_register_huya_phone``（git 全历史都没有 ``def``），
+所以每次调用都在 import 阶段抛 ``ImportError`` → HTTP 500，且因为导入写在函数体
+内部，静态检查之外没有任何东西拦得住它。用例锁死「端点必须调用真实存在的
+``register_huya_with_sms_line``」这个契约，名字再被改就不会静默回归。
+"""
+
+import pytest
+from fastapi import HTTPException
+
+from core.huya.auto_register import HuyaAutoRegisterResult
+from tests.support.db import get_test_engine, make_test_session
+from web.backend.models import HuyaAccount, User
+from web.backend.routers.huya.login import auto_sms_login_account
+
+# 能被 parse_huya_cookie_line 认出来的最小虎牙 CK（含 yyuid= / udb_uid=）。
+LOGIN_COOKIE = "yyuid=100; udb_uid=100"
+
+
+def _ok_result(item, **kwargs):
+    return HuyaAutoRegisterResult(
+        phone=item.phone,
+        provider=item.provider,
+        success=True,
+        status="success",
+        message="接码成功",
+        cookie=LOGIN_COOKIE,
+    )
+
+
+class TestAutoSmsLoginAccount:
+    def setup_method(self):
+        self.engine = get_test_engine()
+        self.session = make_test_session(self.engine)
+        self.user = User(username="operator", password_hash="hash", role="super_admin")
+        self.account = HuyaAccount(
+            uid="",
+            yyuid="",
+            cookie="",
+            game_phone="13800000000",
+            sms_url="https://sms.example.com/get?p=1",
+        )
+        self.session.add_all([self.user, self.account])
+        self.session.commit()
+
+    def teardown_method(self):
+        self.session.close()
+
+    def _call(self):
+        return auto_sms_login_account(
+            self.account.id, db=self.session, current=self.user
+        )
+
+    def test_calls_existing_register_helper_and_saves_cookie(self, monkeypatch):
+        """核心契约：必须调用真实存在的注册函数，并把返回的 CK 回填落库。"""
+        calls: dict = {}
+
+        def fake_register(item, **kwargs):
+            calls["item"] = item
+            calls["kwargs"] = kwargs
+            return _ok_result(item, **kwargs)
+
+        # 用字符串形式 patch 目标模块属性：端点内部是 `from ... import ...`，
+        # 每次调用都会重新取模块属性，所以这层 patch 必须生效。
+        monkeypatch.setattr(
+            "core.huya.auto_register.register_huya_with_sms_line", fake_register
+        )
+
+        out = self._call()
+
+        assert out["success"] is True
+        # 断言序列化结果，不看属性（AGENTS.md 坑 #2）
+        assert out["account"].model_dump()["uid"] == "100"
+        assert calls["item"].phone == "13800000000"
+        assert calls["item"].url == "https://sms.example.com/get?p=1"
+        # 超时/轮询间隔是端点的既有行为，不能被顺手改掉
+        assert calls["kwargs"]["wait_seconds"] == 90
+        assert calls["kwargs"]["poll_interval"] == 2
+
+        self.session.refresh(self.account)
+        assert self.account.status == "active"
+        assert self.account.login_channel == "sms"
+        assert "udb_uid=100" in self.account.cookie
+
+    def test_failure_marks_account_login_failed(self, monkeypatch):
+        """注册失败要落库成 login_failed 并返回 502，而不是假装成功。"""
+
+        def fake_register(item, **kwargs):
+            return HuyaAutoRegisterResult(
+                phone=item.phone,
+                provider=item.provider,
+                success=False,
+                status="failed",
+                message="接码平台超时",
+            )
+
+        monkeypatch.setattr(
+            "core.huya.auto_register.register_huya_with_sms_line", fake_register
+        )
+
+        with pytest.raises(HTTPException) as excinfo:
+            self._call()
+
+        assert excinfo.value.status_code == 502
+        assert "接码平台超时" in str(excinfo.value.detail)
+
+        self.session.refresh(self.account)
+        assert self.account.status == "login_failed"
+
+    def test_missing_phone_or_sms_url_is_rejected_before_import(self):
+        """缺手机号/接码链接要在调用注册前 400，不能先崩在 import 上。"""
+        self.account.game_phone = ""
+        self.session.commit()
+
+        with pytest.raises(HTTPException) as excinfo:
+            self._call()
+
+        assert excinfo.value.status_code == 400
@@ -1,3 +1,5 @@
+import io
+import json
 from pathlib import Path

 from core.huya import browser_harvest
@@ -51,3 +53,50 @@ def test_launch_can_show_window_when_requested(monkeypatch):
    browser_harvest.launch(Path("/tmp/test-huya-profile"), 9556, headless=False)

    assert "--headless=new" not in calls[0]
+
+
+def test_launch_prefers_page_target_over_browser_endpoint(monkeypatch):
+    """页面级 ws 必须优先于 browser 级（Network 域只能挂在 page target 上）。
+
+    这里原本写的是 ``pages[0]``，而 ``pages`` 是集合推导出来的 **set** ——
+    ``set`` 不可下标，抛出的 TypeError 又被外层 ``except Exception: pass``
+    吞掉，于是每次都退回 browser 级 endpoint：页面级探测**从来没有成功过一次**，
+    而且没有任何日志能看出来。这就是本用例存在的意义。
+    """
+    page_ws = "ws://127.0.0.1/devtools/page/PAGE"
+    browser_ws = "ws://127.0.0.1/devtools/browser/test"
+    payload = json.dumps([{"type": "page", "webSocketDebuggerUrl": page_ws}])
+
+    monkeypatch.setattr(
+        browser_harvest.subprocess, "Popen", lambda args, **kwargs: _FakeProcess()
+    )
+    # _FakeProcess.stderr 会给出 browser 级 endpoint，用来证明「选中了页面级」
+    monkeypatch.setattr(
+        browser_harvest.urllib.request,
+        "urlopen",
+        lambda *args, **kwargs: io.StringIO(payload),
+    )
+
+    _proc, ws_url = browser_harvest.launch(Path("/tmp/test-huya-profile"), 9557)
+
+    assert ws_url == page_ws
+    assert ws_url != browser_ws
+
+
+def test_launch_falls_back_to_browser_endpoint_without_page_target(monkeypatch):
+    """没有 page target 时才允许退回 browser 级 endpoint。"""
+    browser_ws = "ws://127.0.0.1/devtools/browser/test"
+    payload = json.dumps([{"type": "service_worker", "webSocketDebuggerUrl": "ws://x"}])
+
+    monkeypatch.setattr(
+        browser_harvest.subprocess, "Popen", lambda args, **kwargs: _FakeProcess()
+    )
+    monkeypatch.setattr(
+        browser_harvest.urllib.request,
+        "urlopen",
+        lambda *args, **kwargs: io.StringIO(payload),
+    )
+
+    _proc, ws_url = browser_harvest.launch(Path("/tmp/test-huya-profile"), 9558)
+
+    assert ws_url == browser_ws