live-hub-py/tests/test_pool_account_visibility.py

"""池化后的账号可见性与操作流水。

两条都是**新增**的语义，且直接决定客服能不能干活：
1. 客服租用的账号要能看到（否则刚做完的单查不了）；
2. 流水要能看出「谁做的」（71% 的历史事件没有操作人，这是要补的）。
"""

from __future__ import annotations

from datetime import UTC, datetime

import pytest
from fastapi import HTTPException

from tests.support.db import get_test_engine, make_test_session
from web.backend.models import Account, AccountEvent, User
from web.backend.routers.accounts import account_events, list_accounts
from web.backend.services import account_pool as pool

NOW = datetime(2026, 10, 7, 15, 0, 0, tzinfo=UTC)


class _Base:
    def setup_method(self):
        self.engine = get_test_engine()
        self.session = make_test_session(self.engine)
        self.operator = User(username="vis-op", password_hash="h", role="operation")
        self.support = User(username="vis-kf", password_hash="h", role="support")
        self.session.add_all([self.operator, self.support])
        self.session.flush()
        self.operator_id = self.operator.id
        self.support_id = self.support.id

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

    def make_account(self, name: str, *, assigned_to=None) -> Account:
        account = Account(
            username=name,
            password="p",
            email="e@e.com",
            email_password="ep",
            status="active",
            ck_check_status="valid",
            cookie="PHPSESSID=x",
            elite_opened=True,
            points=5000,
            assigned_to=assigned_to,
        )
        self.session.add(account)
        self.session.commit()
        return account

    def lease(self, account: Account, holder_id: int):
        return pool.lease_accounts(
            self.session,
            holder_user_id=holder_id,
            candidates=[
                pool.make_candidate(
                    account_id=account.id,
                    username=account.username,
                    points=5000,
                    score=1280,
                )
            ],
            commodity_id="20260120QYOOB_bag10",
        )[0]


class TestVisibleAccountsWithLease(_Base):
    def _visible_names(self, user) -> set[str]:
        rows = list_accounts(
            db=self.session,
            current=user,
            page=None,
            page_size=100,
            search="",
            tag=None,
            status="",
            ck_status="",
            bind_status="",
            has_cookie=False,
            assign_status="",
            assigned_user_id=None,
            high_value_filter="",
            min_points=None,
            max_points=None,
            min_gold=None,
            max_gold=None,
            elite_opened=None,
            esports_opened=None,
            role_bind_status="",
            sort=None,
            order=None,
            include_sensitive=False,
        )
        items = rows["items"] if isinstance(rows, dict) else rows
        # 走 `model_dump()`：只看属性会漏掉「模型上声明了、序列化里没列」这类
        # Pydantic 静默丢字段（本仓库已经踩过三次）。
        return {item.model_dump()["username"] for item in items}

    def test_leased_account_is_visible_to_holder(self):
        account = self.make_account("leased-vis", assigned_to=self.operator_id)
        self.lease(account, self.support_id)
        assert "leased-vis" in self._visible_names(self.support)

    def test_unleased_operator_account_is_not_visible(self):
        self.make_account("hidden-vis", assigned_to=self.operator_id)
        assert "hidden-vis" not in self._visible_names(self.support)

    def test_visibility_ends_after_release(self):
        account = self.make_account("temp-vis", assigned_to=self.operator_id)
        lease = self.lease(account, self.support_id)
        pool.release_lease(self.session, lease, reason="done")
        assert "temp-vis" not in self._visible_names(self.support)


class TestAccountEventActor(_Base):
    def _add_event(self, account_id: int, *, actor_id=None, actor_type=""):
        self.session.add(
            AccountEvent(
                account_id=account_id,
                event_type="recharge",
                category="business",
                occurred_at=NOW,
                title="充值 48",
                amount=48,
                subject_type="recharge",
                detail={"channel": "wechat_qr"},
                source="douyu_task",
                actor_type=actor_type,
                actor_id=actor_id,
            )
        )
        self.session.commit()

    def test_events_return_actor_name(self):
        account = self.make_account("ev-1", assigned_to=self.support_id)
        self._add_event(account.id, actor_id=self.support_id, actor_type="user")

        result = account_events(
            account.id,
            page=1,
            page_size=50,
            event_type="",
            category="",
            actor_id=None,
            db=self.session,
            current=self.support,
        )
        assert result["items"][0]["actor_name"] == "vis-kf"
        # 序列化结果里必须真有这两个字段（Pydantic/dict 静默丢字段在本仓库踩过）
        assert result["items"][0]["actor_id"] == self.support_id

    def test_filter_by_actor_id(self):
        account = self.make_account("ev-2", assigned_to=self.operator_id)
        self._add_event(account.id, actor_id=self.support_id, actor_type="user")
        self._add_event(account.id, actor_id=self.operator_id, actor_type="user")

        mine = account_events(
            account.id,
            page=1,
            page_size=50,
            event_type="",
            category="",
            actor_id=self.support_id,
            db=self.session,
            current=self.operator,
        )
        assert mine["total"] == 1
        assert mine["items"][0]["actor_id"] == self.support_id

    def test_actorless_event_still_returned(self):
        """历史事件没有操作人（71%）—— 不能因此被过滤掉或报错。"""
        account = self.make_account("ev-3", assigned_to=self.support_id)
        self._add_event(account.id)
        result = account_events(
            account.id,
            page=1,
            page_size=50,
            event_type="",
            category="",
            actor_id=None,
            db=self.session,
            current=self.support,
        )
        assert result["total"] == 1
        assert result["items"][0]["actor_name"] == ""

    def test_timeline_of_leased_account_is_readable(self):
        account = self.make_account("ev-4", assigned_to=self.operator_id)
        self._add_event(account.id, actor_id=self.support_id, actor_type="user")
        # 没有租约时客服看不到这个账号
        with pytest.raises(HTTPException):
            account_events(
                account.id,
                page=1,
                page_size=50,
                event_type="",
                category="",
                actor_id=None,
                db=self.session,
                current=self.support,
            )
        # 租下来就能看（这正是池化后客服复盘自己那单的路径）
        self.lease(account, self.support_id)
        result = account_events(
            account.id,
            page=1,
            page_size=50,
            event_type="",
            category="",
            actor_id=None,
            db=self.session,
            current=self.support,
        )
        assert result["total"] == 1