Hermes CI Fix
9d9a79545a
3 new tables + idempotent DDL migration (prod guard: refuses DB name cma without approval): tax_filing_log (P_TAX_TIMELY), hr_monthly_ledger (L_EMPLOYEE_TURNOVER), licence_ledger (P_LICENCE_OK). Entry API /api/cma/ext-ledger: create / CSV batch import / query by entity+period / derive trigger. on_time is server-derived only; HR rows rejected when end != begin + hires - separations. Deriver writes kpi_values only (source_type=derived, idempotent upsert); denominator 0 returns None and writes nothing; never creates or changes kpi_definitions. Tests: 22 new pytest cases (22 passed); scratch-DB e2e battery 28 assertions PASS with zero production write.
365 lines
18 KiB
Python
365 lines
18 KiB
Python
"""外部输入台账 单测 — 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
|