"""外部输入台账 单测 — 3 张外部输入表 + 录入接口 + 台账→kpi_values 派生 （任务 0-P0-project-20260915-cma-eval-m2-ext-input-tables，2026-09-15） 覆盖任务单「验收标准」中的硬性项： ① on_time 由服务端派生（传入值被忽略/拒绝） ② 人事台账「期末 ≠ 期初 + 入职 − 离职」被拒 ③ 已过期证照不计入有效数 ④ 派生函数：**分母为 0 返回 None**（不写 0、不填大数）+ 3 个指标各 1 个正例 ⑤ 幂等：台账导入与派生各跑两次，第二次输出/行数与第一次一致 ⑥ 多租户：跨账套显式 entity_id → 403 另覆盖：CSV 批量导入、按 (entity_id, period) 查询、唯一键冲突 409、季频期间归集。 """ import os import sys from datetime import date from types import SimpleNamespace sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) import pytest # noqa: E402 from app.models import KPIDefinition, KPIValue, HrMonthlyLedger, LicenceLedger, TaxFilingLog # noqa: E402 from app.services import ext_input_derive as D # noqa: E402 from tests.conftest import auth_header, create_test_kpi, create_test_user, get_token_for_user # noqa: E402 TAX_CSV = """period,tax_type,legal_deadline,filed_date,evidence_no 2026-08,增值税,2026-08-15,2026-08-12,HZ-001 2026-08,附加税,2026-08-15,2026-08-12,HZ-002 2026-08,印花税,2026-08-15,2026-08-20,HZ-003 """ HR_CSV = """period,headcount_begin,hires,separations,headcount_end 2026-08,100,1,2,99 2026-07,98,2,0,100 """ LIC_CSV = """licence_name,licence_no,valid_from,valid_to,annual_check_status,issuer 营业执照,91610100MA0001,2020-01-01,2030-01-01,已年检,市场监管局 食品经营许可证,JY1610100001,2024-01-01,2026-03-31,过期,市场监管局 """ def _rows(db, model, **filters): q = db.query(model) for k, v in filters.items(): q = q.filter(getattr(model, k) == v) return q.all() # ============================================================ # 纯函数：on_time 派生 / 分母为 0 / 三个指标正例 # ============================================================ def test_on_time_is_server_derived(): """on_time 唯一规则：按时=1 / 逾期=0 / 未申报=0 / 缺法定期限且已申报=无法判定(None)""" assert D.derive_on_time(date(2026, 8, 12), date(2026, 8, 15)) == 1 assert D.derive_on_time(date(2026, 8, 15), date(2026, 8, 15)) == 1 assert D.derive_on_time(date(2026, 8, 20), date(2026, 8, 15)) == 0 assert D.derive_on_time(None, date(2026, 8, 15)) == 0 # 未申报 assert D.derive_on_time(date(2026, 8, 12), None) is None # 无法判定 def test_denominator_zero_returns_none(): """硬规则：分母为 0 → None（无定义/不参评），不得写 0、不得填大数""" assert D.compute_tax_timely_rate([]) is None assert D.compute_employee_turnover(SimpleNamespace(headcount_begin=0, separations=3, headcount_end=0, hires=0)) is None assert D.compute_licence_ok([], date(2026, 6, 30)) is None assert D.compute_employee_turnover(None) is None # explain_* 同源，no_definition 同步为 True assert D.explain_tax_timely_rate([])["no_definition"] is True assert D.explain_employee_turnover(SimpleNamespace(headcount_begin=0, separations=0, headcount_end=0, hires=0))["no_definition"] is True assert D.explain_licence_ok([], date(2026, 6, 30))["no_definition"] is True def test_tax_timely_rate_positive(): """正例：3 个应申报税种，2 个按时 → 66.7%（未知 on_time 不计入分子，仍计入分母）""" rows = [SimpleNamespace(on_time=1), SimpleNamespace(on_time=1), SimpleNamespace(on_time=0)] assert D.compute_tax_timely_rate(rows) == 66.7 rows2 = [SimpleNamespace(on_time=1), SimpleNamespace(on_time=None)] assert D.compute_tax_timely_rate(rows2) == 50.0 def test_employee_turnover_positive(): """正例：期初 100 离职 2 → 2.0%；并列返回期初期末平均在职口径（99 人 → 2.02%）""" row = SimpleNamespace(headcount_begin=100, headcount_end=98, hires=1, separations=2) assert D.compute_employee_turnover(row) == 2.0 detail = D.explain_employee_turnover(row) assert detail["avg_headcount"] == 99.0 assert detail["value_on_avg"] == 2.02 def test_licence_ok_excludes_expired(): """正例：4 张证照，1 张过期 + 1 张已到期 → 有效 2 张 = 50.0%（valid_to 空=长期有效）""" as_of = date(2026, 6, 30) rows = [ SimpleNamespace(annual_check_status="已年检", valid_to=date(2030, 1, 1), licence_no="A"), SimpleNamespace(annual_check_status="过期", valid_to=date(2030, 1, 1), licence_no="B"), SimpleNamespace(annual_check_status="未到期", valid_to=date(2026, 3, 31), licence_no="C"), SimpleNamespace(annual_check_status="未到期", valid_to=None, licence_no="D"), ] assert D.compute_licence_ok(rows, as_of) == 50.0 detail = D.explain_licence_ok(rows, as_of) assert detail["valid"] == 2 and detail["denominator"] == 4 assert detail["expired_or_outdated"] == ["B"] def test_headcount_equation_check(): assert D.check_headcount_equation(100, 1, 2, 99) is None msg = D.check_headcount_equation(100, 2, 1, 99) assert msg and "期末(99)" in msg and "101" in msg def test_quarter_helpers(): assert D.month_to_quarter("2026-06") == "2026-Q2" assert D.month_to_quarter("2026-01") == "2026-Q1" assert D.quarter_end_date("2026-Q2") == date(2026, 6, 30) assert D.quarter_end_date("2026-06") == date(2026, 6, 30) assert D.quarter_end_date("2026-Q4") == date(2026, 12, 31) # ============================================================ # API：新增 / on_time 忽略 / 唯一键 409 / 校验拒绝 # ============================================================ def test_create_tax_filing_ignores_client_on_time(client, db): """① on_time 服务端派生：请求里带 on_time=1 但 filed_date 晚于法定期限 → 落库 0 + 回报被忽略""" create_test_user(db) token = get_token_for_user(client) resp = client.post("/api/cma/ext-ledger/tax-filings", headers=auth_header(token), json={ "period": "2026-08", "tax_type": "增值税", "legal_deadline": "2026-08-15", "filed_date": "2026-08-20", "on_time": 1, "evidence_no": "HZ-001", }) assert resp.status_code == 201, resp.text body = resp.json() assert body["data"]["on_time"] == 0 # 服务端派生（未被手填值污染） assert body["on_time_derived"] == 0 assert body["on_time_ignored"] is True row = _rows(db, TaxFilingLog, period="2026-08")[0] assert (row.on_time, row.entity_id) == (0, 1) def test_create_tax_filing_duplicate_409(client, db): """唯一键 (entity_id, period, tax_type) 冲突 → 409，不产生第二行""" create_test_user(db) token = get_token_for_user(client) payload = {"period": "2026-08", "tax_type": "增值税", "legal_deadline": "2026-08-15", "filed_date": "2026-08-12"} assert client.post("/api/cma/ext-ledger/tax-filings", headers=auth_header(token), json=payload).status_code == 201 dup = client.post("/api/cma/ext-ledger/tax-filings", headers=auth_header(token), json=payload) assert dup.status_code == 409 assert "已存在" in dup.json()["detail"] assert len(_rows(db, TaxFilingLog)) == 1 def test_hr_equation_rejected(client, db): """② 期末 ≠ 期初 + 入职 − 离职 → 400 拒绝写入（台账零新增）""" create_test_user(db) token = get_token_for_user(client) resp = client.post("/api/cma/ext-ledger/hr-monthly", headers=auth_header(token), json={ "period": "2026-08", "headcount_begin": 100, "hires": 2, "separations": 1, "headcount_end": 99, }) assert resp.status_code == 400 assert "等式不成立" in resp.json()["detail"] assert len(_rows(db, HrMonthlyLedger)) == 0 ok = client.post("/api/cma/ext-ledger/hr-monthly", headers=auth_header(token), json={ "period": "2026-08", "headcount_begin": 100, "hires": 2, "separations": 1, "headcount_end": 101, }) assert ok.status_code == 201 and ok.json()["data"]["turnover_rate"] == 1.0 def test_licence_invalid_status_rejected(client, db): """证照年检状态白名单校验：非法值 → 400（枚举来自 PRD §4.0）""" create_test_user(db) token = get_token_for_user(client) bad = client.post("/api/cma/ext-ledger/licences", headers=auth_header(token), json={ "licence_name": "营业执照", "licence_no": "X1", "annual_check_status": "有效"}) assert bad.status_code == 400 and "annual_check_status" in bad.json()["detail"] assert len(_rows(db, LicenceLedger)) == 0 def test_cross_entity_403(client, db): """⑥ 多租户：会话账套 entity 1 + 显式 entity_id=2 → 403（文案同 deps.get_entity_id）""" create_test_user(db) token = get_token_for_user(client) resp = client.post("/api/cma/ext-ledger/hr-monthly?entity_id=2", headers=auth_header(token), json={ "period": "2026-08", "headcount_begin": 10, "hires": 0, "separations": 0, "headcount_end": 10}) assert resp.status_code == 403 assert "无权访问企业 entity_id=2" in resp.json()["detail"] assert len(_rows(db, HrMonthlyLedger)) == 0 def test_unauthenticated_401(client): """未认证 → 401（router 级 require_auth）""" assert client.get("/api/cma/ext-ledger/options").status_code == 401 assert client.post("/api/cma/ext-ledger/derive", json={"period": "2026-08"}).status_code == 401 # ============================================================ # CSV 批量导入 + 查询 + 幂等 # ============================================================ def test_csv_import_tax_idempotent(client, db): """CSV 导入两次：第一次 inserted=3，第二次 inserted=0/skipped=3，行数不变""" create_test_user(db) token = get_token_for_user(client) r1 = client.post("/api/cma/ext-ledger/tax-filings/import", headers=auth_header(token), json={"csv": TAX_CSV}) assert r1.status_code == 200, r1.text assert (r1.json()["inserted"], r1.json()["skipped"]) == (3, 0) r2 = client.post("/api/cma/ext-ledger/tax-filings/import", headers=auth_header(token), json={"csv": TAX_CSV}) assert (r2.json()["inserted"], r2.json()["skipped"]) == (0, 3) rows = _rows(db, TaxFilingLog, period="2026-08") assert len(rows) == 3 assert {r.on_time for r in rows} == {1, 0} # 印花税 08-20 > 08-15 → 0 # 按 (entity_id, period) 查询 + 及时率摘要（2/3 = 66.7） q = client.get("/api/cma/ext-ledger/tax-filings?period=2026-08", headers=auth_header(token)) assert q.status_code == 200 assert q.json()["total"] == 3 assert q.json()["summary"]["tax_timely_rate"] == 66.7 def test_csv_import_hr_reports_bad_rows(client, db): """批量导入：等式不成立的行被拒绝并在 errors 回报，合法行照常写入""" create_test_user(db) token = get_token_for_user(client) bad_csv = HR_CSV + "2026-06,50,3,1,51\n" resp = client.post("/api/cma/ext-ledger/hr-monthly/import", headers=auth_header(token), json={"csv": bad_csv}) assert resp.status_code == 200 body = resp.json() assert body["inserted"] == 2 and body["error_count"] == 1 assert "等式不成立" in body["errors"][0]["reason"] assert body["errors"][0]["line"] == 4 assert len(_rows(db, HrMonthlyLedger)) == 2 q = client.get("/api/cma/ext-ledger/hr-monthly?period=2026-08", headers=auth_header(token)) assert q.json()["total"] == 1 and q.json()["data"][0]["turnover_rate"] == 2.0 def test_csv_import_dry_run_writes_nothing(client, db): create_test_user(db) token = get_token_for_user(client) resp = client.post("/api/cma/ext-ledger/licences/import", headers=auth_header(token), json={"csv": LIC_CSV, "dry_run": True}) assert resp.status_code == 200 and resp.json()["inserted"] == 2 assert len(_rows(db, LicenceLedger)) == 0 def test_licences_query_valid_only(client, db): """③ 已过期证照不计入有效数（查询侧 valid_only 与摘要口径一致）""" create_test_user(db) token = get_token_for_user(client) assert client.post("/api/cma/ext-ledger/licences/import", headers=auth_header(token), json={"csv": LIC_CSV}).json()["inserted"] == 2 allq = client.get("/api/cma/ext-ledger/licences?as_of=2026-06-30", headers=auth_header(token)).json() assert allq["total"] == 2 assert allq["summary"]["licence_ok"] == 50.0 # 1/2 assert allq["summary"]["valid"] == 1 only = client.get("/api/cma/ext-ledger/licences?as_of=2026-06-30&valid_only=true", headers=auth_header(token)).json() assert only["total"] == 1 and only["data"][0]["licence_no"] == "91610100MA0001" def test_options_endpoint_lists_enums(client, db): """表单下拉选项从系统数据动态取（禁前端硬编码）""" create_test_user(db) token = get_token_for_user(client) body = client.get("/api/cma/ext-ledger/options", headers=auth_header(token)).json() assert body["annual_check_status"] == ["未到期", "已年检", "待年检", "过期"] assert set(body["derivable_kpi_codes"]) == {"P_TAX_TIMELY", "L_EMPLOYEE_TURNOVER", "P_LICENCE_OK"} # ============================================================ # 派生：幂等 / 分母 0 不写 / 季频归集 / 定义缺失跳过 # ============================================================ def _seed_defs(db, entity_id=1): return { "P_TAX_TIMELY": create_test_kpi(db, entity_id=entity_id, kpi_code="P_TAX_TIMELY", kpi_name="税务申报及时率"), "L_EMPLOYEE_TURNOVER": create_test_kpi(db, entity_id=entity_id, kpi_code="L_EMPLOYEE_TURNOVER", kpi_name="员工流失率"), "P_LICENCE_OK": create_test_kpi(db, entity_id=entity_id, kpi_code="P_LICENCE_OK", kpi_name="经营许可合规率"), } def test_derive_idempotent_and_denominator_zero(client, db): """④⑤ 派生跑两次：kpi_values 行数不变（第二次全 upsert）；期初为 0 → 无定义不写值""" create_test_user(db) token = get_token_for_user(client) _seed_defs(db) # 台账：税务 2 条（1 按时）；人事期初 0（→ 无定义）；证照 2 条（1 过期） client.post("/api/cma/ext-ledger/tax-filings/import", headers=auth_header(token), json={ "csv": "period,tax_type,legal_deadline,filed_date\n" "2026-08,增值税,2026-08-15,2026-08-12\n" "2026-08,附加税,2026-08-15,2026-08-20\n"}) client.post("/api/cma/ext-ledger/hr-monthly/import", headers=auth_header(token), json={ "csv": "period,headcount_begin,hires,separations,headcount_end\n2026-08,0,0,0,0\n"}) assert client.post("/api/cma/ext-ledger/licences/import", headers=auth_header(token), json={"csv": LIC_CSV}).json()["inserted"] == 2 first = client.post("/api/cma/ext-ledger/derive", headers=auth_header(token), json={"period": "2026-08"}).json() assert first["items"]["P_TAX_TIMELY"]["actual_value"] == 50.0 assert first["items"]["P_TAX_TIMELY"]["action"] == "insert" assert first["items"]["L_EMPLOYEE_TURNOVER"]["actual_value"] is None assert first["items"]["L_EMPLOYEE_TURNOVER"]["action"] == "skipped_denominator_zero" assert first["items"]["P_LICENCE_OK"]["actual_value"] == 50.0 assert first["items"]["P_LICENCE_OK"]["period"] == "2026-Q3" # 2026-08 → Q3 n1 = db.query(KPIValue).count() assert n1 == 2 # 无定义指标零写入 second = client.post("/api/cma/ext-ledger/derive", headers=auth_header(token), json={"period": "2026-08"}).json() assert second["items"]["P_TAX_TIMELY"]["action"] == "upsert" assert second["items"]["P_LICENCE_OK"]["action"] == "upsert" assert db.query(KPIValue).count() == n1 # 幂等：行数不变 tax_row = _rows(db, KPIValue, period="2026-08")[0] assert (tax_row.actual_value, tax_row.source_type, tax_row.entity_id) == (50.0, "derived", 1) def test_derive_skips_missing_definition(client, db): """不新建/不改动 KPI 定义：定义缺失 → skipped_no_definition（不写值）""" create_test_user(db) token = get_token_for_user(client) create_test_kpi(db, entity_id=1, kpi_code="P_TAX_TIMELY") client.post("/api/cma/ext-ledger/tax-filings/import", headers=auth_header(token), json={ "csv": "period,tax_type,legal_deadline,filed_date\n2026-08,增值税,2026-08-15,2026-08-12\n"}) body = client.post("/api/cma/ext-ledger/derive", headers=auth_header(token), json={"period": "2026-08"}).json() assert body["items"]["P_TAX_TIMELY"]["action"] == "insert" assert body["items"]["L_EMPLOYEE_TURNOVER"]["action"] == "skipped_no_definition" assert body["items"]["P_LICENCE_OK"]["action"] == "skipped_no_definition" assert db.query(KPIValue).count() == 1 assert db.query(KPIDefinition).count() == 1 # 定义数量未变（本模块不新建定义） def test_derive_dry_run_writes_nothing(client, db): create_test_user(db) token = get_token_for_user(client) _seed_defs(db) client.post("/api/cma/ext-ledger/tax-filings/import", headers=auth_header(token), json={ "csv": "period,tax_type,legal_deadline,filed_date\n2026-08,增值税,2026-08-15,2026-08-12\n"}) body = client.post("/api/cma/ext-ledger/derive", headers=auth_header(token), json={"period": "2026-08", "kpi_codes": ["P_TAX_TIMELY"], "dry_run": True}).json() assert body["items"]["P_TAX_TIMELY"]["action"] == "dry-run" assert db.query(KPIValue).count() == 0 def test_derive_status_readonly(client, db): create_test_user(db) token = get_token_for_user(client) _seed_defs(db) body = client.get("/api/cma/ext-ledger/derive/status?period=2026-06", headers=auth_header(token)).json() assert body["items"]["P_LICENCE_OK"]["target_period"] == "2026-Q2" assert body["items"]["P_LICENCE_OK"]["latest_derived"] is None