feat(douyu): 精英宝典工作台支持 CK 检测与失效重登

tests/test_douyu_cookie_visibility.py

@@ -10,12 +10,22 @@
 「根本不看记录」。
 """
 
+from datetime import UTC, datetime
+from unittest.mock import patch
+
 import pytest
+from fastapi import HTTPException
 
 from tests.support.db import get_test_engine, make_test_session
 from web.backend.models import Account, LoginTask, User
 from web.backend.routers.cookies import list_cookies
-from web.backend.routers.douyu import list_task_account_ids, list_task_accounts
+from web.backend.routers.douyu import (
+    check_task_accounts_ck,
+    list_task_account_ids,
+    list_task_accounts,
+    relogin_invalid_task_accounts,
+)
+from web.backend.schemas import CookieReloginRequest
 from web.backend.services.account_service import cookie_account_ids_query
 from web.backend.services.douyu_service import (
     create_douyu_planned_tasks,
@@ -175,3 +185,87 @@ class TestDouyuCookieVisibility:
         assert fresh.cookie == "acf_uid=481802039"
         assert fresh.ck_check_status == "valid"
         assert self._task_account_names(self.admin) == {"远行者103", "排队中账号"}
+
+    def test_task_account_payload_contains_ck_status(self):
+        rows = list_task_accounts(
+            search="",
+            tag="",
+            ids="",
+            account_names="",
+            page=None,
+            page_size=50,
+            db=self.session,
+            current=self.admin,
+        )
+        row = next(item for item in rows if item.id == self.relogin_account.id)
+        assert row.ck_check_status == "valid"
+        assert row.ck_checked_at is None
+
+    def test_workbench_ck_check_marks_invalid(self):
+        checked_at = datetime.now(UTC)
+        with patch(
+            "web.backend.routers.cookies.check_douyu_cookie",
+            return_value={
+                "valid": False,
+                "message": "ck expired",
+                "fish_ball": None,
+                "nickname": None,
+                "level": None,
+                "checked_at": checked_at,
+            },
+        ):
+            result = check_task_accounts_ck(
+                ids=str(self.relogin_account.id),
+                db=self.session,
+                current=self.admin,
+            )
+
+        assert result["success"] is True
+        assert result["results"] == [
+            {
+                "id": self.relogin_account.id,
+                "valid": False,
+                "checked_at": checked_at.isoformat(),
+            }
+        ]
+        self.session.refresh(self.relogin_account)
+        assert self.relogin_account.ck_check_status == "invalid"
+
+    def test_workbench_relogin_invalid_is_admin_only_and_filters_invalid(self):
+        self.relogin_account.ck_check_status = "invalid"
+        self.session.commit()
+        captured: dict[str, object] = {}
+
+        def fake_start(accounts, db, current, *, action=""):
+            captured["ids"] = [account.id for account in accounts]
+            captured["action"] = action
+            return {
+                "batch_id": "workbench-relogin",
+                "count": len(accounts),
+                "skipped": 0,
+                "success": True,
+            }
+
+        with patch(
+            "web.backend.routers.cookies._start_relogin_accounts",
+            side_effect=fake_start,
+        ):
+            result = relogin_invalid_task_accounts(
+                CookieReloginRequest(
+                    ids=[self.relogin_account.id, self.pending_account.id]
+                ),
+                db=self.session,
+                current=self.admin,
+            )
+
+        assert result["count"] == 1
+        assert captured["ids"] == [self.relogin_account.id]
+        assert captured["action"] == "douyu:task:relogin_invalid"
+
+        with pytest.raises(HTTPException) as excinfo:
+            relogin_invalid_task_accounts(
+                CookieReloginRequest(ids=[self.relogin_account.id]),
+                db=self.session,
+                current=self.support,
+            )
+        assert excinfo.value.status_code == 403

tests/test_douyu_workbench_scopes.py

@@ -158,3 +158,10 @@ class TestDouyuWorkbenchScope:
         delete_account(self.account.id, db=self.session, current=self.user)
 
         assert self.session.query(DouyuWorkbenchAccount).count() == 0
+
+    def test_workbench_ck_routes_registered(self):
+        from web.backend.main import app
+
+        paths = app.openapi()["paths"]
+        assert "/api/douyu/accounts/ck-check" in paths
+        assert "/api/douyu/accounts/relogin-invalid" in paths