yml2213
178eba1eea
新增 /huya/browser-exchange 页面（精英宝典下方），支持导入账号与 执行兑换，客服仅见已分配账号；账号管理页移除该功能。导入弹窗对齐 斗鱼操作台：服务端分页筛选、可搜索标签、无 CK 不展示、导入本页/ 一键导入全部/导入选中；新增 /accounts/ids 接口与 workbench browser_exchange scope。
124 lines
4.7 KiB
Python
124 lines
4.7 KiB
Python
import pytest
|
|
from sqlalchemy import create_engine, event
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from web.backend.database import Base
|
|
from web.backend.models import HuyaAccount, 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 = create_engine("sqlite://")
|
|
|
|
@event.listens_for(self.engine, "connect")
|
|
def enable_foreign_keys(connection, _):
|
|
connection.execute("PRAGMA foreign_keys=ON")
|
|
|
|
Base.metadata.create_all(self.engine)
|
|
self.session = sessionmaker(bind=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()
|
|
Base.metadata.drop_all(self.engine)
|
|
self.engine.dispose()
|
|
|
|
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_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"
|