live-hub-py/tests/test_account_check_api_strategy.py

"""账号检测可选择登录接口（wgapi / iframe）。

背景：账号检测原来把 `WgapiLoginAPI()` **写死在**执行器里，而登录任务早就
能让用户在下拉里选 wgapi / iframe。结果是同一个人、同一个账号，
登录任务换 iframe 能收到邮箱验证码，账号检测却永远只有 wgapi 一条路。

本文件守住三件事：
1. 默认仍是 wgapi —— 不改变账号检测原来的行为；
2. 选了 iframe 后，执行器**真的**把 IframeLoginAPI 交给了 DouyuLogin
   （只在批次的快照里记一个字符串不算数）；
3. HTTP 层把 `api_strategy` 透传到执行器，不会中途被丢掉。
"""

from tests.support.db import get_test_engine, make_test_session
from web.backend.models import User
from web.backend.services import account_check_service as svc


class _Base:
    def setup_method(self):
        self.engine = get_test_engine()
        self.session = make_test_session(self.engine)
        self.user = User(username="chk-strategy", password_hash="h", role="super_admin")
        self.session.add(self.user)
        self.session.commit()

    def teardown_method(self):
        self.session.close()

    def _runner(self, strategy: str | None = None):
        inputs = svc.parse_account_check_lines("u1----p1----e1----ep1")
        kwargs = {}
        if strategy is not None:
            kwargs["api_strategy"] = strategy
        return svc.AccountCheckRegistry().create(
            accounts=inputs,
            created_by=self.user.id,
            concurrency=1,
            max_login_retries=0,
            max_total_time=60,
            **kwargs,
        )


class TestCoreFactory:
    """core 层只认 wgapi / iframe，未知值回退默认。"""

    def test_known_names(self):
        from core.douyu import create_login_api_strategy

        assert create_login_api_strategy("wgapi").name == "wgapi"
        assert create_login_api_strategy("iframe").name == "iframe"

    def test_unknown_falls_back_to_wgapi(self):
        from core.douyu import create_login_api_strategy

        assert create_login_api_strategy("").name == "wgapi"
        assert create_login_api_strategy("IFRAME").name == "iframe"
        assert create_login_api_strategy("nonsense").name == "wgapi"


class TestBatchCarriesStrategy(_Base):
    def test_default_is_wgapi(self):
        """不传就用 wgapi —— 账号检测原来的行为一行没改。"""
        runner = self._runner()
        assert runner.batch.api_strategy == "wgapi"
        assert runner.snapshot()["api_strategy"] == "wgapi"

    def test_iframe_is_kept(self):
        runner = self._runner("iframe")
        assert runner.batch.api_strategy == "iframe"
        assert runner.snapshot()["api_strategy"] == "iframe"


class TestRunnerHandsStrategyToDouyuLogin(_Base):
    """关键接线：执行器必须把选中的策略**交给 DouyuLogin**。

    这里如果退回写死 `WgapiLoginAPI()`，`captured["strategy"].name` 会是
    "wgapi"，用例变红 —— 是可以反向验证的那一类断言。
    """

    def _patch(self, monkeypatch):
        captured: dict = {}

        class FakeLogin:
            def __init__(self, account, **kwargs):
                captured["strategy"] = kwargs.get("api_strategy")

            def check_account(self):
                from core.douyu.login import LoginResult

                return LoginResult(
                    success=True, message="账号已认证", code="account_verified"
                )

        monkeypatch.setattr(svc, "DouyuLogin", FakeLogin)
        return captured

    def test_uses_iframe_when_selected(self, monkeypatch):
        captured = self._patch(monkeypatch)
        runner = self._runner("iframe")

        runner._run_one(0, runner.accounts[0])

        assert captured["strategy"] is not None
        assert captured["strategy"].name == "iframe"
        assert runner.batch.items[0].status == "account_verified"

    def test_uses_wgapi_by_default(self, monkeypatch):
        captured = self._patch(monkeypatch)
        runner = self._runner()

        runner._run_one(0, runner.accounts[0])

        assert captured["strategy"].name == "wgapi"


class TestRequestSchemaAndRouter(_Base):
    def test_schema_default_is_wgapi(self):
        from web.backend.schemas import AccountCheckBatchRequest

        assert AccountCheckBatchRequest(text="x").api_strategy == "wgapi"
        assert (
            AccountCheckBatchRequest(text="x", api_strategy="iframe").api_strategy
            == "iframe"
        )

    def test_router_passes_strategy(self, monkeypatch):
        """HTTP 层丢字段是最难发现的：批次照跑，只是用错了接口。"""
        from web.backend.routers import account_check as router_mod
        from web.backend.schemas import AccountCheckBatchRequest

        captured: dict = {}

        class _FakeRunner:
            def snapshot(self):
                return {"ok": True}

            def run(self):
                pass

        def fake_create(**kwargs):
            captured.update(kwargs)
            return _FakeRunner()

        class _NoopThread:
            def __init__(self, **kwargs):
                pass

            def start(self):
                pass

        monkeypatch.setattr(router_mod.account_check_registry, "create", fake_create)
        monkeypatch.setattr(router_mod.threading, "Thread", _NoopThread)

        result = router_mod.create_account_check_batch(
            AccountCheckBatchRequest(text="u----p----e----ep", api_strategy="iframe"),
            db=None,
            current=self.user,
        )

        assert result == {"ok": True}
        assert captured["api_strategy"] == "iframe"

    def test_router_defaults_to_wgapi(self, monkeypatch):
        from web.backend.routers import account_check as router_mod
        from web.backend.schemas import AccountCheckBatchRequest

        captured: dict = {}

        class _FakeRunner:
            def snapshot(self):
                return {"ok": True}

            def run(self):
                pass

        def fake_create(**kwargs):
            captured.update(kwargs)
            return _FakeRunner()

        class _NoopThread:
            def __init__(self, **kwargs):
                pass

            def start(self):
                pass

        monkeypatch.setattr(router_mod.account_check_registry, "create", fake_create)
        monkeypatch.setattr(router_mod.threading, "Thread", _NoopThread)

        router_mod.create_account_check_batch(
            AccountCheckBatchRequest(text="u----p----e----ep"),
            db=None,
            current=self.user,
        )

        assert captured["api_strategy"] == "wgapi"