108 lines
4.8 KiB
Python
108 lines
4.8 KiB
Python
import os
|
|
import unittest
|
|
from types import SimpleNamespace
|
|
|
|
os.environ.setdefault("DATABASE_URL", "sqlite://")
|
|
os.environ.setdefault("APP_ENCRYPTION_KEY", "MDEyMzQ1Njc4OWFiY2RlZjAxMjM0NTY3ODlhYmNkZWY=")
|
|
|
|
from sqlalchemy import create_engine, event
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from web.backend.database import Base
|
|
from web.backend.models import Account, DouyuTask, DouyuWorkbench, DouyuWorkbenchAccount, LoginTask, 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 DouyuWorkbenchScopeTests(unittest.TestCase):
|
|
def setUp(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.session.add(self.user)
|
|
self.session.commit()
|
|
self.account = Account(username="douyu-user", password="password", email="mail@example.com", email_password="mail-password")
|
|
self.session.add(self.account)
|
|
self.session.commit()
|
|
self.session.add(LoginTask(batch_id="login", account_id=self.account.id, created_by=self.user.id, status="success", cookie="acf_uid=1"))
|
|
self.session.commit()
|
|
self.current = SimpleNamespace(id=self.user.id, username=self.user.username, role=self.user.role, custom_permissions=None)
|
|
|
|
def tearDown(self):
|
|
self.session.close()
|
|
Base.metadata.drop_all(self.engine)
|
|
self.engine.dispose()
|
|
|
|
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.current,
|
|
)
|
|
elite = list_workbench_accounts("elite", db=self.session, current=self.current)
|
|
esports = list_workbench_accounts("esports", db=self.session, current=self.current)
|
|
self.assertEqual(elite["account_ids"], [self.account.id])
|
|
self.assertEqual(esports["account_ids"], [])
|
|
self.assertTrue(elite["configured"])
|
|
self.assertFalse(esports["configured"])
|
|
self.assertEqual(self.session.query(DouyuWorkbenchAccount).count(), 1)
|
|
|
|
update_workbench_accounts(
|
|
DouyuWorkbenchAccountsUpdate(handbook_scope="elite", account_ids=[]),
|
|
db=self.session,
|
|
current=self.current,
|
|
)
|
|
empty_elite = list_workbench_accounts("elite", db=self.session, current=self.current)
|
|
self.assertEqual(empty_elite["account_ids"], [])
|
|
self.assertTrue(empty_elite["configured"])
|
|
self.assertEqual(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.current,
|
|
)
|
|
esports = list_tasks(
|
|
batch_id=None, handbook_scope="esports", include_detail=False, page=None, page_size=100,
|
|
db=self.session, current=self.current,
|
|
)
|
|
self.assertEqual({item.batch_id for item in elite}, {elite_batch})
|
|
self.assertEqual({item.batch_id for item in esports}, {esports_batch})
|
|
self.assertEqual({task.handbook_scope for task in self.session.query(DouyuTask).all()}, {"elite", "esports"})
|
|
|
|
def test_rejects_task_from_wrong_workbench(self):
|
|
with self.assertRaisesRegex(ValueError, "不属于当前工作台"):
|
|
create_douyu_planned_tasks(
|
|
self.session, [self.account.id], "get_xpd_bind_qr", "elite", self.user.id,
|
|
)
|
|
|
|
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.current)
|
|
|
|
self.assertEqual(self.session.query(DouyuWorkbenchAccount).count(), 0)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|