yml2213
4eef45efe4
- 补充必须绑定手机号与账号注销/封禁/冻结等错误文案的识别规则 - 增加数据修正迁移 20261008_0052，回写历史账号和登录任务的失败分型 - 增加列表查询与导出时根据错误文案自动兜底细化 failure_kind 的逻辑 - 增加并完善相关分型单元测试
216 lines
8.3 KiB
Python
216 lines
8.3 KiB
Python
"""账号登录失败分类统计、筛选与多选导出测试。
|
||
|
||
涵盖：
|
||
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"
|