"""租约兜底回收:漏释放、孤儿批次、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}