"""定时补全账号资料的作业（`account_overview_refresh`）。 **为什么不能一次性跑全量**：生产约 8,000 个待查账号，一次性发起 ≈ 8 万次斗鱼请求， 很容易触发风控 —— 而风控的代价是所有账号一起受影响。所以做成定时分批， 并复用「自动巡检」的两道保护（`cookie_operable_filter` + 错误率熔断）。 """ 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, AutomationJob, User from web.backend.services.automation.jobs import account_overview_refresh as job class _Base: def setup_method(self): self.engine = get_test_engine() self.session = make_test_session(self.engine) self.admin = User(username="ovr-admin", password_hash="h", role="super_admin") self.session.add(self.admin) self.session.flush() self.job = AutomationJob( key=job.JOB_KEY, title="账号资料定时补全", # 「同一账号多久重检」= 7 天（就是界面上那个字段） interval_seconds=7 * 24 * 3600, batch_size=10, concurrency=3, enabled=False, ) self.session.add(self.job) self.session.commit() def teardown_method(self): self.session.close() def account( self, name: str, *, ck: str = "valid", queried_days_ago: int | None = None, auto_managed: bool = True, status: str = "active", ) -> Account: acc = Account( username=name, password="p", email=f"{name}@e.com", email_password="ep", cookie="acf_uid=1; acf_ccn=c", ck_check_status=ck, auto_managed=auto_managed, status=status, ) if queried_days_ago is not None: acc.overview_queried_at = datetime.now(UTC) - timedelta( days=queried_days_ago ) self.session.add(acc) self.session.commit() return acc def selected(self) -> list[Account]: return job.build_select_query(self.session, self.job, datetime.now(UTC)).all() class TestSelection(_Base): def test_only_valid_ck_is_selected(self): """只查有效 CK —— 失效的查了也是白查，只会多一条失败记录。""" self.account("ovr-valid", ck="valid") self.account("ovr-invalid", ck="invalid") self.account("ovr-error", ck="error") names = [a.username for a in self.selected()] assert "ovr-valid" in names assert "ovr-invalid" not in names assert "ovr-error" not in names def test_never_queried_comes_first(self): """**没查过的优先** —— 否则周期性刷新会永远挤掉从没查过的新账号。""" self.account("ovr-old", queried_days_ago=30) self.account("ovr-never") assert self.selected()[0].username == "ovr-never" def test_assigned_and_recent_login_lead_within_the_same_group(self): """没查过的仍整体优先；同一档里已分配的在前，再按最近登录。""" stale = self.account("ovr-stale-assigned", queried_days_ago=30) stale.assigned_to = self.admin.id recent = self.account("ovr-recent") recent.cookie_updated_at = datetime.now(UTC) older = self.account("ovr-older") older.cookie_updated_at = datetime.now(UTC) - timedelta(days=10) assigned = self.account("ovr-assigned") assigned.assigned_to = self.admin.id assigned.cookie_updated_at = datetime.now(UTC) - timedelta(days=1) self.session.commit() names = [a.username for a in self.selected()] assert names.index("ovr-assigned") < names.index("ovr-recent") assert names.index("ovr-recent") < names.index("ovr-older") assert names.index("ovr-older") < names.index("ovr-stale-assigned") def test_recently_queried_is_skipped(self): """刚查过的要跳过（`refresh_days` 之内不再查）。""" self.account("ovr-fresh", queried_days_ago=1) names = [a.username for a in self.selected()] assert "ovr-fresh" not in names def test_stale_is_selected_again(self): self.account("ovr-stale", queried_days_ago=30) assert "ovr-stale" in [a.username for a in self.selected()] def test_recheck_interval_comes_from_the_ui_field(self): """**用界面上那个「同一账号多久重检」** —— 与其它批量作业同一套口径。 之前这里读的是 `job.config["refresh_days"]`，于是界面上那个字段 对本作业是**死的**（改了没反应）—— 正是用户批评过的 「看起来能配、其实配的是不相干的东西」。 """ self.job.interval_seconds = 100 * 24 * 3600 self.session.commit() self.account("ovr-30d", queried_days_ago=30) assert "ovr-30d" not in [a.username for a in self.selected()], ( "重检间隔 100 天时，30 天前查过的账号不该重新入选" ) def test_manual_and_unmanaged_accounts_are_excluded(self): """人工状态（停用/归档）与「不参与自动化」的账号不该被自动碰。""" self.account("ovr-disabled", status="disabled") self.account("ovr-archived", status="archived") self.account("ovr-unmanaged", auto_managed=False) names = [a.username for a in self.selected()] for bad in ("ovr-disabled", "ovr-archived", "ovr-unmanaged"): assert bad not in names def test_batch_size_limits_the_round(self): for i in range(5): self.account(f"ovr-batch-{i}") self.job.batch_size = 2 self.session.commit() assert len(self.selected()) == 2 def test_remaining_count_works_with_limit_applied(self): """回归：计数不能走 `build_select_query`。 那个查询已经 `LIMIT` 了，再 `order_by(None)` 会被 SQLAlchemy 拒绝： "order_by() being called on a Query which already has LIMIT"。 （真踩过 —— 作业第一版就是这么挂的。） """ for i in range(4): self.account(f"ovr-count-{i}") self.job.batch_size = 1 self.session.commit() assert job.count_remaining(self.session, self.job, datetime.now(UTC)) >= 4 class TestCircuitBreaker(_Base): """失败率过高就**自己把自己关掉**（风控保护）。 与「自动巡检」同口径。大面积失败通常意味着被风控或斗鱼接口抖动 —— 这时候继续跑只会把更多账号写坏，停手交给人确认才是对的。 """ def _run_with(self, total: int, failed: int): """打桩执行器，只关心失败率熔断的逻辑。""" from web.backend.models import DouyuTask for i in range(total): self.account(f"ovr-cb-{i}") def fake_create(db, ids, task_type, scope, created_by, payload=None): batch_id = "cb-batch" for account_id in ids: self.session.add( DouyuTask( batch_id=batch_id, account_id=account_id, task_type=task_type, handbook_scope=scope, status="planned", created_by=created_by, ) ) self.session.commit() return batch_id, len(ids) session = self.session # 闭包捕获：下面 _FakeRunner 里的 self 是 runner class _FakeRunner: def __init__(self, *a, **k): pass def run(self): # 把前 `failed` 个标成失败，其余成功 rows = ( session.query(DouyuTask) .filter(DouyuTask.batch_id == "cb-batch") .order_by(DouyuTask.id.asc()) .all() ) for index, row in enumerate(rows): row.status = "failed" if index < failed else "success" session.commit() import web.backend.services.automation.jobs.account_overview_refresh as mod original_create = mod.create_douyu_planned_tasks original_runner = mod.DouyuBatchRunner mod.create_douyu_planned_tasks = fake_create mod.DouyuBatchRunner = _FakeRunner try: return job.run_account_overview_refresh(self.session, self.job) finally: mod.create_douyu_planned_tasks = original_create mod.DouyuBatchRunner = original_runner def test_high_failure_ratio_disables_the_job(self): out = self._run_with(total=20, failed=15) # 75% 失败 self.session.refresh(self.job) assert out["paused"] is True assert self.job.enabled is False, "失败率过高应自动关闭作业" assert "风控" in out["message"] def test_normal_failure_ratio_keeps_job_running(self): out = self._run_with(total=20, failed=1) # 5% 失败（真实的 CK 失效率） self.session.refresh(self.job) assert out["paused"] is False assert "风控" not in out["message"] def test_small_sample_does_not_trigger_breaker(self): """样本太少时比例没有意义（1 个失败就是 100%），不该熔断。""" out = self._run_with(total=3, failed=3) self.session.refresh(self.job) assert out["paused"] is False, "样本不足时不该熔断"