318 lines
14 KiB
Python
318 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="test_vouchers.xlsx"):
|
|
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
|