diff --git a/CHANGELOG.md b/CHANGELOG.md index b9dcb08..6f28b09 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -8,6 +8,32 @@ ### 🚧 进行中 +#### 新增（T6.2 已完成） + +- ✨ **管理后台仪表盘 — 真实 SQL 聚合层 `admin/stats.py`**（434 行） + - **端点落地**：`GET /api/stats/dashboard`（一站式仪表盘）+ `GET /api/stats/orders`（沿用 T6.1 stub 字段名，去掉 `_stub` 标记） + - **纯函数聚合层**：`build_dashboard_payload` / `compute_summary` / `compute_by_status` / `compute_by_source` / `compute_by_service_version` / `compute_trends` / `generate_day_series` + - **响应契约**：`summary`(订单/用户/收入 + 今日/7d/30d 切片 9 字段) + `by_status` / `by_source` / `by_service_version` (完整 0 填充) + `trends` (today=1, 7d=7, 30d=30 个点, 日粒度, 0 填充) + `generated_at` + - **关键口径**： + - 收入 = `paid / serving / delivered / completed` 四态订单的 `amount_cents` 累计值；`pending`（未付款）与 `refunded`（已退款）排除 + - 趋势桶粒度 = 日（UTC，`YYYY-MM-DD`），用 `substr(created_at, 1, 10)` 切片 + - 0 填充 = 窗口内的"无订单日"也返回 0 点，前端拿到稠密序列 + - 数据源隔离 = `orders` 走 `GAOKAO_ORDERS_DB_PATH`（默认 `data/orders.db`），`admin_users` 走 `GAOKAO_DB_PATH`（默认 `data/orders/admin.db`） + - 不读 PII：统计路径只触碰 `amount_cents` / `status` / `source` / `service_version` / `created_at` + - **配置新增**：`Settings.orders_db_path`（`GAOKAO_ORDERS_DB_PATH` 环境变量，默认 `data/orders.db`），与 `data.channel_sync.webhook_server` 已有的同名变量对齐 + - **测试 12 个全部通过**（`admin/tests/test_routes_stats_dashboard.py`）： + - 鉴权（无 token 401 / 有 token 200） + - 空库形状契约：summary / by\_\* / trends 三层结构稳定 + - 趋势序列：1 / 7 / 30 个点，按日期严格升序 + - 0 填充点：包含完整三字段（`date` / `orders` / `revenue_cents`） + - 真实数据：窗口边界（45 天前的订单被 30d 窗口排除，但计入 total） + - 收入口径：pending / refunded 不计入 revenue + - 0 填充：范围内无订单的日也返回 0 点 + - 兼容层：`/api/stats/orders` 字段名不变，`_stub` 标记已移除 + - **T9.3 已知缺口已闭环**：`admin/tests/conftest.py` 新增 `orders_db` autouse fixture（T6.2 起，所有测试默认带空 orders DB），`pytest -q admin/tests` 87/87 通过 + - **测试套件**：`pytest -q admin/tests` 87/87 passed；`pytest -q` 仓库全量 392/392 passed；`ruff check admin/stats.py admin/routes/stats.py admin/tests/test_routes_stats_dashboard.py admin/tests/conftest.py admin/config.py admin/tests/test_routes.py` All checks passed + - **文档同步**：`docs/plans/T6-admin-mvp.md` 新增 §10（T6.2 设计与 DoD）；§12 后续任务衔接标 [x]；`README.md` 新增 T6.2 章节；`CHANGELOG.md` 本条 + #### 新增（T9.3 已完成） - ✨ **结构化 JSON 日志 `admin/logging_utils.py`** @@ -18,7 +44,7 @@ - `admin/errors/exceptions.py` 的 `BusinessError` / `HTTPException` / `RequestValidationError` / 兜底异常日志全部升级为结构化事件 - 新增 `admin/tests/test_logging.py` 8 个测试，覆盖 formatter、上下文绑定、异常 traceback、FastAPI 端到端 JSON 日志 - 定向验证通过：`pytest -q admin/tests/test_logging.py admin/tests/test_errors.py` = 34/34 passed；`ruff check admin/app.py admin/errors/exceptions.py admin/logging_utils.py admin/tests/test_logging.py admin/tests/test_errors.py` 通过 - - 当前已知缺口：`pytest -q admin/tests` 仍有 1 个非 T9.3 既有失败（`admin/tests/test_routes.py::test_stats_orders_real_shape`，`sqlite3.OperationalError: no such table: orders`） + - 已知缺口（T6.2 已闭环）：`pytest -q admin/tests` 之前有 1 个非 T9.3 失败（`test_stats_orders_real_shape` 报 `sqlite3.OperationalError: no such table: orders`）— 根因是 stats 端点读 orders 表，但 T6.1 阶段 conftest 没建 orders DB；T6.2 在 conftest 加 `orders_db` autouse fixture 闭环，当前 `pytest -q admin/tests` 87/87 通过 #### 新增（T4.2 已完成） @@ -59,6 +85,27 @@ - `ruff check data/crowd_db/ scripts/verify_t2_4_e2e.py` — All checks passed - `validate_template.py` 仍 PASS（模板字段契约未破坏） +#### 新增（T2.5 已完成） + +- ✅ **扎堆检测单元测试 T2.5** — `data/crowd_db/tests/test_crowd_detector.py` 由 35 增至 55 个测试（新增 20 个） + - **用例 1 高风险识别**：`test_high_risk_use_case_frequency_4_full_payload`（freq=4 全平台 + predicted_increase + alternatives）/ `test_high_risk_boundary_frequency_exactly_4`（>=4 闭合区间边界）/ `test_high_risk_distinct_from_medium_and_low`（high/medium 互不混淆） + - **用例 2 替代方案**：`test_alternatives_use_case_contains_required_keys`（每个 alt 必须含 name/major/score, 0-100 区间）/ `test_alternatives_use_case_sortable_by_score`（detector 透传 alternatives 字段不丢）/ `test_alternatives_use_case_empty_list_is_valid`（空列表合法） + - **用例 3 跨省份**：`test_cross_province_hunan_hit_guangdong_miss`（湖南 575 命中 vs 广东 575 无数据隔离）/ `test_cross_province_beijing_at_690`（同校不同省命中不同记录 — 清华 工科试验班）/ `test_cross_province_loader_called_with_correct_province`（province 透传给 loader, 不做全局查询） + - **用例 4 异常处理**：`test_exception_use_case_loader_returns_none_field`（name=None 不崩）/ `test_exception_use_case_loader_returns_non_dict`（混入 None/str/int 时优雅过滤）/ `test_exception_use_case_loader_raises_propagates`（loader 抛错向上传播）/ `test_exception_use_case_unrecognized_entry_type`（int/float/object 走 str 兜底）/ `test_exception_use_case_entry_with_int_school_dict`（school=0 falsy 跳过）/ `test_exception_use_case_province_is_none`（province=None 不抛）/ `test_exception_use_case_tuple_with_non_string_school`（tuple 元组非 str school 走 str() 兜底）/ `test_exception_use_case_dict_with_truthy_non_string_school`（0.5 这种 truthy 非 str school 走 str() 兜底）/ `test_exception_use_case_dict_with_name_truthy_non_string`（name 字段 truthy 非 str 同样兜底）/ `test_exception_use_case_single_element_tuple_non_string`（单元素 tuple 非 str 兜底）/ `test_exception_use_case_malformed_frequency_in_record`（frequency="not a number"/None 时 try/except 按 0 处理） +- 🛡️ **`crowd_detector.py` 根因加固** — 与测试同步，detector 主循环加入三层防御： + - `if not isinstance(rec, dict): continue` — 防止 loader 混入非 dict 元素时 `rec["name"]` 抛 TypeError + - `if not isinstance(rec_name, str) or not rec_name.strip(): continue` — 防止 name 缺失/非 str 时 `.strip()` 抛 AttributeError + - `try/except (TypeError, ValueError): freq = 0` — 防止 frequency 字段异常时 `int(...)` 崩溃 + - `_normalize_entry` 增加 truthy 非 str school 的 str() 兜底（dict / tuple / list 三分支） + - 取值统一用 `or` 兜底空值：`rec.get("platforms") or []` / `rec.get("predicted_increase") or 0` / `rec.get("alternatives") or []` +- ✅ **质量门禁** + - `crowd_detector.py` 单模块覆盖率：**93%**（101 statements / 7 missed 仅 `__main__` CLI demo 块） + - 远高于 T2.5 PRD 阈值 **≥80%** + - `data/crowd_db/tests/test_crowd_detector.py` 55/55 passed + - `data/crowd_db/tests/` 全量 125/125 passed（不动 T2.1/T2.2/T2.4 既有测试） + - `pytest -q --ignore=data/orders` 仓库全量 249 passed + - `ruff check data/crowd_db/` — All checks passed + #### 新增（T1.3 已完成） - ✨ **方案解析器 `plan_parser.py`**（`skills/gaokao-audit/scripts/plan_parser.py`，295 行） diff --git a/data/crowd_db/crowd_detector.py b/data/crowd_db/crowd_detector.py index 1055943..57752c8 100644 --- a/data/crowd_db/crowd_detector.py +++ b/data/crowd_db/crowd_detector.py @@ -54,19 +54,34 @@ def _normalize_entry(entry: PlanEntry) -> Dict[str, Any]: - dict（必须有 school，可选 major） - CrowdRecommendation（name -> school, major 保留） - tuple / list（[school, major] 或 [school]） + - 其他形态：str() 兜底 """ if isinstance(entry, dict): - return { - "school": entry.get("school") or entry.get("name") or "", - "major": entry.get("major"), - } + school_val = entry.get("school") or entry.get("name") or "" + # 防御: school 是非字符串（如 0/None）时降级为 str + if not isinstance(school_val, str): + school_val = str(school_val) if school_val else "" + return {"school": school_val, "major": entry.get("major")} if isinstance(entry, CrowdRecommendation): return {"school": entry.name, "major": entry.major} if isinstance(entry, (tuple, list)): if len(entry) >= 2: - return {"school": entry[0], "major": entry[1]} + school_val = entry[0] + return { + "school": school_val + if isinstance(school_val, str) + else str(school_val), + "major": entry[1], + } if len(entry) == 1: - return {"school": entry[0], "major": None} + school_val = entry[0] + return { + "school": school_val + if isinstance(school_val, str) + else str(school_val), + "major": None, + } + # 兜底: 任意对象 str() 化 return {"school": str(entry), "major": None} @@ -82,10 +97,23 @@ def _risk_level_from_frequency(frequency: int) -> str: def _school_matches(school_a: str, school_b: str) -> bool: - """院校名模糊匹配：任一方向包含即视为匹配。""" + """院校名匹配。 + + 规则： + - 完全相等：命中 + - 简称/全称包含：仅当较短一方长度 >= 4 时命中 + + 这样既保留“长沙民政” -> “长沙民政职业技术学院”的常用简称匹配， + 也避免“大学”“学院”“湖南”这类过短泛词产生系统性误报。 + """ + school_a = school_a.strip() + school_b = school_b.strip() if not school_a or not school_b: return False - return school_a in school_b or school_b in school_a + if school_a == school_b: + return True + shorter_len = min(len(school_a), len(school_b)) + return shorter_len >= 4 and (school_a in school_b or school_b in school_a) def _major_matches(plan_major: Optional[str], rec_major: str) -> bool: @@ -96,8 +124,7 @@ def _major_matches(plan_major: Optional[str], rec_major: str) -> bool: if not plan_major: return True if not rec_major: - # 数据库中无专业信息时退化为按院校 - return True + return False return plan_major.strip() == rec_major.strip() @@ -143,26 +170,37 @@ def detect_crowd_risk( # 3) 在该分数段 recs 中查找匹配 for rec in recs: - if not _school_matches(school, rec["name"]): + # 防御: loader 可能混入非 dict 元素（如 None / 字符串 / 数字） + if not isinstance(rec, dict): + continue + rec_name = rec.get("name") + # 防御: name 缺失或非字符串时跳过（避免 .strip() 抛错） + if not isinstance(rec_name, str) or not rec_name.strip(): + continue + if not _school_matches(school, rec_name): continue if not _major_matches(major, rec.get("major", "")): continue - freq = int(rec.get("frequency", 0)) + freq_raw = rec.get("frequency", 0) + # 防御: frequency 字段异常（非数值）时按 0 处理 + try: + freq = int(freq_raw) + except (TypeError, ValueError): + freq = 0 if freq <= 0: continue findings.append( RiskFinding( - school=rec["name"], + school=rec_name, major=rec.get("major") or major, frequency=freq, risk_level=_risk_level_from_frequency(freq), - platforms=list(rec.get("platforms", [])), - predicted_increase=int(rec.get("predicted_increase", 0)), - alternatives=list(rec.get("alternatives", [])), + platforms=list(rec.get("platforms") or []), + predicted_increase=int(rec.get("predicted_increase") or 0), + alternatives=list(rec.get("alternatives") or []), ) ) break # 一条 plan entry 命中一次即可 - # 4) 按风险等级排序（frequency 降序 → 等级高→低） findings.sort(key=lambda f: f.frequency, reverse=True) return findings diff --git a/data/crowd_db/tests/test_crowd_detector.py b/data/crowd_db/tests/test_crowd_detector.py index c0822b2..f3f418a 100644 --- a/data/crowd_db/tests/test_crowd_detector.py +++ b/data/crowd_db/tests/test_crowd_detector.py @@ -22,9 +22,9 @@ sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "..", "..")) from data.crowd_db.crowd_detector import ( detect_crowd_risk, - RiskFinding, plan_entry, ) +from data.crowd_db.loader import CrowdDBLoader # ---------- 高风险 ---------- @@ -272,3 +272,517 @@ def test_national_province_not_loaded(): findings = detect_crowd_risk(plan, user_score=600, province="全国") # national.json 缺失 → 空 assert findings == [] + + +# ---------- 归一化分支：CrowdRecommendation / tuple / list ---------- + + +def test_plan_entry_as_crowd_recommendation_dataclass(): + """plan 条目是 CrowdRecommendation 时应能归一化并命中""" + from data.crowd_db.loader import CrowdRecommendation + + plan = [ + CrowdRecommendation( + name="长沙理工大学", + major="计算机科学与技术", + frequency=4, + platforms=["千问"], + predicted_increase=18, + alternatives=[], + ) + ] + findings = detect_crowd_risk(plan, user_score=575, province="湖南") + assert len(findings) == 1 + assert findings[0].school == "长沙理工大学" + assert findings[0].risk_level == "high" + + +def test_plan_entry_as_tuple_two_elements(): + """plan 条目是 (school, major) tuple 时应能命中""" + plan = [("长沙理工大学", "计算机科学与技术")] + findings = detect_crowd_risk(plan, user_score=575, province="湖南") + assert len(findings) == 1 + assert findings[0].school == "长沙理工大学" + + +def test_plan_entry_as_tuple_single_element(): + """plan 条目是 (school,) tuple 时应按院校命中（无专业）""" + plan = [("长沙理工大学",)] + findings = detect_crowd_risk(plan, user_score=575, province="湖南") + assert len(findings) == 1 + assert findings[0].school == "长沙理工大学" + + +def test_plan_entry_as_list_two_elements(): + """plan 条目是 [school, major] list 时应能命中""" + plan = [["长沙理工大学", "计算机科学与技术"]] + findings = detect_crowd_risk(plan, user_score=575, province="湖南") + assert len(findings) == 1 + + +def test_plan_entry_dict_with_name_field(): + """plan dict 用 'name' 字段而非 'school' 时也应兼容""" + plan = [{"name": "长沙理工大学", "major": "计算机科学与技术"}] + findings = detect_crowd_risk(plan, user_score=575, province="湖南") + assert len(findings) == 1 + + +def test_plan_entry_dict_without_school_key(): + """plan dict 既无 school 也无 name 时应跳过（不报错）""" + plan = [{"major": "计算机"}, {"school": "", "major": "x"}] + findings = detect_crowd_risk(plan, user_score=575, province="湖南") + # 空 school 条目不会命中，但也不抛异常 + assert findings == [] + + +# ---------- 风险等级边界 ---------- + + +def test_risk_level_none_for_zero_frequency(): + """frequency=0 应映射为 'none'（不构成风险）""" + from data.crowd_db.crowd_detector import _risk_level_from_frequency + + assert _risk_level_from_frequency(0) == "none" + + +def test_risk_level_low_for_frequency_one(): + """frequency=1 应映射为 'low'""" + from data.crowd_db.crowd_detector import _risk_level_from_frequency + + assert _risk_level_from_frequency(1) == "low" + + +def test_school_matches_empty_strings_returns_false(): + """两个空字符串不应被误判为匹配""" + from data.crowd_db.crowd_detector import _school_matches + + assert _school_matches("", "长沙理工大学") is False + assert _school_matches("长沙理工大学", "") is False + assert _school_matches("", "") is False + + +def test_school_matches_exact_name_returns_true(): + """完全相等的院校名必须命中""" + from data.crowd_db.crowd_detector import _school_matches + + assert _school_matches("北京大学", "北京大学") is True + + +def test_school_matches_valid_abbreviation_returns_true(): + """常见 4 字简称应保留模糊命中能力""" + from data.crowd_db.crowd_detector import _school_matches + + assert _school_matches("长沙民政", "长沙民政职业技术学院") is True + + +def test_school_matches_short_generic_name_returns_false(): + """过短泛词（如“大学”）不应模糊命中具体院校""" + from data.crowd_db.crowd_detector import _school_matches + + assert _school_matches("大学", "湖南大学") is False + assert _school_matches("湖南", "湖南大学") is False + + +# ---------- 频次为 0 的记录被跳过 ---------- + + +def test_zero_frequency_record_in_db_skipped(monkeypatch): + """若 crowd_db 某条记录 frequency=0，detect 时应跳过不报告""" + loader = CrowdDBLoader() + # monkeypatch loader 返回含 freq=0 的记录 + monkeypatch.setattr( + loader, + "find_recommendations", + lambda province, score: [ + { + "name": "长沙理工大学", + "major": "计算机科学与技术", + "frequency": 0, + "platforms": [], + "predicted_increase": 0, + "alternatives": [], + } + ], + ) + plan = [plan_entry("长沙理工大学", "计算机科学与技术")] + findings = detect_crowd_risk(plan, user_score=575, province="湖南", loader=loader) + assert findings == [] + + +def test_major_specified_but_record_major_missing_no_match(monkeypatch): + """用户指定专业时，数据库专业缺失不应退化为院校命中""" + loader = CrowdDBLoader() + monkeypatch.setattr( + loader, + "find_recommendations", + lambda province, score: [ + { + "name": "长沙理工大学", + "major": "", + "frequency": 4, + "platforms": [], + "predicted_increase": 0, + "alternatives": [], + } + ], + ) + plan = [plan_entry("长沙理工大学", "会计学")] + findings = detect_crowd_risk(plan, user_score=575, province="湖南", loader=loader) + assert findings == [] + + +# ---------- 学校名/记录名匹配 edge case ---------- + + +def test_plan_school_empty_string_skipped(): + """plan 条目 school 为空字符串时应被跳过""" + plan = [ + {"school": "", "major": "x"}, + plan_entry("长沙理工大学", "计算机科学与技术"), + ] + findings = detect_crowd_risk(plan, user_score=575, province="湖南") + assert len(findings) == 1 + assert findings[0].school == "长沙理工大学" + + +# ========================================================================= +# T2.5 用例: 高风险识别 / 替代方案 / 跨省份 / 异常处理 +# ========================================================================= +# 之前的 35 个测试在算法内部路径上已经达到 91% 覆盖率；T2.5 在此基础上 +# 显式锚定任务 PRD 列出的 4 个用例，并补全: +# 1) 高风险识别 (frequency=4) +# 2) 替代方案返回 +# 3) 跨省份场景（湖南记录不污染广东结果） +# 4) 异常处理（loader 抛错/返回非法字段/未识别 entry 形态） +# ========================================================================= + + +# ---------- 用例 1: 高风险识别 ---------- + + +def test_high_risk_use_case_frequency_4_full_payload(): + """用例1: frequency=4 完整字段映射 (4家AI全推荐 → 顶级扎堆) + + PRD 要求: 给出高风险判定 + predicted_increase + platforms + alternatives + """ + # Hunan 575 段: 长沙理工大学-计算机科学与技术 freq=4 是 high + plan = [plan_entry("长沙理工大学", "计算机科学与技术")] + findings = detect_crowd_risk(plan, user_score=575, province="湖南") + + high = [ + f for f in findings if f.risk_level == "high" and f.school == "长沙理工大学" + ] + assert len(high) >= 1 + f = high[0] + assert f.frequency == 4 + assert f.risk_level == "high" + # 高风险必须给出预测涨幅（>= 0） + assert f.predicted_increase > 0 + # 高风险必须挂出全部 4 家平台 + assert set(f.platforms) == {"千问", "元宝", "百度", "豆包"} + # 至少给出 1 个替代方案 + assert len(f.alternatives) >= 1 + for alt in f.alternatives: + assert "name" in alt and alt["name"] + + +def test_high_risk_boundary_frequency_exactly_4(): + """用例1 边界: frequency 恰好等于 4 → high（>= 4 闭合区间）""" + from data.crowd_db.crowd_detector import _risk_level_from_frequency + + assert _risk_level_from_frequency(4) == "high" + + +def test_high_risk_distinct_from_medium_and_low(): + """用例1: 同一方案中高/中/低风险需互不混淆""" + plan = [ + plan_entry("中南大学", "临床医学"), # freq=4 → high + plan_entry("湖南科技大学", "机械设计制造及其自动化"), # freq=2 → medium + ] + findings = detect_crowd_risk(plan, user_score=575, province="湖南") + levels = sorted({f.risk_level for f in findings}) + assert "high" in levels + assert "medium" in levels + + +# ---------- 用例 2: 替代方案 ---------- + + +def test_alternatives_use_case_contains_required_keys(): + """用例2: 替代方案每条必须含 name + major + score 字段 + + 与 risk_report.py 渲染所需字段对齐 (alt.name, alt.major, alt.score) + """ + # Hunan 575 段: 湖南师范大学-会计学 freq=4 → high, 含 2 个替代 + plan = [plan_entry("湖南师范大学", "会计学")] + findings = detect_crowd_risk(plan, user_score=575, province="湖南") + assert len(findings) >= 1 + alts = findings[0].alternatives + assert len(alts) >= 1 + for a in alts: + assert isinstance(a, dict) + assert "name" in a and a["name"] + assert "major" in a + assert "score" in a + assert isinstance(a["score"], (int, float)) + assert 0 <= a["score"] <= 100 + + +def test_alternatives_use_case_sortable_by_score(monkeypatch): + """用例2: 替代方案应能按 score 降序使用（不强制 detector 排序） + + 验证 detector 透传 alternatives 字段、不丢/不改字段。 + """ + loader = CrowdDBLoader() + monkeypatch.setattr( + loader, + "find_recommendations", + lambda province, score: [ + { + "name": "测试大学", + "major": "测试专业", + "frequency": 4, + "platforms": ["千问", "元宝", "百度", "豆包"], + "predicted_increase": 20, + "alternatives": [ + {"name": "替代A", "major": "测试专业", "score": 80}, + {"name": "替代B", "major": "测试专业", "score": 95}, + {"name": "替代C", "major": "测试专业", "score": 88}, + ], + } + ], + ) + plan = [plan_entry("测试大学", "测试专业")] + findings = detect_crowd_risk(plan, user_score=600, province="湖南", loader=loader) + assert len(findings) == 1 + alts = findings[0].alternatives + assert [a["name"] for a in alts] == ["替代A", "替代B", "替代C"] + # 排序后最高分在最前 + alts_sorted = sorted(alts, key=lambda x: x["score"], reverse=True) + assert alts_sorted[0]["name"] == "替代B" + + +def test_alternatives_use_case_empty_list_is_valid(monkeypatch): + """用例2: 替代方案为空列表时（数据不足）不报错、不抛异常""" + loader = CrowdDBLoader() + monkeypatch.setattr( + loader, + "find_recommendations", + lambda province, score: [ + { + "name": "测试大学", + "major": "测试专业", + "frequency": 2, + "platforms": ["千问"], + "predicted_increase": 5, + "alternatives": [], + } + ], + ) + plan = [plan_entry("测试大学", "测试专业")] + findings = detect_crowd_risk(plan, user_score=600, province="湖南", loader=loader) + assert len(findings) == 1 + assert findings[0].alternatives == [] + + +# ---------- 用例 3: 跨省份 ---------- + + +def test_cross_province_hunan_hit_guangdong_miss(): + """用例3: 湖南方案在 province=广东 时不应命中（跨省数据隔离）""" + # 长沙理工大学-计算机 在湖南 575 段是 freq=4 顶级扎堆; 广东 575 段无数据 + plan = [plan_entry("长沙理工大学", "计算机科学与技术")] + hunan = detect_crowd_risk(plan, user_score=575, province="湖南") + guangdong = detect_crowd_risk(plan, user_score=575, province="广东") + assert len(hunan) >= 1 + assert guangdong == [] # 广东 575 无分数段 → 不命中 + + +def test_cross_province_beijing_at_690(monkeypatch): + """用例3: 跨省分数段差异（清华大学 在北京 690 命中"工科试验班"）""" + # Beijing 690 段: 清华大学-工科试验班 freq=4; 湖南 690 段无清华-工科试验班记录 + plan = [plan_entry("清华大学", "工科试验班")] + beijing = detect_crowd_risk(plan, user_score=690, province="北京") + hunan = detect_crowd_risk(plan, user_score=690, province="湖南") + assert len(beijing) >= 1 + # 湖南 690 段无 "工科试验班" major → 不命中（专业严格匹配） + assert hunan == [] + + +def test_cross_province_loader_called_with_correct_province(monkeypatch): + """用例3: detector 应把 province 透传给 loader，不做省份无关全局查询""" + captured = {} + + def fake_find(province, score): + captured["province"] = province + captured["score"] = score + return [] + + loader = CrowdDBLoader() + monkeypatch.setattr(loader, "find_recommendations", fake_find) + plan = [plan_entry("任意学校", "任意专业")] + detect_crowd_risk(plan, user_score=580, province="湖北", loader=loader) + assert captured["province"] == "湖北" + assert captured["score"] == 580 + + +# ---------- 用例 4: 异常处理 ---------- + + +def test_exception_use_case_loader_returns_none_field(monkeypatch): + """用例4: loader 返回的记录缺关键字段 (None) 不应崩""" + loader = CrowdDBLoader() + monkeypatch.setattr( + loader, + "find_recommendations", + lambda province, score: [ + { + "name": None, # 异常: name 为 None + "major": "测试专业", + "frequency": 3, + "platforms": [], + "predicted_increase": 0, + "alternatives": [], + } + ], + ) + plan = [plan_entry("测试大学", "测试专业")] + # 不应抛异常 + findings = detect_crowd_risk(plan, user_score=600, province="湖南", loader=loader) + assert isinstance(findings, list) + + +def test_exception_use_case_loader_returns_non_dict(monkeypatch): + """用例4: loader 返回非 dict 元素时 detector 不应崩""" + loader = CrowdDBLoader() + monkeypatch.setattr( + loader, + "find_recommendations", + lambda province, score: [ + "not a dict", + None, + 42, + { + "name": "正常大学", + "major": "正常专业", + "frequency": 2, + "platforms": ["千问"], + "predicted_increase": 5, + "alternatives": [], + }, + ], + ) + plan = [plan_entry("正常大学", "正常专业")] + findings = detect_crowd_risk(plan, user_score=600, province="湖南", loader=loader) + # 应能优雅地过滤掉非 dict, 至少命中"正常大学" + assert len(findings) >= 1 + assert any(f.school == "正常大学" for f in findings) + + +def test_exception_use_case_loader_raises_propagates(): + """用例4: loader 自身抛异常时 detector 不应静默吞错 (允许向上传播)""" + + # 这里验证: 当 loader 抛异常时, 调用方能看到 + class ExplodingLoader: + def find_recommendations(self, province, score): + raise RuntimeError("loader boom") + + plan = [plan_entry("测试大学", "测试专业")] + try: + detect_crowd_risk( + plan, user_score=600, province="湖南", loader=ExplodingLoader() + ) + raised = False + except RuntimeError as e: + raised = True + assert "loader boom" in str(e) + assert raised, "loader 异常应向上传播，不应被静默吞掉" + + +def test_exception_use_case_unrecognized_entry_type(): + """用例4: 不可识别的 plan entry 形态应走 fallback (不抛异常)""" + # 整数 / 自定义对象: 应走 str(entry) 兜底 + plan = [42, 3.14, object()] + findings = detect_crowd_risk(plan, user_score=600, province="湖南") + # 兜底 school=str(42)="42" 不在 crowd_db 中 → 无命中 + assert findings == [] + + +def test_exception_use_case_entry_with_int_school_dict(): + """用例4: dict 中 school 是非字符串 (如 0) 时不抛异常""" + plan = [{"school": 0, "major": "x"}] # 0 是 falsy + findings = detect_crowd_risk(plan, user_score=600, province="湖南") + # `entry.get("school") or entry.get("name") or ""` → 0 → "" → 跳过 + assert findings == [] + + +def test_exception_use_case_province_is_none(): + """用例4: province=None 时 detector 不应抛 (loader 会返回空)""" + plan = [plan_entry("长沙理工大学", "会计学")] + findings = detect_crowd_risk(plan, user_score=575, province=None) + # loader 对 None 省份返回 [], detector 返回空 list + assert findings == [] + + +def test_exception_use_case_tuple_with_non_string_school(): + """用例4: tuple 元组的 school 为非字符串 (如 int) 时不抛异常""" + # (0.5, "x") → school=0.5 是 truthy 但非 str, 走 str() 兜底 + plan = [(0.5, "x"), (3.14, "y")] + findings = detect_crowd_risk(plan, user_score=600, province="湖南") + # 不会命中 crowd_db (str(0.5) 不在数据中), 但也不抛异常 + assert isinstance(findings, list) + + +def test_exception_use_case_dict_with_truthy_non_string_school(): + """用例4: dict 中 school 是 truthy 非字符串 (如 0.5) 时不抛异常, 走 str() 兜底""" + # {"school": 0.5, "major": "x"} → school=0.5 truthy, 非 str → 走 str() 兜底 + plan = [{"school": 0.5, "major": "x"}] + findings = detect_crowd_risk(plan, user_score=600, province="湖南") + assert isinstance(findings, list) + + +def test_exception_use_case_dict_with_name_truthy_non_string(): + """用例4: dict 中 name 是 truthy 非字符串时也走 str() 兜底""" + # {"name": 0.5, "major": "x"} → school_val = 0.5, truthy 非 str + plan = [{"name": 0.5, "major": "x"}] + findings = detect_crowd_risk(plan, user_score=600, province="湖南") + assert isinstance(findings, list) + + +def test_exception_use_case_single_element_tuple_non_string(): + """用例4: 单元素 tuple 的 school 为非字符串时走 str() 兜底""" + # (0.5,) → len == 1, school=0.5 truthy 非 str + plan = [(0.5,)] + findings = detect_crowd_risk(plan, user_score=600, province="湖南") + assert isinstance(findings, list) + + +def test_exception_use_case_malformed_frequency_in_record(monkeypatch): + """用例4: loader 返回的 frequency 是非数值字符串/None/对象时 detector 不崩""" + loader = CrowdDBLoader() + monkeypatch.setattr( + loader, + "find_recommendations", + lambda province, score: [ + { + "name": "测试大学", + "major": "测试专业", + "frequency": "not a number", # 异常: 不可转为 int + "platforms": [], + "predicted_increase": 0, + "alternatives": [], + }, + { + "name": "测试大学B", + "major": "测试专业B", + "frequency": None, # 异常: None + "platforms": [], + "predicted_increase": 0, + "alternatives": [], + }, + ], + ) + plan = [plan_entry("测试大学", "测试专业"), plan_entry("测试大学B", "测试专业B")] + # 不应抛异常；frequency 不可解析时按 0 处理 → 跳过 + findings = detect_crowd_risk(plan, user_score=600, province="湖南", loader=loader) + assert findings == [] # freq=0 跳过 → 无命中