live-hub-py/tests/test_cookie_operations.py

from types import SimpleNamespace
from typing import cast
from unittest.mock import patch

import pytest

from tests.support.db import get_test_engine, make_test_session
from web.backend.models import Account, LoginTask, ProxyConfig, User
from web.backend.routers.cookies import (
    check_cookie_operations,
    get_cookie,
    list_cookie_operation_tags,
    list_cookie_operations,
    list_cookies,
    relogin_invalid_cookie_operations,
)
from web.backend.services.cookie_check_service import check_douyu_cookie
from web.backend.services.login_service import (
    LoginBatchRunner,
    cleanup_orphan_login_tasks,
    cleanup_orphan_relogin_tasks,
)


class TestCookieOperation:
    def setup_method(self):
        self.engine = get_test_engine()
        self.session = make_test_session(self.engine)

        self.support = User(username="support", password_hash="hash", role="support")
        self.other_support = User(
            username="other", password_hash="hash", role="support"
        )
        self.session.add_all([self.support, self.other_support])
        self.session.commit()

        owned = Account(
            username="owned-account",
            password="owned-password",
            email="owned@example.com",
            email_password="mail-password",
            assigned_to=self.support.id,
        )
        other = Account(
            username="other-account",
            password="other-password",
            email="other@example.com",
            email_password="mail-password",
            assigned_to=self.other_support.id,
        )
        self.session.add_all([owned, other])
        self.session.commit()
        self.session.add_all(
            [
                LoginTask(
                    batch_id="owned",
                    account_id=owned.id,
                    created_by=self.support.id,
                    status="success",
                    cookie="owned-secret",
                ),
                LoginTask(
                    batch_id="other",
                    account_id=other.id,
                    created_by=self.other_support.id,
                    status="success",
                    cookie="other-secret",
                ),
            ]
        )
        self.session.commit()
        self.owned_task = (
            self.session.query(LoginTask).filter(LoginTask.batch_id == "owned").one()
        )
        self.other_task = (
            self.session.query(LoginTask).filter(LoginTask.batch_id == "other").one()
        )

    def teardown_method(self):
        self.session.close()

    def test_support_operation_list_is_scoped_and_never_returns_credentials(self):
        result = list_cookie_operations(
            search="",
            tag="",
            page=1,
            page_size=20,
            db=self.session,
            current=self.support,
        )

        assert result["total"] == 1
        assert result["items"] == [
            {
                "id": self.owned_task.id,
                "account_username": "owned-account",
                "tag": "",
                "ck_check_status": "",
                "ck_checked_at": None,
                "created_at": None,
                "relogin_status": "",
                "relogin_message": "",
                "relogin_batch_id": "",
            }
        ]
        assert "owned-secret" not in str(result)
        assert "owned-password" not in str(result)

    def test_support_cannot_use_cookie_management_endpoints(self):
        with pytest.raises(Exception, match="cookie:view"):
            list_cookies(db=self.session, current=self.support)
        with pytest.raises(Exception, match="cookie:view"):
            get_cookie(self.owned_task.id, db=self.session, current=self.support)

    def test_cookie_list_filters_by_account_tag(self):
        admin = User(username="admin", password_hash="hash", role="super_admin")
        self.session.add(admin)
        self.session.query(Account).filter(
            Account.id == self.owned_task.account_id
        ).update({"tag": "A组"})
        self.session.query(Account).filter(
            Account.id == self.other_task.account_id
        ).update({"tag": "B组"})
        self.session.commit()

        result = list_cookies(
            search="",
            tag="A组",
            account_names="",
            page=1,
            page_size=20,
            include_cookie=False,
            db=self.session,
            current=admin,
        )

        assert isinstance(result, dict)
        assert result["total"] == 1
        assert [item["id"] for item in result["items"]] == [self.owned_task.id]

    def test_support_operation_tag_filter_and_list_stay_scoped(self):
        self.session.query(Account).filter(
            Account.id == self.owned_task.account_id
        ).update({"tag": "我的标签"})
        self.session.query(Account).filter(
            Account.id == self.other_task.account_id
        ).update({"tag": "别人的标签"})
        self.session.commit()

        tags = list_cookie_operation_tags(db=self.session, current=self.support)
        result = list_cookie_operations(
            search="",
            tag="别人的标签",
            page=1,
            page_size=20,
            db=self.session,
            current=self.support,
        )

        assert tags == ["我的标签"]
        assert result["total"] == 0

    @patch("web.backend.routers.cookies._check_one_cookie")
    def test_support_can_check_only_assigned_cookie(self, check_one_cookie):
        check_one_cookie.return_value = {
            "id": self.owned_task.id,
            "valid": True,
            "message": "有效",
            "fish_ball": 9,
            "nickname": "tester",
            "level": 1,
            "checked_at": "2026-01-01T00:00:00+00:00",
        }

        result = check_cookie_operations(
            ids=f"{self.owned_task.id},{self.other_task.id}",
            db=self.session,
            current=self.support,
        )

        assert [item["id"] for item in result["results"]] == [self.owned_task.id]
        self.session.refresh(self.owned_task)
        self.session.refresh(self.other_task)
        assert self.owned_task.ck_check_status == "valid"
        assert self.other_task.ck_check_status == ""

    def test_operation_permission_does_not_allow_unrelated_users(self):
        no_permission_user = SimpleNamespace(
            id=999,
            username="no-permission",
            role="support",
            custom_permissions=[],
        )
        with pytest.raises(Exception, match="cookie:operate"):
            list_cookie_operations(
                search="",
                tag="",
                page=1,
                page_size=20,
                db=self.session,
                current=cast(User, no_permission_user),
            )

    def test_relogin_failed_record_stays_visible_and_can_be_checked(self):
        self.owned_task.attempt_status = "failed"
        self.owned_task.attempt_message = "重新登录失败: 密码错误（旧 Cookie 已保留）"
        self.session.commit()

        result = list_cookie_operations(
            search="",
            tag="",
            page=1,
            page_size=20,
            db=self.session,
            current=self.support,
        )

        assert result["total"] == 1
        assert result["items"][0]["relogin_status"] == "relogin_failed"
        assert "旧 Cookie 已保留" in result["items"][0]["relogin_message"]
        # 失败的重登尝试不影响记录持有的 CK
        assert self.owned_task.cookie == "owned-secret"
        assert self.owned_task.status == "success"

    def test_service_restart_cleans_orphan_relogin_tasks(self):
        self.owned_task.attempt_status = "running"
        self.owned_task.attempt_message = "正在重新登录，旧 Cookie 保留中"
        self.session.commit()

        cleaned = cleanup_orphan_relogin_tasks(self.session)

        assert cleaned == 1
        self.session.refresh(self.owned_task)
        self.session.refresh(self.other_task)
        # 只收敛尝试状态，已有 CK 与成功状态原封不动
        assert self.owned_task.attempt_status == "failed"
        assert "服务重启" in self.owned_task.attempt_message
        assert self.owned_task.status == "success"
        assert self.owned_task.cookie == "owned-secret"
        assert self.other_task.attempt_status == ""
        assert self.other_task.status == "success"

    def test_service_restart_cleans_orphan_login_tasks(self):
        """普通登录批次遗留的在途尝试必须收敛，且保留旧 CK。"""
        self.owned_task.attempt_status = "pending"
        self.owned_task.cookie = "acf_uid=owned"
        self.session.commit()

        cleaned = cleanup_orphan_login_tasks(self.session)

        assert cleaned == 1
        self.session.refresh(self.owned_task)
        self.session.refresh(self.other_task)
        assert self.owned_task.attempt_status == "failed"
        assert self.owned_task.cookie == "acf_uid=owned"
        assert self.owned_task.status == "success"
        assert "服务重启" in self.owned_task.attempt_message
        assert self.other_task.status == "success"

    def test_service_restart_cleans_legacy_in_flight_status(self):
        """升级前直接把状态写在 status 上的在途记录也要能收敛。"""
        self.owned_task.status = "relogin_running"
        self.owned_task.attempt_status = ""
        self.session.commit()

        cleaned = cleanup_orphan_login_tasks(self.session)

        assert cleaned == 1
        self.session.refresh(self.owned_task)
        assert self.owned_task.attempt_status == "failed"
        assert self.owned_task.cookie == "owned-secret"

    def test_runner_copies_proxy_config_before_background_execution(self):
        proxy = ProxyConfig(
            enabled=True,
            http="http://127.0.0.1:8080",
            https="",
            api_url="",
            whitelist_enabled=False,
            whitelist_platform=None,
            whitelist_credentials=None,
            whitelist_uid="",
            whitelist_ukey="",
        )
        runner = LoginBatchRunner(
            db=self.session,
            account_ids=[],
            created_by=self.support.id,
            creator_permissions=[],
            proxy_config=proxy,
        )
        proxy.http = "http://changed.example:8080"

        proxy_dict, _ = runner._resolve_static_proxy()

        assert proxy_dict == {
            "http": "http://127.0.0.1:8080",
            "https": "http://127.0.0.1:8080",
        }

    @patch("web.backend.services.cookie_check_service.requests.get")
    def test_cookie_check_uses_new_cookie_result(self, mock_get):
        mock_get.side_effect = [
            SimpleNamespace(json=lambda: {"error": 0, "data": {"count": 9}}),
            SimpleNamespace(
                json=lambda: {"error": 0, "data": {"nn": "new-name", "lv": 12}}
            ),
        ]

        result = check_douyu_cookie("new-cookie")

        assert result["valid"]
        assert result["fish_ball"] == 9
        assert result["nickname"] == "new-name"
        assert result["level"] == 12

    @patch("web.backend.routers.cookies._start_relogin_tasks")
    def test_relogin_invalid_only_targets_support_visible_accounts(self, start_relogin):
        self.owned_task.ck_check_status = "invalid"
        self.other_task.ck_check_status = "invalid"
        self.session.commit()
        start_relogin.return_value = {
            "batch_id": "batch",
            "count": 1,
            "skipped": 0,
            "success": True,
        }

        result = relogin_invalid_cookie_operations(
            search="",
            tag="",
            db=self.session,
            current=self.support,
        )

        assert result["success"]
        selected_tasks = start_relogin.call_args.args[0]
        assert [task.id for task in selected_tasks] == [self.owned_task.id]