"""账号登录失败分类统计、筛选与多选导出测试。 涵盖： 1. `list_accounts` 携带 `failure_kind` 筛选； 2. `AccountOut` 序列化包含 `relogin_failure_kind`、`relogin_failure_label`、`relogin_last_error`； 3. `accounts_summary` 聚合返回 `login_failures` 分类统计； 4. `POST /api/accounts/export` 导出纯文本账号名单 (txt) 与结构化报表 (csv)； 5. `must_bind_mobile` 自动分型与进入人工态判定。 """ from __future__ import annotations import csv import io from tests.support.db import get_test_engine, make_test_session from web.backend.models import Account, User from web.backend.routers.accounts import ( accounts_summary, export_accounts, list_accounts, ) from web.backend.schemas import AccountExportRequest from web.backend.services.account_lifecycle import STATUS_MANUAL, derive_status from web.backend.services.login_failure import classify_failure, failure_kind_label class TestAccountFailureClassificationAndExport: def setup_method(self): self.engine = get_test_engine() self.session = make_test_session(self.engine) self.admin = User( username="fail-admin", password_hash="h", role="super_admin", is_active=True ) self.support = User( username="fail-support", password_hash="h", role="support", is_active=True ) self.session.add_all([self.admin, self.support]) self.session.commit() def teardown_method(self): self.session.close() def _create_account(self, name: str, **kwargs) -> Account: acc = Account( username=name, password="pwd", email=f"{name}@test.com", email_password="epwd", cookie=kwargs.pop("cookie", "acf_uid=123"), ck_check_status=kwargs.pop("ck_check_status", "valid"), **kwargs, ) self.session.add(acc) self.session.commit() return acc def test_classify_must_bind_mobile(self): """测试必须绑定手机号分型判定。""" assert ( classify_failure("跳过手机号绑定失败: 必须绑定手机号") == "must_bind_mobile" ) assert classify_failure("登录失败: 必须绑定手机号") == "must_bind_mobile" assert ( classify_failure( "重新登录失败: 跳过手机号绑定失败: 必须绑定手机号（旧 Cookie 已保留）" ) == "must_bind_mobile" ) assert failure_kind_label("must_bind_mobile") == "必须绑定手机号" # 验证 must_bind_mobile 在 CK 失效时被 derive_status 推导为 manual 待人工 acc = Account( username="bind_test", password="p", email="e@e.com", email_password="ep", cookie="c", ck_check_status="invalid", relogin_failure_kind="must_bind_mobile", relogin_attempts=1, ) assert derive_status(acc) == STATUS_MANUAL def test_classify_account_canceled_or_banned(self): """测试账号已注销/封禁/冻结归入凭证/账号异常问题。""" assert ( classify_failure( "重新登录失败: 第二次登录失败: 账号已注销（旧 Cookie 已保留）" ) == "credential" ) assert classify_failure("第二次登录失败: 账号已注销") == "credential" assert classify_failure("登录失败: 该账号已被封禁") == "credential" assert classify_failure("账号已冻结，请联系客服") == "credential" def test_list_accounts_filter_by_failure_kind_and_serialization(self): """测试按 failure_kind 筛选与 AccountOut 序列化包含失败字段。""" self._create_account( "acc_mobile", relogin_failure_kind="must_bind_mobile", relogin_last_error="跳过手机号绑定失败: 必须绑定手机号", ) self._create_account( "acc_captcha", relogin_failure_kind="captcha", relogin_last_error="极验验证码拦截", ) self._create_account("acc_normal", relogin_failure_kind="") # 1. 筛选 must_bind_mobile res_mobile = list_accounts( failure_kind="must_bind_mobile", page=1, page_size=20, db=self.session, current=self.admin, ) assert res_mobile["total"] == 1 item = res_mobile["items"][0] assert item.username == "acc_mobile" # 检查 Pydantic 序列化 (防静默丢字段) dumped = item.model_dump() assert dumped["relogin_failure_kind"] == "must_bind_mobile" assert dumped["relogin_failure_label"] == "必须绑定手机号" assert "必须绑定手机号" in dumped["relogin_last_error"] # 2. 筛选 captcha res_captcha = list_accounts( failure_kind="captcha", page=1, page_size=20, db=self.session, current=self.admin, ) assert res_captcha["total"] == 1 assert res_captcha["items"][0].username == "acc_captcha" def test_accounts_summary_login_failures_aggregation(self): """测试 accounts_summary 接口聚合各错误分类统计。""" self._create_account("acc_m1", relogin_failure_kind="must_bind_mobile") self._create_account("acc_m2", relogin_failure_kind="must_bind_mobile") self._create_account("acc_c1", relogin_failure_kind="captcha") self._create_account("acc_q1", relogin_failure_kind="email_quota") self._create_account("acc_norm", relogin_failure_kind="") summary = accounts_summary(db=self.session, current=self.admin) assert "login_failures" in summary failures = summary["login_failures"] assert failures.get("must_bind_mobile") == 2 assert failures.get("captcha") == 1 assert failures.get("email_quota") == 1 def test_export_accounts_txt_format(self): """测试导出纯账号名名单 (txt)。""" acc1 = self._create_account("user_alpha", tag="VIP") acc2 = self._create_account("user_beta", tag="VIP") self._create_account("user_gamma", tag="OTHER") # 1. 显式 ID 多选导出 req1 = AccountExportRequest( account_ids=[acc1.id, acc2.id], format="txt", ) resp1 = export_accounts(req1, db=self.session, current=self.admin) assert resp1.media_type.startswith("text/plain") body1 = resp1.body.decode("utf-8") assert body1.split("\n") == ["user_alpha", "user_beta"] # 2. 按筛选条件全选导出 (all_matching=True) req2 = AccountExportRequest( all_matching=True, tag="VIP", format="txt", ) resp2 = export_accounts(req2, db=self.session, current=self.admin) body2 = resp2.body.decode("utf-8") assert body2.split("\n") == ["user_alpha", "user_beta"] def test_export_accounts_csv_format(self): """测试导出完整明细报表 (csv，带 BOM 和中文分型)。""" acc = self._create_account( "user_csv_test", points=500, gold_balance=10, tag="TEST_TAG", relogin_failure_kind="must_bind_mobile", relogin_last_error="跳过手机号绑定失败: 必须绑定手机号", assigned_to=self.support.id, ) req = AccountExportRequest( account_ids=[acc.id], format="csv", ) resp = export_accounts(req, db=self.session, current=self.admin) assert resp.media_type.startswith("text/csv") content = resp.body.decode("utf-8") assert content.startswith("\ufeff") # 校验 UTF-8 BOM reader = csv.reader(io.StringIO(content.lstrip("\ufeff"))) rows = list(reader) header = rows[0] assert "账号" in header assert "登录异常分类" in header assert "最新报错" in header assert "归属客服" in header data_row = rows[1] assert data_row[0] == "user_csv_test" assert data_row[2] == "必须绑定手机号" assert "必须绑定手机号" in data_row[3] assert data_row[4] == "TEST_TAG" assert data_row[5] == "fail-support"