虎牙精英宝典

tests/test_huya_workbench_scopes.py

@@ -0,0 +1,91 @@
+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"

tests/test_migrations.py

@@ -22,7 +22,7 @@ TESTS_DIR = Path(__file__).resolve().parent
 PROJECT_ROOT = TESTS_DIR.parent
 VERSIONS_DIR = PROJECT_ROOT / "web" / "backend" / "migrations" / "versions"
 
-HEAD_REVISION = "20260830_0031"
+HEAD_REVISION = "20260831_0032"
 
 # 迁移新增、但不声明在模型里的复合查询索引。
 EXTRA_INDEXES = (