live-hub-py/tests/test_huya_workbench_scopes.py

import pytest

from tests.support.db import get_test_engine, make_test_session
from web.backend.models import (
    HuyaAccount,
    HuyaTask,
    HuyaWorkbench,
    HuyaWorkbenchAccount,
    User,
)
from web.backend.routers.huya import (
    delete_account,
    list_huya_account_ids,
    list_huya_workbench_accounts,
    list_tasks,
    update_huya_workbench_accounts,
)
from web.backend.schemas import HuyaTaskBatchRequest, HuyaWorkbenchAccountsUpdate
from web.backend.services.huya_service import create_planned_tasks


class TestHuyaWorkbenchScope:
    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="100", yyuid="100", cookie="yyuid=100; udb_uid=100"
        )
        self.session.add_all([self.user, self.account])
        self.session.commit()

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

    def test_accounts_and_tasks_are_scoped(self):
        update_huya_workbench_accounts(
            HuyaWorkbenchAccountsUpdate(
                handbook_scope="elite", account_ids=[self.account.id]
            ),
            db=self.session,
            current=self.user,
        )
        result = list_huya_workbench_accounts(
            "elite", db=self.session, current=self.user
        )
        assert result == {"account_ids": [self.account.id], "configured": True}
        assert self.session.query(HuyaWorkbench).count() == 1

        elite_batch, count = create_planned_tasks(
            self.session,
            [self.account.id],
            "query_act_tasks",
            self.user.id,
            handbook_scope="elite",
        )
        assert count == 1
        tasks = list_tasks(
            handbook_scope="elite",
            include_images=False,
            db=self.session,
            current=self.user,
        )
        assert [task.batch_id for task in tasks] == [elite_batch]
        assert tasks[0].handbook_scope == "elite"

        with pytest.raises(ValueError, match="无效的工作台"):
            create_planned_tasks(
                self.session,
                [self.account.id],
                "query_points",
                self.user.id,
                handbook_scope="invalid",
            )

    def test_browser_exchange_scope_and_account_ids(self):
        no_cookie = HuyaAccount(uid="200", yyuid="200", cookie="")
        self.session.add(no_cookie)
        self.session.commit()

        update_huya_workbench_accounts(
            HuyaWorkbenchAccountsUpdate(
                handbook_scope="browser_exchange", account_ids=[self.account.id]
            ),
            db=self.session,
            current=self.user,
        )
        assert list_huya_workbench_accounts(
            "browser_exchange", db=self.session, current=self.user
        ) == {
            "account_ids": [self.account.id],
            "configured": True,
        }
        # browser_exchange 与 elite 两个 scope 的账号集合互不影响
        assert list_huya_workbench_accounts(
            "elite", db=self.session, current=self.user
        ) == {
            "account_ids": [],
            "configured": False,
        }

        # ids 接口：has_cookie=True 只返回有 Cookie 的账号
        result = list_huya_account_ids(
            has_cookie=True, db=self.session, current=self.user
        )
        assert result == {"account_ids": [self.account.id], "total": 1}
        all_ids = list_huya_account_ids(db=self.session, current=self.user)
        assert sorted(all_ids["account_ids"]) == sorted([self.account.id, no_cookie.id])
        assert all_ids["total"] == 2

        # ids 接口：search 匹配 uid
        searched = list_huya_account_ids(
            search="200", db=self.session, current=self.user
        )
        assert searched["account_ids"] == [no_cookie.id]

    def test_recharge_order_tasks_created_for_every_account(self):
        """多选账号生成支付码：每个账号一个任务（一个账号一张码，各自监听支付）。

        商品快照类任务（refresh_*）仍然是全局的，多选也只建一个。
        """
        other = HuyaAccount(uid="300", yyuid="300", cookie="yyuid=300; udb_uid=300")
        self.session.add(other)
        self.session.commit()
        account_ids = [self.account.id, other.id]

        batch_id, count = create_planned_tasks(
            self.session,
            account_ids,
            "create_recharge_order",
            self.user.id,
            payload={"spu_id": "hy-8328015"},
            handbook_scope="elite",
        )

        assert count == 2
        tasks = self.session.query(HuyaTask).filter(HuyaTask.batch_id == batch_id).all()
        assert len(tasks) == 2
        # 每个选中的账号都拿到一个码，而不是只有第一个
        assert sorted(task.account_id for task in tasks) == sorted(account_ids)
        assert {task.task_type for task in tasks} == {"create_recharge_order"}

        snapshot_batch, snapshot_count = create_planned_tasks(
            self.session,
            account_ids,
            "refresh_recharge_goods",
            self.user.id,
            handbook_scope="elite",
        )
        assert snapshot_count == 1
        assert (
            self.session.query(HuyaTask)
            .filter(HuyaTask.batch_id == snapshot_batch)
            .count()
            == 1
        )

    def test_account_delete_cleans_workbench_membership(self):
        self.session.add(
            HuyaWorkbenchAccount(
                user_id=self.user.id,
                handbook_scope="elite",
                account_id=self.account.id,
            )
        )
        self.session.commit()

        result = delete_account(self.account.id, db=self.session, current=self.user)
        assert result["success"] is True
        assert self.session.query(HuyaAccount).count() == 0
        assert self.session.query(HuyaWorkbenchAccount).count() == 0


def test_huya_batch_schema_defaults_to_legacy():
    request = HuyaTaskBatchRequest(account_ids=[1], task_type="query_points")
    assert request.handbook_scope == "legacy"