92 lines
3.2 KiB
Python
92 lines
3.2 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_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_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"
|