Hermes CI Fix
fdf805ab8f
No production code changed. Root causes were all test-side staleness: - test_cash TestVoucherImport: fixture filename contained 'test' -> rejected by the intentional import guard (d71c7fe); renamed, and missing-columns case now asserts the column check is what returns 400 - test_budget_tech_improve TestP1ValueCollect: seeded voucher_details source rows; collector deliberately skips empty aggregates sinced71c7fe(no more implicit 0), and preview call now passes an explicit period - test_roadmap_r2r5 / test_budget / test_ai_suggestions: seeds hardcoded 2026-08 (authoring month) while detectors/rule engine use the current month -> replaced with current-period seeds pytest tests/ -q -rf: 1094 passed, 1 xfailed, 0 failed (baseline was 10 failed, 1084 passed, 1 xfailed)
321 lines
14 KiB
Python
321 lines
14 KiB
Python
"""现金流模块测试 — 收付款计划 + 资金缺口预测 + 看板 + 网银流水导入
|
||
|
||
覆盖 cash.py 核心端点：
|
||
gap-forecast / balance GET+POST / plans CRUD / plans{id}/complete /
|
||
upcoming / dashboard / check-alerts / alerts/status / import/vouchers
|
||
"""
|
||
import io
|
||
from datetime import datetime, timedelta
|
||
|
||
import pandas as pd
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.models import CashPlan
|
||
from tests.conftest import create_test_user, get_token_for_user, auth_header
|
||
|
||
BASE = "/api/cma/cash"
|
||
|
||
|
||
def _future_date(days: int = 5) -> str:
|
||
return (datetime.now() + timedelta(days=days)).strftime("%Y-%m-%d")
|
||
|
||
|
||
def _past_date(days: int = 5) -> str:
|
||
return (datetime.now() - timedelta(days=days)).strftime("%Y-%m-%d")
|
||
|
||
|
||
class TestBalance:
|
||
def test_get_balance_default(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
resp = client.get(f"{BASE}/balance", headers=auth_header(token))
|
||
assert resp.status_code == 200
|
||
assert resp.json()["entity_id"] == 1
|
||
|
||
def test_set_balance(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
resp = client.post(f"{BASE}/balance", headers=auth_header(token),
|
||
json={"current_cash": 88.5})
|
||
assert resp.status_code == 200
|
||
assert resp.json()["current_cash"] == 88.5
|
||
|
||
get_resp = client.get(f"{BASE}/balance", headers=auth_header(token))
|
||
assert get_resp.json()["current_cash"] == 88.5
|
||
|
||
def test_set_balance_negative(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
resp = client.post(f"{BASE}/balance", headers=auth_header(token),
|
||
json={"current_cash": -5})
|
||
assert resp.status_code == 400
|
||
|
||
|
||
class TestPlans:
|
||
def test_create_plan(self, client: TestClient, db: Session):
|
||
"""创建收款计划"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
resp = client.post(f"{BASE}/plans", headers=auth_header(token), json={
|
||
"plan_type": "receive", "amount": 50.0, "plan_date": _future_date(),
|
||
"counterparty": "客户A", "owner": "张三",
|
||
})
|
||
assert resp.status_code == 200
|
||
data = resp.json()["data"]
|
||
assert data["plan_type"] == "receive"
|
||
assert data["status"] == "pending"
|
||
|
||
def test_create_plan_invalid_type(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
resp = client.post(f"{BASE}/plans", headers=auth_header(token),
|
||
json={"plan_type": "weird", "amount": 10, "plan_date": _future_date()})
|
||
assert resp.status_code == 400
|
||
|
||
def test_create_plan_zero_amount(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
resp = client.post(f"{BASE}/plans", headers=auth_header(token),
|
||
json={"plan_type": "pay", "amount": 0, "plan_date": _future_date()})
|
||
assert resp.status_code == 400
|
||
|
||
def test_create_plan_missing_date(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
resp = client.post(f"{BASE}/plans", headers=auth_header(token),
|
||
json={"plan_type": "pay", "amount": 10})
|
||
assert resp.status_code == 400
|
||
|
||
def test_create_plan_bad_date(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
resp = client.post(f"{BASE}/plans", headers=auth_header(token),
|
||
json={"plan_type": "pay", "amount": 10, "plan_date": "not-a-date"})
|
||
assert resp.status_code == 400
|
||
|
||
def test_list_plans_filters(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
client.post(f"{BASE}/plans", headers=auth_header(token),
|
||
json={"plan_type": "receive", "amount": 10, "plan_date": _future_date()})
|
||
client.post(f"{BASE}/plans", headers=auth_header(token),
|
||
json={"plan_type": "pay", "amount": 20, "plan_date": _future_date()})
|
||
|
||
resp = client.get(f"{BASE}/plans?plan_type=receive", headers=auth_header(token))
|
||
assert resp.status_code == 200
|
||
assert resp.json()["total"] == 1
|
||
assert resp.json()["data"][0]["plan_type"] == "receive"
|
||
|
||
def test_list_plans_bad_month(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
resp = client.get(f"{BASE}/plans?month=bad", headers=auth_header(token))
|
||
assert resp.status_code == 400
|
||
|
||
def test_update_plan(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
pid = client.post(f"{BASE}/plans", headers=auth_header(token),
|
||
json={"plan_type": "receive", "amount": 10, "plan_date": _future_date()}
|
||
).json()["data"]["id"]
|
||
resp = client.put(f"{BASE}/plans/{pid}", headers=auth_header(token),
|
||
json={"amount": 30, "counterparty": "客户B"})
|
||
assert resp.status_code == 200
|
||
assert resp.json()["data"]["amount"] == 30.0
|
||
|
||
def test_update_plan_not_found(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
resp = client.put(f"{BASE}/plans/99999", headers=auth_header(token), json={"amount": 10})
|
||
assert resp.status_code == 404
|
||
|
||
def test_delete_plan(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
pid = client.post(f"{BASE}/plans", headers=auth_header(token),
|
||
json={"plan_type": "pay", "amount": 10, "plan_date": _future_date()}
|
||
).json()["data"]["id"]
|
||
resp = client.delete(f"{BASE}/plans/{pid}", headers=auth_header(token))
|
||
assert resp.status_code == 200
|
||
assert resp.json()["message"] == "计划已删除"
|
||
|
||
def test_complete_plan(self, client: TestClient, db: Session):
|
||
"""标记收款完成"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
pid = client.post(f"{BASE}/plans", headers=auth_header(token),
|
||
json={"plan_type": "receive", "amount": 10, "plan_date": _future_date()}
|
||
).json()["data"]["id"]
|
||
resp = client.post(f"{BASE}/plans/{pid}/complete", headers=auth_header(token))
|
||
assert resp.status_code == 200
|
||
data = resp.json()["data"]
|
||
assert data["status"] == "completed"
|
||
assert data["paid_amount"] == 10.0
|
||
|
||
def test_complete_plan_not_found(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
resp = client.post(f"{BASE}/plans/99999/complete", headers=auth_header(token))
|
||
assert resp.status_code == 404
|
||
|
||
|
||
class TestUpcomingDashboard:
|
||
def test_upcoming(self, client: TestClient, db: Session):
|
||
"""到期提醒 + 逾期"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
client.post(f"{BASE}/plans", headers=auth_header(token),
|
||
json={"plan_type": "receive", "amount": 10, "plan_date": _future_date(3)})
|
||
client.post(f"{BASE}/plans", headers=auth_header(token),
|
||
json={"plan_type": "pay", "amount": 20, "plan_date": _past_date(3)})
|
||
resp = client.get(f"{BASE}/upcoming?days=7", headers=auth_header(token))
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert len(data["upcoming"]) == 1
|
||
assert len(data["overdue"]) == 1
|
||
assert data["overdue_receive_amount"] == 0.0
|
||
assert data["overdue_pay_amount"] == 20.0
|
||
|
||
def test_dashboard(self, client: TestClient, db: Session):
|
||
"""资金看板"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
client.post(f"{BASE}/plans", headers=auth_header(token),
|
||
json={"plan_type": "receive", "amount": 10, "plan_date": _future_date(3)})
|
||
resp = client.get(f"{BASE}/dashboard", headers=auth_header(token))
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert "calendar" in data
|
||
|
||
def test_dashboard_bad_month(self, client: TestClient, db: Session):
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
resp = client.get(f"{BASE}/dashboard?month=bad", headers=auth_header(token))
|
||
assert resp.status_code == 400
|
||
|
||
def test_gap_forecast(self, client: TestClient, db: Session):
|
||
"""资金缺口预测"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
resp = client.get(f"{BASE}/gap-forecast?days=10¤t_cash=100",
|
||
headers=auth_header(token))
|
||
assert resp.status_code == 200
|
||
|
||
|
||
class TestAlerts:
|
||
def test_check_alerts(self, client: TestClient, db: Session):
|
||
"""触发资金预警检查"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
resp = client.post(f"{BASE}/check-alerts", headers=auth_header(token), json={})
|
||
assert resp.status_code == 200
|
||
|
||
def test_alerts_status(self, client: TestClient, db: Session):
|
||
"""预警状态查询"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
resp = client.get(f"{BASE}/alerts/status", headers=auth_header(token))
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert "entity_id" in data and "critical_line" in data
|
||
|
||
|
||
class TestVoucherImport:
|
||
"""网银流水导入 — 三校验规则（借贷平衡/期间合计/结转行）+ 入库 + 现金流联动"""
|
||
|
||
COLS = ["凭证日期", "凭证号", "科目编码", "科目名称", "借方金额", "贷方金额", "摘要"]
|
||
|
||
def _xlsx(self, rows: list, cols: list = None) -> io.BytesIO:
|
||
df = pd.DataFrame(rows, columns=cols or self.COLS)
|
||
buf = io.BytesIO()
|
||
df.to_excel(buf, index=False)
|
||
buf.seek(0)
|
||
return buf
|
||
|
||
def _upload(self, client, token, buf, fname="vouchers_import.xlsx"):
|
||
"""注意: 文件名不得含 test/示例/sample/demo 等测试标识——导入端点有生产防线
|
||
（d71c7fe「Excel导入示例文件防线」），含标识直接 400，与三条校验规则的用例无关。"""
|
||
return client.post(
|
||
f"{BASE}/import/vouchers", headers=auth_header(token),
|
||
files={"file": (fname, buf,
|
||
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet")},
|
||
)
|
||
|
||
def test_import_ok_with_rules(self, client: TestClient, db: Session):
|
||
"""正常导入：借贷平衡+结转行识别+期间合计，全部成功"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
buf = self._xlsx([
|
||
["2026-08-01", "记-001", "1002", "银行存款-工行", 50000, 0, "收到客户回款"],
|
||
["2026-08-02", "记-001", "1001", "库存现金", 0, 50000, "提现备用"],
|
||
["2026-08-31", "记-099", "4103", "本年利润", 2000, 0, "结转利润"],
|
||
["2026-08-31", "记-099", "6001", "主营业务收入", 0, 2000, "结转收入"],
|
||
])
|
||
resp = self._upload(client, token, buf)
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["success"] is True
|
||
assert data["total"] == 4
|
||
assert data["success_rows"] == 4
|
||
assert data["failed_rows"] == 0
|
||
assert data["balance_check"]["passed"] is True
|
||
assert data["balance_check"]["debit_total"] == 52000
|
||
assert data["carry_forward_count"] == 2 # 结转行识别（本年利润+结转摘要）
|
||
assert "2026-08" in data["period_totals"]
|
||
assert data["cash_balance"] == 0.0 # 货币资金联动（50000-50000=0万元）
|
||
# 入库验证
|
||
from app.models import VoucherDetail, ImportLog
|
||
details = db.query(VoucherDetail).all()
|
||
assert len(details) == 4
|
||
assert all(d.entity_id == 1 for d in details)
|
||
cf = [d for d in details if d.carry_forward == 1]
|
||
assert len(cf) == 2 and all("结转" in (d.summary or "") for d in cf)
|
||
log = db.query(ImportLog).order_by(ImportLog.id.desc()).first()
|
||
assert log is not None and log.import_type == "vouchers"
|
||
assert log.success_rows == 4 and log.total_rows == 4
|
||
|
||
def test_import_unbalanced(self, client: TestClient, db: Session):
|
||
"""借贷不平衡：balance_check.passed=False + 差额报告，行仍入库"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
buf = self._xlsx([
|
||
["2026-08-01", "记-001", "1002", "银行存款-工行", 10000, 0, "回款"],
|
||
["2026-08-01", "记-002", "1002", "银行存款-工行", 0, 3000, "付款"],
|
||
])
|
||
resp = self._upload(client, token, buf)
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["balance_check"]["passed"] is False
|
||
assert data["balance_check"]["diff"] == 7000
|
||
|
||
def test_import_partial_fail(self, client: TestClient, db: Session):
|
||
"""部分失败模式：坏行进errors，好行入库"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
buf = self._xlsx([
|
||
["2026-08-01", "记-001", "1002", "银行存款-工行", 8000, 0, "回款"],
|
||
["bad-date", "记-002", "1001", "库存现金", 0, 8000, "提现"],
|
||
["2026-08-01", "", "1001", "库存现金", 100, 0, "缺凭证号"],
|
||
])
|
||
resp = self._upload(client, token, buf)
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert data["success_rows"] == 1
|
||
assert data["failed_rows"] == 2
|
||
# 行级错误2条 + 借贷平衡全局错误1条（仅8000借无贷）
|
||
assert len(data["errors"]) == 3
|
||
assert any(e["field"] == "balance" for e in data["errors"])
|
||
from app.models import VoucherDetail
|
||
assert db.query(VoucherDetail).count() == 1
|
||
|
||
def test_import_missing_cols(self, client: TestClient, db: Session):
|
||
"""缺必要列 → 400"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
buf = self._xlsx([["2026-08-01", "记-001", 100, 0]], cols=["日期", "凭证号", "借方金额", "贷方金额"])
|
||
resp = self._upload(client, token, buf)
|
||
assert resp.status_code == 400
|
||
assert "缺少必要列" in resp.json()["detail"] # 断言是列校验拦下，而非文件名防线
|