209 lines
7.2 KiB
Python
209 lines
7.2 KiB
Python
"""预算管理模块测试 — 预算计划CRUD + 自动分解 + 版本"""
|
||
import pytest
|
||
from fastapi.testclient import TestClient
|
||
from sqlalchemy.orm import Session
|
||
from tests.conftest import create_test_user, get_token_for_user, auth_header, create_test_kpi
|
||
|
||
|
||
class TestBudgetPlans:
|
||
"""预算计划CRUD测试"""
|
||
|
||
BASE = "/api/cma/budget"
|
||
|
||
def test_list_plans_empty(self, client: TestClient, db: Session):
|
||
"""空列表"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
|
||
resp = client.get(f"{self.BASE}/plans", headers=auth_header(token))
|
||
assert resp.status_code == 200
|
||
assert resp.json()["data"] == []
|
||
|
||
def test_create_plan(self, client: TestClient, db: Session):
|
||
"""创建预算计划"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db, kpi_code="BUDGET_KPI_01")
|
||
|
||
resp = client.post(
|
||
f"{self.BASE}/plans",
|
||
headers=auth_header(token),
|
||
json={"kpi_id": kpi.id, "period": "2026-06", "budget_value": 50000.0},
|
||
)
|
||
assert resp.status_code == 200
|
||
assert resp.json()["message"] == "预算已创建"
|
||
assert resp.json()["id"] > 0
|
||
|
||
def test_create_plan_duplicate_upsert(self, client: TestClient, db: Session):
|
||
"""重复创建同一KPI+期间 → 更新而非新增"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db, kpi_code="BUDGET_KPI_02")
|
||
|
||
# 创建
|
||
client.post(
|
||
f"{self.BASE}/plans",
|
||
headers=auth_header(token),
|
||
json={"kpi_id": kpi.id, "period": "2026-06", "budget_value": 30000.0},
|
||
)
|
||
# 再次创建（更新）
|
||
resp = client.post(
|
||
f"{self.BASE}/plans",
|
||
headers=auth_header(token),
|
||
json={"kpi_id": kpi.id, "period": "2026-06", "budget_value": 35000.0},
|
||
)
|
||
assert resp.status_code == 200
|
||
assert resp.json()["message"] == "预算已更新"
|
||
|
||
def test_create_plan_missing_fields(self, client: TestClient, db: Session):
|
||
"""缺少必要参数被拒绝"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
|
||
resp = client.post(
|
||
f"{self.BASE}/plans",
|
||
headers=auth_header(token),
|
||
json={"kpi_id": 1}, # 缺少 period 和 budget_value
|
||
)
|
||
assert resp.status_code == 400
|
||
|
||
def test_update_plan(self, client: TestClient, db: Session):
|
||
"""更新预算计划"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db, kpi_code="BUDGET_KPI_03")
|
||
|
||
create_resp = client.post(
|
||
f"{self.BASE}/plans",
|
||
headers=auth_header(token),
|
||
json={"kpi_id": kpi.id, "period": "2026-06", "budget_value": 100.0},
|
||
)
|
||
plan_id = create_resp.json()["id"]
|
||
|
||
resp = client.put(
|
||
f"{self.BASE}/plans/{plan_id}",
|
||
headers=auth_header(token),
|
||
json={"budget_value": 200.0, "remark": "已更新"},
|
||
)
|
||
assert resp.status_code == 200
|
||
|
||
def test_delete_plan(self, client: TestClient, db: Session):
|
||
"""删除预算计划"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db, kpi_code="BUDGET_KPI_04")
|
||
|
||
create_resp = client.post(
|
||
f"{self.BASE}/plans",
|
||
headers=auth_header(token),
|
||
json={"kpi_id": kpi.id, "period": "2026-06", "budget_value": 100.0},
|
||
)
|
||
plan_id = create_resp.json()["id"]
|
||
|
||
resp = client.delete(f"{self.BASE}/plans/{plan_id}", headers=auth_header(token))
|
||
assert resp.status_code == 200
|
||
assert resp.json()["message"] == "预算已删除"
|
||
|
||
# 验证已删除
|
||
get_resp = client.get(f"{self.BASE}/plans", headers=auth_header(token))
|
||
ids = [p["id"] for p in get_resp.json()["data"]]
|
||
assert plan_id not in ids
|
||
|
||
|
||
class TestBudgetAutoDecompose:
|
||
"""预算自动分解测试"""
|
||
|
||
BASE = "/api/cma/budget"
|
||
|
||
def test_auto_decompose_single_kpi(self, client: TestClient, db: Session):
|
||
"""单KPI自动分解年度预算为月度（均分）"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db, kpi_code="BUDGET_DECOMPOSE")
|
||
|
||
resp = client.post(
|
||
f"{self.BASE}/auto-decompose",
|
||
headers=auth_header(token),
|
||
json={"kpi_id": kpi.id, "annual_budget": 120000, "year": 2026, "method": "equal"},
|
||
)
|
||
assert resp.status_code == 200
|
||
data = resp.json()
|
||
assert "已分解" in data["message"]
|
||
assert len(data["results"]) == 1
|
||
# 年度预算120000，12个月均分，每月10000
|
||
monthly = data["results"][0]["monthly"]
|
||
assert len(monthly) == 12
|
||
assert monthly[0]["value"] == 10000.0
|
||
|
||
def test_auto_decompose_missing(self, client: TestClient, db: Session):
|
||
"""没有年度预算数据时尝试分解 → 400"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
create_test_kpi(db, kpi_code="BUDGET_NO_DATA")
|
||
|
||
resp = client.post(
|
||
f"{self.BASE}/auto-decompose",
|
||
headers=auth_header(token),
|
||
json={"year": 2026, "method": "equal"},
|
||
)
|
||
assert resp.status_code == 400
|
||
|
||
|
||
class TestBudgetVersions:
|
||
"""预算版本管理测试"""
|
||
|
||
BASE = "/api/cma/budget"
|
||
|
||
def test_create_and_submit_version(self, client: TestClient, db: Session):
|
||
"""创建预算后查询版本并提交"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
kpi = create_test_kpi(db, kpi_code="BUDGET_VER_KPI")
|
||
|
||
# 创建一条预算
|
||
client.post(
|
||
f"{self.BASE}/plans",
|
||
headers=auth_header(token),
|
||
json={"kpi_id": kpi.id, "period": "2026-06", "budget_value": 10000.0},
|
||
)
|
||
|
||
# 查询版本
|
||
ver_resp = client.get(f"{self.BASE}/versions", headers=auth_header(token))
|
||
assert ver_resp.status_code == 200
|
||
assert len(ver_resp.json()["data"]) >= 1
|
||
|
||
# 提交版本
|
||
submit_resp = client.post(
|
||
f"{self.BASE}/versions/submit",
|
||
headers=auth_header(token),
|
||
json={"version": "v1.0"},
|
||
)
|
||
assert submit_resp.status_code == 200
|
||
assert "已提交审批" in submit_resp.json()["message"]
|
||
|
||
def test_approve_version(self, client: TestClient, db: Session):
|
||
"""审批通过版本"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
|
||
resp = client.post(
|
||
f"{self.BASE}/versions/approve",
|
||
headers=auth_header(token),
|
||
json={"version": "v1.0", "action": "approved"},
|
||
)
|
||
assert resp.status_code == 200
|
||
assert "已批准" in resp.json()["message"]
|
||
|
||
def test_reject_version(self, client: TestClient, db: Session):
|
||
"""驳回版本"""
|
||
create_test_user(db)
|
||
token = get_token_for_user(client)
|
||
|
||
resp = client.post(
|
||
f"{self.BASE}/versions/approve",
|
||
headers=auth_header(token),
|
||
json={"version": "v2.0", "action": "rejected"},
|
||
)
|
||
assert resp.status_code == 200
|
||
assert "已驳回" in resp.json()["message"]
|