功能: 按用户同步斗鱼工作台

2026-08-13 15:47:07 +08:00
parent 090698e6b2
commit cf3fe68de3
14 changed files with 503 additions and 44 deletions
@@ -0,0 +1,107 @@
+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()