live-hub-py/tests/test_douyu_workbench_scopes.py

import pytest

from tests.support.db import get_test_engine, make_test_session
from web.backend.models import (
    Account,
    DouyuTask,
    DouyuWorkbench,
    DouyuWorkbenchAccount,
    User,
)
from web.backend.routers.accounts import delete_account
from web.backend.routers.douyu import (
    list_tasks,
    list_workbench_accounts,
    update_workbench_accounts,
)
from web.backend.schemas import DouyuWorkbenchAccountsUpdate
from web.backend.services.douyu_service import create_douyu_planned_tasks


class TestDouyuWorkbenchScope:
    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.session.add(self.user)
        self.session.commit()
        self.account = Account(
            username="douyu-user",
            password="password",
            email="mail@example.com",
            email_password="mail-password",
            # CK 挂在账号上（2026-10-04 CK 解耦），不再需要造一条登录记录
            cookie="acf_uid=1",
        )
        self.session.add(self.account)
        self.session.commit()

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

    def test_workbench_accounts_are_per_user_and_scope(self):
        update_workbench_accounts(
            DouyuWorkbenchAccountsUpdate(
                handbook_scope="elite", account_ids=[self.account.id]
            ),
            db=self.session,
            current=self.user,
        )
        elite = list_workbench_accounts("elite", db=self.session, current=self.user)
        esports = list_workbench_accounts("esports", db=self.session, current=self.user)
        assert elite["account_ids"] == [self.account.id]
        assert esports["account_ids"] == []
        assert elite["configured"]
        assert not esports["configured"]
        assert self.session.query(DouyuWorkbenchAccount).count() == 1

        update_workbench_accounts(
            DouyuWorkbenchAccountsUpdate(handbook_scope="elite", account_ids=[]),
            db=self.session,
            current=self.user,
        )
        empty_elite = list_workbench_accounts(
            "elite", db=self.session, current=self.user
        )
        assert empty_elite["account_ids"] == []
        assert empty_elite["configured"]
        assert self.session.query(DouyuWorkbench).count() == 1

    def test_task_scope_is_persisted_and_filters_shared_task_types(self):
        elite_batch, _ = create_douyu_planned_tasks(
            self.session,
            [self.account.id],
            "create_gold_qr",
            "elite",
            self.user.id,
        )
        esports_batch, _ = create_douyu_planned_tasks(
            self.session,
            [self.account.id],
            "create_gold_qr",
            "esports",
            self.user.id,
        )
        elite = list_tasks(
            batch_id=None,
            handbook_scope="elite",
            include_detail=False,
            page=None,
            page_size=100,
            db=self.session,
            current=self.user,
        )
        esports = list_tasks(
            batch_id=None,
            handbook_scope="esports",
            include_detail=False,
            page=None,
            page_size=100,
            db=self.session,
            current=self.user,
        )
        assert isinstance(elite, list)
        assert isinstance(esports, list)
        assert {item.batch_id for item in elite} == {elite_batch}
        assert {item.batch_id for item in esports} == {esports_batch}
        assert {
            task.handbook_scope for task in self.session.query(DouyuTask).all()
        } == {"elite", "esports"}

    def test_rejects_task_from_wrong_workbench(self):
        with pytest.raises(ValueError, match="不属于当前工作台"):
            create_douyu_planned_tasks(
                self.session,
                [self.account.id],
                "get_xpd_bind_qr",
                "elite",
                self.user.id,
            )

    def test_elite_workbench_accepts_lock_goods_task(self):
        batch_id, count = create_douyu_planned_tasks(
            self.session,
            [self.account.id],
            "lock_goods",
            "elite",
            self.user.id,
            {"commodity_id": "goods-1"},
        )

        task = (
            self.session.query(DouyuTask).filter(DouyuTask.batch_id == batch_id).one()
        )
        assert count == 1
        assert task.task_type == "lock_goods"
        assert task.result == {"payload": {"commodity_id": "goods-1"}}

        _, pay_count = create_douyu_planned_tasks(
            self.session,
            [self.account.id],
            "pay_locked_order",
            "elite",
            self.user.id,
            {"order_id": "5874", "commodity_id": "goods-1"},
        )
        assert pay_count == 1

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

        delete_account(self.account.id, db=self.session, current=self.user)

        assert self.session.query(DouyuWorkbenchAccount).count() == 0

    def test_workbench_ck_routes_registered(self):
        from web.backend.main import app

        paths = app.openapi()["paths"]
        assert "/api/douyu/accounts/ck-check" in paths
        assert "/api/douyu/accounts/relogin-invalid" in paths