Files
live-hub-py/tests/test_douyu_pool_api.py
T
yml2213 18bc7be4b1 feat(pool): 账号自动分配与回收(租约 + 补给链 + 独立工作台)
在既有「管理员分配 → 客服导入工作台 → 手动选号」之外**新增**一条
「客服选皮肤 + 数量 → 系统分配账号 + 按需补积分 → 用完自动收回」的路径。
旧路径只做并集与可选参数扩展,行为不变(有用例守着)。

- 归属与使用权分离:新增 account_leases(生成列 + 唯一索引,并发抢号由数据库裁决),
  assigned_to 收窄为管理员/运营的维护责任;客服可见范围扩为「名下 ∪ 租用中」
- account_pool 服务:候选筛选、两梯队排序(先花快过期的 / 先选充值少的)、
  供给与成本预估、租约领取/释放/回收
- 补给链:查余额 → 充鱼翅(单次 ≤800 元,超限拆单)→ 赠精英令 → 兑换;
  运行期两道闸门:_skip_when(余额已够就别再花钱)、_depends_on(前置未成功就跳过)
- 转人工:鱼翅支付等 10 分钟未到账转 manual_required,租约先不还;
  account_lease_reaper 兜 TTL 到期 / 漏释放 / 孤儿批次
- 运营流水:业务事件补 actor_id = task.created_by;账号详情页展示操作人、渠道
  与租赁记录
- 新工作台:/douyu/pool + 菜单「账号分配台」(新建,不改造旧工作台)
- 修复 D1:官方补录的兑换事件漏写 subject_key(10,612 条主键为空),
  会让 365 天同皮肤规则漏判 87% 的历史;补代码 + 回填脚本
- 迁移 20261007_0052;权限 account:pool_use / account:pool_admin

测试:766 passed, 6 skipped(改动前基线 727);ruff 与 tsc -b 干净。
设计稿与实施记录:docs/设计-账号自动分配与回收.md
2026-10-07 12:29:08 +08:00

343 lines
13 KiB
Python

"""账号池接口:供给预览、分配下单、租约释放。
**这是新增的一条路**,所以用例同时守住「新增的能用」与「旧的没被动」:
- 池路径要能领到号、建出任务、并让租用人看得到自己的任务;
- `create_douyu_planned_tasks` 不传新参数时行为必须与以前一模一样。
"""
from __future__ import annotations
import asyncio
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,
AccountLease,
DouyuGoodsSnapshot,
DouyuTask,
User,
)
from web.backend.routers import douyu_pool as pool_router
from web.backend.schemas import DouyuPoolExchangeRequest
from web.backend.services import account_pool as pool
COMMODITY = "20260120QYOOB_bag10"
NOW = datetime(2026, 10, 7, 12, 0, 0, tzinfo=UTC)
class _ApiBase:
def setup_method(self, monkeypatch=None):
self.engine = get_test_engine()
self.session = make_test_session(self.engine)
self.admin = User(username="api-admin", password_hash="h", role="super_admin")
self.support = User(username="api-kf", password_hash="h", role="support")
self.other = User(username="api-kf2", password_hash="h", role="support")
self.session.add_all([self.admin, self.support, self.other])
self.session.flush()
self.admin_id = self.admin.id
self.support_id = self.support.id
self.other_id = self.other.id
self.session.add(
DouyuGoodsSnapshot(
commodity_id=COMMODITY,
name="套装-兔兔白日梦",
score=1280,
)
)
self.session.commit()
def teardown_method(self):
self.session.close()
def make_account(self, name, *, points=5000, opened=True):
account = Account(
username=name,
password="p",
email="e@e.com",
email_password="ep",
status="active",
ck_check_status="valid",
cookie="PHPSESSID=x",
elite_opened=opened,
points=points,
)
self.session.add(account)
self.session.commit()
return account
class TestSupply(_ApiBase):
def test_counts_and_channel(self):
self.make_account("rich", points=5000)
self.make_account("poor", points=0)
self.make_account("closed", opened=False, points=5000)
supply = pool_router.pool_supply(
scope="elite", commodity_id=COMMODITY, db=self.session, current=self.support
)
assert supply["zero_topup"] == 1
assert supply["after_topup"] == 2
assert supply["blocked"]["elite_not_opened"] == 1
assert supply["commodity"]["name"] == "套装-兔兔白日梦"
assert supply["commodity"]["score"] == 1280
# 生产当前是微信扫码 → 前端要提前告诉客服"这单得你去扫"
assert supply["recharge_channel"] == "wechat_qr"
assert supply["manual_payment_required"] is True
def test_missing_goods_is_404(self):
with pytest.raises(HTTPException) as exc:
pool_router.pool_supply(
scope="elite",
commodity_id="nope",
db=self.session,
current=self.support,
)
assert exc.value.status_code == 404
def test_unsupported_scope_is_400(self):
with pytest.raises(HTTPException) as exc:
pool_router.pool_supply(
scope="peace",
commodity_id=COMMODITY,
db=self.session,
current=self.support,
)
assert exc.value.status_code == 400
class TestExchange(_ApiBase):
def _patch_runner(self, monkeypatch, captured: list):
def fake_start(**kwargs):
captured.append(kwargs)
monkeypatch.setattr(pool_router, "start_douyu_batch", fake_start)
def _exchange(self, monkeypatch, **overrides):
captured: list = []
self._patch_runner(monkeypatch, captured)
payload = {
"scope": "elite",
"commodity_id": COMMODITY,
"quantity": 1,
}
payload.update(overrides)
result = asyncio.run(
pool_router.pool_exchange(
DouyuPoolExchangeRequest(**payload),
db=self.session,
current=self.support,
)
)
return result, captured
def test_allocates_creates_tasks_and_binds_batch(self, monkeypatch):
account = self.make_account("a1", points=5000)
result, captured = self._exchange(monkeypatch)
assert result.allocated == 1 and result.shortfall == 0
assert [a.account_id for a in result.accounts] == [account.id]
assert captured and captured[0]["batch_id"] == result.batch_id
lease = self.session.query(AccountLease).one()
assert lease.status == "active"
assert lease.batch_id == result.batch_id
assert lease.commodity_id == COMMODITY
task = self.session.query(DouyuTask).one()
assert task.task_type == "exchange_goods"
assert task.handbook_scope == "elite"
assert task.batch_id == result.batch_id
assert task.result["payload"]["commodity_id"] == COMMODITY
assert task.result["payload"]["num"] == 1
def test_task_is_visible_to_holder_not_to_account_owner(self, monkeypatch):
"""池化后账号归属是运营;任务必须写成**租用人**,否则客服看不到自己的单。"""
self.make_account("a2", points=5000)
result, _ = self._exchange(monkeypatch)
task = self.session.query(DouyuTask).one()
assert task.account_assigned_to == self.support_id
assert task.account_assigned_to != self.admin_id
assert result.batch_id
def test_shortfall_when_not_enough_accounts(self, monkeypatch):
self.make_account("only", points=5000)
result, _ = self._exchange(monkeypatch, quantity=3)
assert result.allocated == 1
assert result.shortfall == 2
assert any("可用账号不足" in reason for reason in result.reasons)
def test_allow_topup_false_excludes_tier_b(self, monkeypatch):
self.make_account("poor", points=0)
result, captured = self._exchange(monkeypatch, allow_topup=False)
assert result.allocated == 0
assert captured == [], "没领到号就不该启动批次"
assert any("未允许充值" in reason for reason in result.reasons)
def test_tier_b_reports_recharge_plan(self, monkeypatch):
self.make_account("poor", points=300)
result, _ = self._exchange(monkeypatch)
account = result.accounts[0]
assert account.tier == "B"
assert account.points == 300
assert account.topup_gifts == 98 # (1280-300)/10
assert account.topup_yuan == 98
assert account.recharge_orders == [98]
def test_tier_b_split_orders_above_800(self, monkeypatch):
"""单次 800 元上限:缺口超过时拆单,而不是被拒。"""
self.session.add(
DouyuGoodsSnapshot(
commodity_id="20260120QYOOB_big", name="仙剑奇侠传大礼包", score=9000
)
)
self.session.commit()
self.make_account("poor2", points=0)
result, _ = self._exchange(monkeypatch, commodity_id="20260120QYOOB_big")
account = result.accounts[0]
assert account.topup_yuan == 900
assert account.recharge_orders == [800, 100]
def test_tier_b_creates_full_supply_chain(self, monkeypatch):
"""B 梯队要排出完整补给链,且每一步都指向前一步(运行期的 `_depends_on`)。"""
self.make_account("chain", points=300)
result, _ = self._exchange(monkeypatch)
assert result.allocated == 1
tasks = self.session.query(DouyuTask).order_by(DouyuTask.id.asc()).all()
assert [t.task_type for t in tasks] == [
"query_gold_balance",
"query_points",
"create_gold_qr",
"donate_elite_gift",
"exchange_goods",
]
# 前两步(查余额)没有载荷 —— `create_douyu_task` 对空载荷存 None,与旧行为一致
payloads = [(t.result or {}).get("payload", {}) for t in tasks]
assert payloads[2]["amount"] == 98 # (1280-300)/10
assert payloads[2]["_skip_when"]["column"] == "gold_balance"
assert payloads[3]["gift_count"] == 98
assert payloads[3]["_skip_when"]["column"] == "points"
assert "_depends_on" not in payloads[0]
assert payloads[1]["_depends_on"] == tasks[0].id
assert payloads[2]["_depends_on"] == tasks[1].id
assert payloads[3]["_depends_on"] == tasks[2].id
assert payloads[4]["_depends_on"] == tasks[3].id
# 每条链上的任务都要让租用人看得到
assert {t.account_assigned_to for t in tasks} == {self.support_id}
def test_tier_a_creates_single_task(self, monkeypatch):
"""积分已够的账号**不生成任何充值/赠令步骤** —— 那是在花钱。"""
self.make_account("rich2", points=5000)
self._exchange(monkeypatch)
tasks = self.session.query(DouyuTask).all()
assert [t.task_type for t in tasks] == ["exchange_goods"]
def test_no_accounts_does_not_start_batch(self, monkeypatch):
result, captured = self._exchange(monkeypatch)
assert result.allocated == 0
assert result.shortfall == 1
assert captured == []
class TestLeaseEndpoints(_ApiBase):
def _lease(self, holder_id):
account = self.make_account(f"l{holder_id}", points=5000)
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=COMMODITY,
)[0]
def test_release_own_lease(self):
lease = self._lease(self.support_id)
result = pool_router.release_lease(
lease.id, db=self.session, current=self.support
)
assert result["released"] is True
self.session.expire_all()
assert (
self.session.query(AccountLease)
.filter(AccountLease.id == lease.id)
.one()
.status
== "released"
)
def test_cannot_release_someone_elses_lease(self):
lease = self._lease(self.support_id)
with pytest.raises(HTTPException) as exc:
pool_router.release_lease(lease.id, db=self.session, current=self.other)
assert exc.value.status_code == 403
def test_pool_admin_can_release_others(self):
lease = self._lease(self.support_id)
result = pool_router.release_lease(
lease.id, db=self.session, current=self.admin
)
assert result["released"] is True
def test_release_batch_requires_admin_permission(self):
lease = self._lease(self.support_id)
pool.bind_leases_to_batch(self.session, [lease], "batch-x")
result = pool_router.release_batch(
"batch-x", db=self.session, current=self.admin
)
assert result["released"] == 1
def test_list_leases_only_mine(self):
self._lease(self.support_id)
self._lease(self.other_id)
rows = pool_router.list_leases(
mine=True,
account_id=None,
include_finished=False,
db=self.session,
current=self.support,
)
assert len(rows) == 1
assert rows[0].status == "active"
class TestLegacyTaskCreationUnchanged(_ApiBase):
def test_override_defaults_to_account_owner(self):
"""不传新参数时 `account_assigned_to` 仍取账号归属 —— 旧路径行为不变。"""
from web.backend.services.douyu_service import create_douyu_planned_tasks
account = Account(
username="legacy",
password="p",
email="e@e.com",
email_password="ep",
status="active",
ck_check_status="valid",
cookie="PHPSESSID=x",
assigned_to=self.support_id,
)
self.session.add(account)
self.session.commit()
batch_id, count = create_douyu_planned_tasks(
self.session,
[account.id],
"exchange_goods",
"elite",
self.support_id,
{"commodity_id": COMMODITY},
)
assert count == 1
task = (
self.session.query(DouyuTask).filter(DouyuTask.batch_id == batch_id).one()
)
assert task.account_assigned_to == self.support_id