"""P1a 回归：CK 自动巡检。 两块约束必须一起守住： 1. **调度器行为**：只跑到点的、打开开关的作业；作业抛异常只记 failed，不能 把调度线程带崩；跑完排下一轮时间；已有配置不能被重启重置。 2. **本作业绝不改 CK**：P1a 的定位是「只检测、只标记」，所以巡检后 ``cookie`` / ``status`` / ``message`` / ``attempt_*`` 必须一字未动 —— 线上全靠这条保证「自动化上线了但人工流程照旧」，回归时一旦有人顺手写了 ``status``，这里要立刻红。 """ from datetime import UTC, datetime, timedelta import pytest from fastapi import HTTPException from tests.support.db import get_test_engine, make_test_session from web.backend.deps import require_permission from web.backend.models import Account, AutomationJob, AutomationRun, LoginTask, User from web.backend.services.automation import ( JobSpec, register_default_jobs, register_job, reset_registry, run_due_jobs, run_job_now, ) from web.backend.services.automation.jobs import ck_health class _Base: def setup_method(self): self.engine = get_test_engine() self.session = make_test_session(self.engine) self.admin = User( username="admin", password_hash="h", role="super_admin", is_active=True ) self.support = User( username="kf", password_hash="h", role="support", is_active=True ) self.session.add_all([self.admin, self.support]) self.session.commit() # 注册表是模块级的，不重置会被上一个用例注册的探针作业污染。 reset_registry() register_default_jobs() def teardown_method(self): self.session.close() def make_account(self, name="acc", *, auto_managed=True, assigned_to=None): account = Account( username=name, password="p", email=f"{name}@example.com", email_password="ep", assigned_to=assigned_to, auto_managed=auto_managed, ) self.session.add(account) self.session.commit() return account def make_task( self, account, *, cookie="acf_uid=1", status="success", attempt_status="", ck_checked_at=None, ck_check_status="", ): task = LoginTask( batch_id="b1", account_id=account.id, created_by=self.admin.id, status=status, cookie=cookie, attempt_status=attempt_status, ck_checked_at=ck_checked_at, ck_check_status=ck_check_status, ) self.session.add(task) self.session.commit() return task class TestCkHealthSelection(_Base): def _selected_ids(self, job): return [ t.id for t in ck_health._select_tasks(self.session, job, datetime.now(UTC)) ] def test_picks_only_rows_that_still_hold_cookie(self): job = self._job() with_cookie = self.make_task(self.make_account("a")) self.make_task(self.make_account("b"), cookie="") assert self._selected_ids(job) == [with_cookie.id] def test_skips_accounts_with_inflight_attempt(self): job = self._job() idle = self.make_task(self.make_account("a")) self.make_task(self.make_account("b"), attempt_status="running") self.make_task(self.make_account("c"), attempt_status="pending") assert self._selected_ids(job) == [idle.id] def test_skips_accounts_opted_out_of_automation(self): job = self._job() managed = self.make_task(self.make_account("a")) self.make_task(self.make_account("b", auto_managed=False)) assert self._selected_ids(job) == [managed.id] def test_respects_interval_and_prefers_stalest(self): job = self._job() fresh = self.make_task( self.make_account("fresh"), ck_checked_at=datetime.now(UTC) ) stale = self.make_task( self.make_account("stale"), ck_checked_at=datetime.now(UTC) - timedelta(hours=30), ) never = self.make_task(self.make_account("never"), ck_checked_at=None) ids = self._selected_ids(job) assert fresh.id not in ids # 从没检测过的排最前，其次是久未检测的 assert ids[0] == never.id assert stale.id in ids def test_generated_sql_is_mysql_compatible(self): """回归：选行 SQL 不能出现 MySQL 不认识的 NULLS FIRST。 生产是 MySQL、本地测试是 SQLite（SQLite 支持 NULLS FIRST，所以行为测试 发现不了），这里按 MySQL 方言编译一次断言。 """ from sqlalchemy.dialects import mysql query = ck_health.build_select_query( self.session, self._job(), datetime.now(UTC) ) sql = str(query.statement.compile(dialect=mysql.dialect())).upper() assert "NULLS FIRST" not in sql and "NULLS LAST" not in sql # 「没检测过的优先」要用可移植写法表达 assert "IS NULL DESC" in sql def test_batch_size_caps_the_round(self): job = self._job() job.batch_size = 2 for index in range(5): self.make_task(self.make_account(f"a{index}")) assert len(self._selected_ids(job)) == 2 def _job(self): return AutomationJob( key=ck_health.JOB_KEY, interval_seconds=6 * 3600, batch_size=200 ) class TestCkHealthNeverTouchesCookie(_Base): def test_patrol_only_writes_check_fields(self): job = self._job() account = self.make_account("acc") task = self.make_task(account, attempt_status="failed", ck_check_status="") before = (task.cookie, task.status, task.message, task.attempt_status) ck_health.check_douyu_cookie = _fake_check(valid=True) try: counters = ck_health.run_ck_health(self.session, job) finally: ck_health.check_douyu_cookie = _REAL_CHECK self.session.refresh(task) assert counters["scanned"] == 1 and counters["valid"] == 1 assert task.ck_check_status == "valid" assert task.ck_checked_at is not None assert task.ck_check_result["nickname"] == "测试" # 核心断言：CK 本体与尝试状态一字未动 assert (task.cookie, task.status, task.message, task.attempt_status) == before def test_invalid_result_marks_invalid_but_keeps_cookie(self): job = self._job() task = self.make_task(self.make_account("acc")) # 必须是鉴权类失败（斗鱼说未登录）才会被标成 invalid ck_health.check_douyu_cookie = _fake_check( valid=False, message="鱼丸接口: 账户登陆异常，请重新登陆；等级接口: 用户未登录", ) try: counters = ck_health.run_ck_health(self.session, job) finally: ck_health.check_douyu_cookie = _REAL_CHECK self.session.refresh(task) assert counters["invalid"] == 1 assert task.ck_check_status == "invalid" assert task.cookie == "acf_uid=1" def test_detection_exception_does_not_abort_round(self): """单个账号检测抛异常不影响整轮，但也不能把它标成失效。""" job = self._job() first = self.make_task(self.make_account("a"), ck_check_status="valid") second = self.make_task(self.make_account("b"), ck_check_status="valid") ck_health.check_douyu_cookie = _fake_check(raises=True) try: counters = ck_health.run_ck_health(self.session, job) finally: ck_health.check_douyu_cookie = _REAL_CHECK assert counters["scanned"] == 2 # 查不到不等于失效：两行都保留原结论，只记录这次异常 assert counters["invalid"] == 0 assert counters["skipped"] == 2 for task in (first, second): self.session.refresh(task) assert task.ck_check_status == "valid" assert task.ck_check_result["verdict"] == "error" def _job(self): return AutomationJob( key=ck_health.JOB_KEY, interval_seconds=6 * 3600, batch_size=200 ) class TestScheduler(_Base): def _register_probe(self, calls): def run(db, job): calls.append(job.key) return {"scanned": 3, "valid": 2, "invalid": 1, "message": "ok"} register_job( JobSpec( key="probe", title="探针", interval_seconds=3600, batch_size=1, concurrency=1, run=run, ) ) def test_only_enabled_jobs_run(self): from web.backend.services.automation import ensure_jobs calls: list[str] = [] self._register_probe(calls) ensure_jobs(self.session) self.session.query(AutomationJob).filter(AutomationJob.key == "probe").update( {"enabled": True} ) self.session.commit() run_due_jobs(self.session) assert calls == ["probe"] calls.clear() run_due_jobs(self.session) # 上一轮已排了 next_run_at，未到点不再跑 assert calls == [] def test_job_failure_is_recorded_not_raised(self): from web.backend.services.automation import ensure_jobs def boom(db, job): raise RuntimeError("接口挂了") register_job( JobSpec( key="boom", title="会炸的作业", interval_seconds=3600, batch_size=1, concurrency=1, run=boom, ) ) ensure_jobs(self.session) self.session.query(AutomationJob).filter(AutomationJob.key == "boom").update( {"enabled": True} ) self.session.commit() outcomes = run_due_jobs(self.session) assert outcomes[0]["status"] == "failed" run = self.session.query(AutomationRun).one() assert run.status == "failed" and "接口挂了" in run.message def test_next_run_follows_tick_and_counters_persist(self): from web.backend.services.automation import ensure_jobs calls: list[str] = [] self._register_probe(calls) ensure_jobs(self.session) self.session.query(AutomationJob).filter(AutomationJob.key == "probe").update( {"enabled": True} ) self.session.commit() run_due_jobs(self.session) job = ( self.session.query(AutomationJob).filter(AutomationJob.key == "probe").one() ) assert job.last_status == "success" # 下一轮在一个作业周期之后，与 interval_seconds（每行多久重检）无关 from web.backend.services.automation.scheduler import JOB_INTERVAL_SECONDS assert _aware(job.next_run_at) - _aware(job.last_run_at) == timedelta( seconds=JOB_INTERVAL_SECONDS ) run = self.session.query(AutomationRun).one() assert (run.scanned, run.valid, run.invalid) == (3, 2, 1) def test_ensure_jobs_keeps_operator_settings(self): from web.backend.services.automation import ensure_jobs ensure_jobs(self.session) job = ( self.session.query(AutomationJob) .filter(AutomationJob.key == ck_health.JOB_KEY) .one() ) job.enabled = True job.batch_size = 7 self.session.commit() ensure_jobs(self.session) # 模拟重启 self.session.refresh(job) assert job.enabled is True and job.batch_size == 7 def test_run_job_now_ignores_switch(self): from web.backend.services.automation import ensure_jobs ensure_jobs(self.session) outcome = run_job_now(self.session, ck_health.JOB_KEY) assert outcome["job_key"] == ck_health.JOB_KEY assert self.session.query(AutomationRun).count() == 1 def test_run_job_now_unknown_key(self): with pytest.raises(KeyError): run_job_now(self.session, "nope") class TestAutomationPermissions(_Base): def test_view_allows_operation_denies_support(self): from web.backend.permissions import ROLE_PERMISSIONS assert "automation:view" in ROLE_PERMISSIONS["operation"] assert "automation:view" not in ROLE_PERMISSIONS["support"] assert "automation:manage" in ROLE_PERMISSIONS["super_admin"] assert "automation:manage" not in ROLE_PERMISSIONS["operation"] def test_manage_dependency_rejects_operator(self): operator = User( username="op", password_hash="h", role="operation", is_active=True ) checker = require_permission("automation:manage") with pytest.raises(HTTPException) as exc: checker(current_user=operator) assert exc.value.status_code == 403 def test_view_dependency_accepts_operator(self): operator = User( username="op", password_hash="h", role="operation", is_active=True ) checker = require_permission("automation:view") assert checker(current_user=operator).id == operator.id def _aware(value): """库里读出的是 naive UTC，比较前补时区。""" return value if value.tzinfo else value.replace(tzinfo=UTC) _REAL_CHECK = ck_health.check_douyu_cookie def _fake_check(*, valid=True, raises=False, message=None): def fake(cookie: str) -> dict: if raises: raise RuntimeError("网络异常") return { "valid": valid, "message": message if message is not None else ("ok" if valid else "Cookie 已失效"), "fish_ball": 1, "nickname": "测试", "level": 30, "checked_at": datetime.now(UTC), } return fake class TestAutomationApi(_Base): """接口层：序列化字段、开关落库、手动触发不受开关限制。""" def _api(self): from web.backend.routers import automation as api return api def test_list_jobs_exposes_scheduler_and_job_state(self): api = self._api() payload = api.list_jobs(db=self.session, current=self.admin) # 心跳（轮询间隔）与作业周期是两个概念，接口都要暴露出来： # 心跳密、只做「有没有到点」的检查；作业周期决定同一个作业隔多久跑下一轮。 assert ( payload["scheduler"]["tick_seconds"] < payload["scheduler"]["job_interval_seconds"] ) assert payload["scheduler"]["job_interval_seconds"] == 300 assert payload["scheduler"]["running"] is False keys = [item["key"] for item in payload["jobs"]] assert ck_health.JOB_KEY in keys job = next(item for item in payload["jobs"] if item["key"] == ck_health.JOB_KEY) assert job["enabled"] is False assert job["interval_seconds"] == 6 * 3600 assert job["batch_size"] == 200 assert job["latest_run"] is None def test_update_job_persists_and_schedules_next_run(self): api = self._api() api.update_job( ck_health.JOB_KEY, api.JobUpdate( enabled=True, interval_seconds=3600, batch_size=5, concurrency=2 ), db=self.session, current=self.admin, ) job = ( self.session.query(AutomationJob) .filter(AutomationJob.key == ck_health.JOB_KEY) .one() ) assert job.enabled is True assert job.interval_seconds == 3600 assert job.batch_size == 5 assert job.next_run_at is not None # 打开开关就排了一轮 # 审计留痕 from web.backend.models import AuditLog entry = ( self.session.query(AuditLog) .filter(AuditLog.action == "automation:update") .one() ) assert entry.target == ck_health.JOB_KEY def test_run_endpoint_records_run_and_audit(self): api = self._api() self.make_task(self.make_account("a")) ck_health.check_douyu_cookie = _fake_check(valid=True) try: outcome = api.run_job( ck_health.JOB_KEY, db=self.session, current=self.admin ) finally: ck_health.check_douyu_cookie = _REAL_CHECK assert outcome["status"] == "success" assert outcome["scanned"] == 1 assert self.session.query(AutomationRun).count() == 1 from web.backend.models import AuditLog assert ( self.session.query(AuditLog) .filter(AuditLog.action == "automation:run") .count() == 1 ) def test_run_endpoint_unknown_job_returns_404(self): api = self._api() with pytest.raises(HTTPException) as exc: api.run_job("nope", db=self.session, current=self.admin) assert exc.value.status_code == 404 def test_runs_endpoint_filters_by_job_key(self): api = self._api() self.session.add( AutomationRun(job_key=ck_health.JOB_KEY, status="success", scanned=1) ) self.session.add(AutomationRun(job_key="other", status="failed")) self.session.commit() payload = api.list_runs( job_key=ck_health.JOB_KEY, limit=50, db=self.session, current=self.admin ) assert len(payload["items"]) == 1 assert payload["items"][0]["job_key"] == ck_health.JOB_KEY class TestFailureClassification(_Base): """「查不到」不等于「失效」：非鉴权类失败不能覆盖原来的结论。 斗鱼接口抖动/被风控时，如果不加区分地写 invalid，一小时就能把上千个好账号 标成失效，客服会误判成 CK 大面积挂掉。这里盯三件事：分类正确、error 不改 结论、失败占比过高时自动暂停作业。 """ def test_auth_failure_message_counts_as_invalid(self): assert ( ck_health.classify_result( { "valid": False, "message": "鱼丸接口: 账户登陆异常，请重新登陆；等级接口: 用户未登录", } ) == "invalid" ) def test_network_failure_counts_as_error_not_invalid(self): assert ( ck_health.classify_result( { "valid": False, "message": "鱼丸接口: 请求失败: timeout；等级接口: 请求失败: timeout", } ) == "error" ) assert ( ck_health.classify_result({"valid": False, "message": "响应异常"}) == "error" ) def test_valid_still_valid(self): assert ck_health.classify_result({"valid": True, "message": "有效"}) == "valid" def test_error_keeps_previous_verdict_but_marks_checked(self): job = self._job() task = self.make_task(self.make_account("acc"), ck_check_status="valid") ck_health.check_douyu_cookie = _fake_check( valid=False, message="请求失败: timeout" ) try: counters = ck_health.run_ck_health(self.session, job) finally: ck_health.check_douyu_cookie = _REAL_CHECK self.session.refresh(task) assert counters["valid"] == 0 and counters["invalid"] == 0 assert counters["skipped"] == 1 # error 计入 skipped assert task.ck_check_status == "valid" # 原结论保留 assert task.ck_check_result["verdict"] == "error" # 但记录了这次查不到 assert task.ck_checked_at is not None def test_mass_network_failure_pauses_the_job(self): job = self._job() job.enabled = True job.batch_size = 40 for index in range(40): self.make_task(self.make_account(f"a{index}")) ck_health.check_douyu_cookie = _fake_check( valid=False, message="请求失败: Connection refused" ) try: counters = ck_health.run_ck_health(self.session, job) finally: ck_health.check_douyu_cookie = _REAL_CHECK assert counters["skipped"] == 40 assert counters["detail"]["paused"] is True assert job.enabled is False # 自动暂停，不再继续污染 assert "已自动暂停" in counters["message"] def test_normal_invalid_does_not_pause(self): """真的都是死号时不能暂停 —— 首轮 100% invalid 是合法且预期的情况。""" job = self._job() job.enabled = True job.batch_size = 40 for index in range(40): self.make_task(self.make_account(f"a{index}")) ck_health.check_douyu_cookie = _fake_check( valid=False, message="鱼丸接口: 账户登陆异常，请重新登陆；等级接口: 用户未登录", ) try: counters = ck_health.run_ck_health(self.session, job) finally: ck_health.check_douyu_cookie = _REAL_CHECK assert counters["invalid"] == 40 assert counters["skipped"] == 0 assert job.enabled is True assert counters["detail"]["paused"] is False def _job(self): return AutomationJob( key=ck_health.JOB_KEY, interval_seconds=6 * 3600, batch_size=200, concurrency=3, ) class TestOrphanRunCleanup(_Base): """回归：进程重启后，卡在 running 的运行记录必须被收尾。 作业在调度线程里同步跑，进程一重启那轮就没机会写回结果，记录会永远停在 running，页面上一直显示「执行中」。登录任务早有孤儿清理，运行记录原先漏了 —— 线上部署时打断了一轮，run 14 就是这么卡住的。 """ def test_running_runs_are_closed_on_startup(self): from web.backend.services.automation.scheduler import cleanup_orphan_runs stale = AutomationRun( job_key="ck_relogin", started_at=datetime.now(UTC), status="running" ) done = AutomationRun( job_key="ck_health", started_at=datetime.now(UTC), finished_at=datetime.now(UTC), status="success", ) self.session.add_all([stale, done]) self.session.commit() cleaned = cleanup_orphan_runs(self.session) self.session.refresh(stale) self.session.refresh(done) assert cleaned == 1 assert stale.status == "interrupted" assert stale.finished_at is not None assert stale.message == "服务重启中断" assert done.status == "success" # 已完成的不受影响 class TestHeartbeatDecoupledFromInterval(_Base): """回归：心跳频率不能直接当作业周期用。 曾经 TICK_SECONDS 既是心跳又是作业周期（next_run_at = 结束 + 300s，心跳也是 300s）。两者同频会系统性跳轮：作业 19 秒跑完 → next 在 5 分钟后，而下一次 心跳正好在 5 分钟整，常常差几十秒没到点，于是整轮空转 —— 实际变成每 10 分钟 才跑一次。线上表现就是「检测只要 19 秒，却总是排队等很久」。 """ def test_heartbeat_is_much_shorter_than_job_interval(self): from web.backend.services.automation.scheduler import ( JOB_INTERVAL_SECONDS, TICK_SECONDS, ) assert TICK_SECONDS < JOB_INTERVAL_SECONDS, ( "心跳必须比作业周期密，否则会出现「差几十秒没到点→整轮空转」的跳轮" ) # 至少密 4 倍：作业耗时波动时不会积压成整轮跳过 assert JOB_INTERVAL_SECONDS / TICK_SECONDS >= 4 def test_job_runs_again_once_its_own_interval_elapsed(self): """心跳多跳几次后，到点的作业必须被跑到。""" job = self._register_probe() run_due_jobs(self.session) first = self.session.query(AutomationRun).count() # 周期未到：无论心跳跳多少次都不跑 for _ in range(3): run_due_jobs(self.session) assert self.session.query(AutomationRun).count() == first # 把 next_run_at 拨到过去（等价于「周期已到」），下一次心跳就该跑 job.next_run_at = datetime.now(UTC) - timedelta(seconds=1) self.session.commit() run_due_jobs(self.session) assert self.session.query(AutomationRun).count() == first + 1 def _register_probe(self): from web.backend.services.automation.scheduler import ( JobSpec, register_job, reset_registry, ) reset_registry() register_job( JobSpec( key="probe", title="探针", interval_seconds=6 * 3600, batch_size=10, concurrency=1, run=lambda db, job: {"scanned": 1, "valid": 1, "invalid": 0}, ) ) from web.backend.services.automation import ensure_jobs ensure_jobs(self.session) job = ( self.session.query(AutomationJob).filter(AutomationJob.key == "probe").one() ) job.enabled = True job.next_run_at = datetime.now(UTC) - timedelta(seconds=1) self.session.commit() return job