178 lines
6.4 KiB
Python
178 lines
6.4 KiB
Python
"""租约兜底回收：漏释放、孤儿批次、TTL 到期。
|
|
|
|
主释放路径是任务终态钩子（有单独用例）；这里守的是**保险丝**：
|
|
进程重启、释放抛异常、任务行被清 —— 这三种情况下账号不能永远被锁着。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from datetime import UTC, datetime, timedelta
|
|
|
|
from tests.support.db import get_test_engine, make_test_session
|
|
from web.backend.models import (
|
|
Account,
|
|
AccountLease,
|
|
AutomationJob,
|
|
DouyuTask,
|
|
User,
|
|
)
|
|
from web.backend.services import account_pool as pool
|
|
from web.backend.services.automation.jobs import account_lease_reaper as reaper
|
|
|
|
NOW = datetime.now(UTC).replace(microsecond=0)
|
|
|
|
|
|
class _ReaperBase:
|
|
def setup_method(self):
|
|
self.engine = get_test_engine()
|
|
self.session = make_test_session(self.engine)
|
|
self.user = User(username="reap-kf", password_hash="h", role="support")
|
|
self.session.add(self.user)
|
|
self.session.flush()
|
|
self.user_id = self.user.id
|
|
|
|
def teardown_method(self):
|
|
self.session.close()
|
|
|
|
def _lease(
|
|
self, name: str, *, batch_id: str = "", ttl_minutes: int = 60
|
|
) -> AccountLease:
|
|
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,
|
|
)
|
|
self.session.add(account)
|
|
self.session.commit()
|
|
lease = pool.lease_accounts(
|
|
self.session,
|
|
holder_user_id=self.user_id,
|
|
candidates=[
|
|
pool.make_candidate(
|
|
account_id=account.id,
|
|
username=name,
|
|
points=5000,
|
|
score=1280,
|
|
)
|
|
],
|
|
commodity_id="20260101X_bag1",
|
|
batch_id=batch_id,
|
|
ttl_minutes=ttl_minutes,
|
|
now=NOW,
|
|
)[0]
|
|
return lease
|
|
|
|
def _task(self, batch_id: str, account_id: int, status: str):
|
|
self.session.add(
|
|
DouyuTask(
|
|
batch_id=batch_id,
|
|
account_id=account_id,
|
|
task_type="exchange_goods",
|
|
handbook_scope="elite",
|
|
status=status,
|
|
created_by=self.user_id,
|
|
)
|
|
)
|
|
self.session.commit()
|
|
|
|
def _status(self, lease_id: int) -> str:
|
|
self.session.rollback()
|
|
lease = self.session.get(AccountLease, lease_id)
|
|
assert lease is not None
|
|
return lease.status
|
|
|
|
|
|
class TestReapMissedReleases(_ReaperBase):
|
|
def test_releases_when_batch_has_no_active_task(self):
|
|
lease = self._lease("m1", batch_id="batch-m1")
|
|
self._task("batch-m1", lease.account_id, "success")
|
|
|
|
assert pool.reap_missed_releases(self.session) == 1
|
|
assert self._status(lease.id) == "released"
|
|
|
|
def test_keeps_lease_while_batch_still_running(self):
|
|
lease = self._lease("m2", batch_id="batch-m2")
|
|
self._task("batch-m2", lease.account_id, "running")
|
|
|
|
assert pool.reap_missed_releases(self.session) == 0
|
|
assert self._status(lease.id) == "active"
|
|
|
|
def test_keeps_lease_when_batch_not_created_yet(self):
|
|
"""租约刚建、任务还没建出来（batch_id 为空）—— 不能当成漏释放。"""
|
|
lease = self._lease("m3", batch_id="")
|
|
assert pool.reap_missed_releases(self.session) == 0
|
|
assert self._status(lease.id) == "active"
|
|
|
|
def test_one_running_task_protects_the_whole_batch(self):
|
|
first = self._lease("m4a", batch_id="batch-m4")
|
|
second = self._lease("m4b", batch_id="batch-m4")
|
|
self._task("batch-m4", first.account_id, "success")
|
|
self._task("batch-m4", second.account_id, "pending")
|
|
|
|
# 批次里还有 pending → 两条都不动（正在跑的那条可能还没结束）
|
|
assert pool.reap_missed_releases(self.session) == 0
|
|
assert self._status(first.id) == "active"
|
|
|
|
|
|
class TestReaperJob(_ReaperBase):
|
|
def test_expires_ttl_leases(self):
|
|
lease = self._lease("j1", batch_id="batch-j1", ttl_minutes=1)
|
|
self._task("batch-j1", lease.account_id, "running")
|
|
|
|
job = AutomationJob(key=reaper.JOB_KEY, batch_size=100)
|
|
result = reaper.run_account_lease_reaper(
|
|
self.session, job, now=NOW + timedelta(minutes=5)
|
|
)
|
|
|
|
assert result["expired"] >= 1
|
|
assert self._status(lease.id) == "expired"
|
|
|
|
def test_releases_orphan_batch(self):
|
|
"""注册表里没有这个批次（模拟进程重启）→ 即使任务还是 running 也要收。"""
|
|
lease = self._lease("j2", batch_id="batch-gone")
|
|
self._task("batch-gone", lease.account_id, "running")
|
|
|
|
job = AutomationJob(key=reaper.JOB_KEY, batch_size=100)
|
|
result = reaper.run_account_lease_reaper(self.session, job)
|
|
|
|
assert result["orphan_batch"] >= 1
|
|
assert self._status(lease.id) == "expired"
|
|
|
|
def test_message_is_honest_about_counts(self):
|
|
self._lease("j3", batch_id="batch-j3", ttl_minutes=1)
|
|
job = AutomationJob(key=reaper.JOB_KEY, batch_size=100)
|
|
result = reaper.run_account_lease_reaper(self.session, job)
|
|
assert result["scanned"] == (
|
|
result["expired"] + result["missed_release"] + result["orphan_batch"]
|
|
)
|
|
assert "租约回收" in result["message"]
|
|
|
|
def test_idempotent_second_run_finds_nothing(self):
|
|
lease = self._lease("j4", batch_id="batch-gone")
|
|
self._task("batch-gone", lease.account_id, "running")
|
|
job = AutomationJob(key=reaper.JOB_KEY, batch_size=100)
|
|
reaper.run_account_lease_reaper(self.session, job)
|
|
second = reaper.run_account_lease_reaper(self.session, job)
|
|
assert second["scanned"] == 0
|
|
assert self._status(lease.id) == "expired"
|
|
|
|
def test_freed_account_becomes_available_again(self):
|
|
lease = self._lease("j5", batch_id="batch-gone")
|
|
account_id = lease.account_id
|
|
job = AutomationJob(key=reaper.JOB_KEY, batch_size=100)
|
|
reaper.run_account_lease_reaper(self.session, job)
|
|
|
|
candidates = pool.eligible_accounts(
|
|
self.session,
|
|
score=1280,
|
|
commodity_id="20260101X_bag1",
|
|
now=NOW + timedelta(minutes=5),
|
|
)
|
|
assert account_id in {c.account_id for c in candidates}
|