live-hub-py/tests/test_huya_selected_login_queue.py

"""虎牙多选登录：单线程、失败重试、号间间隔。"""

from unittest.mock import patch

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from core.huya.app_login import HuyaAppQrAuthRequiredError
from core.huya.login import HuyaCredentialError, HuyaLoginResult
from web.backend.database import Base
from web.backend.models import HuyaAccount, User
from web.backend.routers.huya.login import app_password_login_selected_accounts
from web.backend.schemas import HuyaPasswordLoginSelectedRequest

SUCCESS_COOKIE = (
    "udb_cred=mock_cred_123; yyuid=1199666914671; "
    "udb_uid=1199666914671; username=mock_test_huya_user"
)


def _ok_result() -> HuyaLoginResult:
    return HuyaLoginResult(
        success=True,
        cookie=SUCCESS_COOKIE,
        message="App协议登录成功",
        sdid="mock_sdid_123",
    )


def _fail_result(
    message: str = "网络超时", code: str = "LOGIN_FAILED"
) -> HuyaLoginResult:
    return HuyaLoginResult(success=False, message=message, code=code)


class TestSelectedAppLoginQueue:
    def setup_method(self):
        self.engine = create_engine("sqlite://")
        Base.metadata.create_all(self.engine)
        self.session = sessionmaker(bind=self.engine)()
        self.admin = User(username="admin", password_hash="hash", role="super_admin")
        self.session.add(self.admin)
        self.session.commit()

    def teardown_method(self):
        self.session.close()
        Base.metadata.drop_all(self.engine)
        self.engine.dispose()

    def _add_account(self, username: str, password: str = "pw") -> HuyaAccount:
        account = HuyaAccount(
            username=username,
            account_password=password,
            cookie="",
            status="password_imported",
        )
        self.session.add(account)
        self.session.commit()
        return account

    def _run(self, ids: list[int], **kwargs):
        req = HuyaPasswordLoginSelectedRequest(
            account_ids=ids,
            max_attempts=kwargs.get("max_attempts", 2),
            interval_seconds=kwargs.get("interval_seconds", 8),
            retry_delay_seconds=kwargs.get("retry_delay_seconds", 3),
        )
        return app_password_login_selected_accounts(
            req=req, db=self.session, current=self.admin
        )

    def test_retries_then_succeeds(self):
        account = self._add_account("hy_retry_ok")
        with (
            patch("web.backend.routers.huya.login.time.sleep") as m_sleep,
            patch(
                "web.backend.routers.huya.login.login_huya_app_password",
                side_effect=[_fail_result(), _ok_result()],
            ) as m_login,
        ):
            resp = self._run([account.id])
        assert resp["count"] == 1
        assert resp["failed"] == 0
        assert resp["results"][0]["success"]
        assert resp["results"][0]["attempts"] == 2
        assert m_login.call_count == 2
        m_sleep.assert_called_once_with(3)

    def test_fails_twice_then_moves_on(self):
        first = self._add_account("hy_fail_twice")
        second = self._add_account("hy_next_ok")
        with (
            patch("web.backend.routers.huya.login.time.sleep") as m_sleep,
            patch(
                "web.backend.routers.huya.login.login_huya_app_password",
                side_effect=[_fail_result(), _fail_result(), _ok_result()],
            ) as m_login,
        ):
            resp = self._run(
                [first.id, second.id],
                interval_seconds=8,
                retry_delay_seconds=3,
            )
        assert resp["count"] == 1
        assert resp["failed"] == 1
        assert resp["results"][0]["success"] is False
        assert resp["results"][0]["attempts"] == 2
        assert "已尝试 2 次" in resp["results"][0]["message"]
        assert resp["results"][1]["success"]
        assert m_login.call_count == 3
        assert m_sleep.call_args_list == [((3,),), ((8,),)]

    def test_qr_auth_is_not_retried(self):
        account = self._add_account("hy_need_qr")
        with (
            patch("web.backend.routers.huya.login.time.sleep") as m_sleep,
            patch(
                "web.backend.routers.huya.login.login_huya_app_password",
                side_effect=HuyaAppQrAuthRequiredError("要求扫码验证"),
            ) as m_login,
        ):
            resp = self._run([account.id])
        assert resp["failed"] == 1
        assert resp["results"][0]["attempts"] == 1
        assert m_login.call_count == 1
        m_sleep.assert_not_called()

    def test_wrong_password_is_not_retried(self):
        account = self._add_account("hy_bad_pw")
        with (
            patch("web.backend.routers.huya.login.time.sleep") as m_sleep,
            patch(
                "web.backend.routers.huya.login.login_huya_app_password",
                side_effect=HuyaCredentialError("账号或密码错误"),
            ) as m_login,
        ):
            resp = self._run([account.id])
        assert resp["failed"] == 1
        assert resp["results"][0]["attempts"] == 1
        assert m_login.call_count == 1
        m_sleep.assert_not_called()

    def test_sms_auth_result_is_not_retried(self):
        account = self._add_account("hy_sms")
        with (
            patch("web.backend.routers.huya.login.time.sleep") as m_sleep,
            patch(
                "web.backend.routers.huya.login.login_huya_app_password",
                return_value=_fail_result("需要短信验证", "SMS_AUTH_REQUIRED"),
            ) as m_login,
        ):
            resp = self._run([account.id])
        assert resp["failed"] == 1
        assert resp["results"][0]["attempts"] == 1
        assert m_login.call_count == 1
        m_sleep.assert_not_called()