cma-management/backend/scripts/verify_zxt_vouchers_json.py

"""zxt vouchers-json 端点验收测试（本地 SQLite，不动生产库）"""
import os
os.environ.setdefault("CMA_DB_USER", "test")
os.environ.setdefault("CMA_DB_PASS", "test")
os.environ.setdefault("CMA_DB_HOST", "localhost")
os.environ.setdefault("CMA_DB_PORT", "3306")
os.environ.setdefault("CMA_DB_NAME", "test")

from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import StaticPool

from app import database as db_module
from app.database import Base, get_db

TEST_ENGINE = create_engine(
    "sqlite:///:memory:", connect_args={"check_same_thread": False}, poolclass=StaticPool
)
TEST_SESSION_LOCAL = sessionmaker(autocommit=False, autoflush=False, bind=TEST_ENGINE)
db_module._engine = TEST_ENGINE
db_module._SessionLocal = TEST_SESSION_LOCAL
TEST_ENGINE2 = TEST_ENGINE
TEST_SESSION2 = TEST_SESSION_LOCAL

# 建表
from app import models  # noqa
from app.models import Entity, VoucherDetail, ImportLog
Base.metadata.create_all(bind=TEST_ENGINE)

# seed entity 1 (active)
s = TEST_SESSION_LOCAL()
ent = Entity(id=1, name="陕西酣客文化传媒", status="active")
s.add(ent)
s.commit()
s.close()

from app.main import app

def override_get_db():
    db = TEST_SESSION_LOCAL()
    try:
        yield db
    finally:
        db.close()

app.dependency_overrides[get_db] = override_get_db
client = TestClient(app)

# 凭据来源：CMA_BOT_KEY_ADMIN → CMA_BOT_API_KEYS(role=ceo/admin) → 兼容期历史共享键（P3 待轮换）
_LEGACY_ADMIN_KEY = "cma-bot-admin-2026"


def _resolve_admin_key() -> str:
    key = (os.getenv("CMA_BOT_KEY_ADMIN") or "").strip()
    if key:
        return key
    raw = (os.getenv("CMA_BOT_API_KEYS") or "").strip()
    if raw:
        try:
            import json as _json
            for k, v in _json.loads(raw).items():
                if isinstance(v, dict) and v.get("role") in ("ceo", "admin"):
                    return k
        except Exception:
            pass
    return _LEGACY_ADMIN_KEY


HDR = {"X-BOT-KEY": _resolve_admin_key()}


def check(name, cond, detail=""):
    print(("PASS " if cond else "FAIL ") + name + (" | " + detail if detail else ""))
    return cond

results = []

# ── 用例1: 带 subject_map 的正常批次入库 ──
rows_ok = [
    {"Voucher_ID": "31000", "CwAccountID": "1001", "summary": "XSD-2026-06-30-00001_客户A_张三", "debit": "1000.00", "credit": "0.00"},
    {"Voucher_ID": "31000", "CwAccountID": "6001", "summary": "XSD-2026-06-30-00001_客户A_张三", "debit": "0.00", "credit": "1000.00"},
]
payload = {"rows": rows_ok, "subject_map": {
    "1001": {"subject_code": "1001", "subject_name": "库存现金"},
    "6001": {"subject_code": "6001", "subject_name": "主营业务收入"},
}}
r = client.post("/api/cma/cash/import/vouchers-json", json=payload, headers=HDR)
results.append(check("用例1 正常批次 HTTP200", r.status_code == 200, f"status={r.status_code} body={r.text[:200]}"))
j = r.json()
results.append(check("用例1 success_rows=2", j.get("success_rows") == 2, f"got {j.get('success_rows')} errors={j.get('errors')}"))
results.append(check("用例1 balance passed", j.get("balance_check", {}).get("passed") is True))
batch1 = j.get("batch")

# ── 用例2: 凭证级不平衡被拒（30694 单边场景）──
rows_unbal = [
    {"Voucher_ID": "30694", "CwAccountID": "6001", "summary": "XSD-2026-04-14-00001_客户X_李四", "debit": "0.00", "credit": "5479.65"},
]
payload2 = {"rows": rows_unbal, "subject_map": {"6001": {"subject_code": "6001", "subject_name": "主营业务收入"}}}
r2 = client.post("/api/cma/cash/import/vouchers-json", json=payload2, headers=HDR)
j2 = r2.json()
results.append(check("用例2 不平凭证 HTTP200且success_rows=0", r2.status_code == 200 and j2.get("success_rows") == 0, f"body={r2.text[:200]}"))
results.append(check("用例2 abnormal_vouchers 含30694", any(a.get("voucher_no") == "30694" for a in (j2.get("balance_check") or {}).get("abnormal_vouchers", [])), r2.text[:200]))

# ── 用例3: 科目字典未命中 → 拒收提示（无 subject_map 时 2000行全 null 场景）──
rows_nodict = [
    {"Voucher_ID": "31001", "CwAccountID": "401", "summary": "期间损益结转", "debit": "0.00", "credit": "35414.00"},
]
r3 = client.post("/api/cma/cash/import/vouchers-json", json={"rows": rows_nodict}, headers=HDR)
j3 = r3.json()
results.append(check("用例3 字典未命中 拒收0行+提示", r3.status_code == 200 and j3.get("success_rows") == 0 and "科目字典" in (j3.get("hint") or ""), f"body={r3.text[:250]}"))

# ── 用例4: 幂等 — 同内容重复POST不重复入库 ──
r4 = client.post("/api/cma/cash/import/vouchers-json", json=payload, headers=HDR)
j4 = r4.json()
results.append(check("用例4 同指纹重复导入 already_imported", r4.status_code == 200 and j4.get("already_imported") is True, f"body={r4.text[:200]}"))

# ── 用例4b: 同内容不同「模拟时间」仍同 batch → 幂等（batch 不含时间戳）──
def j4b_is_already(resp, j4):
    j = resp.json()
    return j.get("already_imported") is True and j.get("batch") == j4.get("batch")
r4b = client.post("/api/cma/cash/import/vouchers-json", json=payload, headers=HDR)
results.append(check("用例4b 再次重发仍 already_imported 且 batch 不变", r4b.status_code == 200 and j4b_is_already(r4b, j4), f"body={r4b.text[:200]}"))

# ── 用例5: 无 X-BOT-KEY → 401 ──
r5 = client.post("/api/cma/cash/import/vouchers-json", json=payload)
results.append(check("用例5 无X-BOT-KEY 401", r5.status_code == 401, f"status={r5.status_code}"))

# ── 用例6: DB 回查 voucher_details ──
s = TEST_SESSION_LOCAL()
cnt = s.query(VoucherDetail).filter(VoucherDetail.batch == batch1).count()
results.append(check("用例6 DB回查 batch行数=2", cnt == 2, f"cnt={cnt}"))
il = s.query(ImportLog).filter(ImportLog.batch == batch1, ImportLog.import_type == "vouchers-json").count()
results.append(check("用例6b import_logs 有记录", il == 1, f"cnt={il}"))
s.close()

# ── 用例7: 期间损益结转行（summary无日期）→ 拒收记 pending_undated ──
rows_undated = [
    {"Voucher_ID": "31454", "CwAccountID": "401", "summary": "期间损益结转", "debit": "0.00", "credit": "35414.00"},
]
r7 = client.post("/api/cma/cash/import/vouchers-json", json={"rows": rows_undated, "subject_map": {"401": {"subject_code": "4103", "subject_name": "本年利润"}}}, headers=HDR)
j7 = r7.json()
results.append(check("用例7 无日期行拒收+计数", r7.status_code == 200 and j7.get("success_rows") == 0 and j7.get("pending_undated", 0) >= 1, f"body={r7.text[:200]}"))

# ── 用例8: 现有 Excel 端点 /import/template 仍可用（回归）──
r8 = client.get("/api/cma/cash/import/template")
results.append(check("用例8 Excel模板端点回归 有鉴权响应", r8.status_code in (200, 403, 401), f"status={r8.status_code}(403=需用户token，原Excel端点未受影响)"))

fails = [x for x in results if not x]
print(f"\n==== {len(results)-len(fails)}/{len(results)} PASS ====")
if fails:
    print("FAILED CASES PRESENT")
    raise SystemExit(1)
print("ALL PASS")