1461a8d113
按 2026-11-02 需求补齐计划交付后端能力： - 角色授权：权限目录化，接口统一准入校验，未登记默认拒绝； 管理员判断改为按数据库角色重新判定，防止权限码持有者绕过 - 系统监控：单进程采集资源、请求与异常指标，分钟级聚合， 阈值配置化；敏感配置增加 repr 保护，Redis 密码 URL 转义 - 试用期计提：按回款日判定并写入快照，年度结清与调整单联动； 人员模型补试用期、离职日期与工资版本历史字段 - 计提模板：公式骨架受控，页面维护名称与参数绑定 - T+ 采集：自动化配置表与进度查询 内置角色禁停用保护对齐模块注释；核算规则版本提为常量并 补充批量核算失败堆栈日志。 影响范围：后端全部模块、数据库迁移 V22-V28 验证：uv run ruff check 通过；922 个单元测试全部通过； 集成测试因本机 MySQL 数据卷被其他容器锁定暂未运行
226 lines
8.9 KiB
Python
226 lines
8.9 KiB
Python
"""隔离 MySQL 验证需求补齐的页面接口、版本和工资原子提交。"""
|
||
|
||
from datetime import date, timedelta
|
||
from decimal import Decimal
|
||
from io import BytesIO
|
||
|
||
import pytest
|
||
from openpyxl import load_workbook
|
||
from sqlalchemy import func, select
|
||
|
||
from app.core.config import get_settings
|
||
from app.core.db import get_session_factory
|
||
from app.models import ParamChangeBatch, ProbationPayment
|
||
from tests.integration.test_formal_payroll import formal_chain as formal_chain
|
||
|
||
pytestmark = pytest.mark.integration
|
||
|
||
|
||
def test_系数表未来版本与工作簿往返(client, admin):
|
||
headers = admin["headers"]
|
||
base = "/api/v1/system/coefficients"
|
||
data = client.get(base, headers=headers).json()["data"]
|
||
assert len(data["rows"]) == 10
|
||
rows = [
|
||
{
|
||
k: row.get(k)
|
||
for k in ("product_category", "customer_type", "stage", "value", "slope", "boundary")
|
||
}
|
||
for row in data["rows"]
|
||
if not row.get("readonly")
|
||
]
|
||
rows[-1]["value"] = "5"
|
||
payload = {
|
||
"effective_date": str(date.today() + timedelta(days=2)),
|
||
"base_date": str(date.today()),
|
||
"fingerprint": data["fingerprint"],
|
||
"rows": rows,
|
||
}
|
||
response = client.put(base, headers=headers, json=payload)
|
||
assert response.status_code == 200, response.text
|
||
assert client.put(base, headers=headers, json=payload).status_code == 400
|
||
future = client.get(base, headers=headers, params={"as_of": payload["effective_date"]}).json()[
|
||
"data"
|
||
]
|
||
assert future["rows"][-2]["value"] == "5"
|
||
export = client.get(base + "/export", headers=headers)
|
||
assert export.status_code == 200
|
||
preview = client.post(
|
||
base + "/import-preview",
|
||
headers=headers,
|
||
files={
|
||
"file": (
|
||
"coefficients.xlsx",
|
||
export.content,
|
||
"application/vnd.openxmlformats-officedocument.spreadsheetml.sheet",
|
||
)
|
||
},
|
||
)
|
||
assert preview.status_code == 200, preview.text
|
||
assert preview.json()["data"]["can_apply"]
|
||
book = load_workbook(BytesIO(export.content))
|
||
book["系数参数"].cell(2, 4, "bad-value")
|
||
output = BytesIO()
|
||
book.save(output)
|
||
with get_session_factory()() as session:
|
||
before = session.scalar(select(func.count()).select_from(ParamChangeBatch))
|
||
bad = client.post(
|
||
base + "/import-preview", headers=headers, files={"file": ("bad.xlsx", output.getvalue())}
|
||
)
|
||
assert bad.status_code == 200, bad.text
|
||
assert not bad.json()["data"]["can_apply"]
|
||
with get_session_factory()() as session:
|
||
assert session.scalar(select(func.count()).select_from(ParamChangeBatch)) == before
|
||
|
||
|
||
def test_业务提取产品及重复预览不重复建档(client, admin, formal_chain):
|
||
headers = admin["headers"]
|
||
business, _, _, _, _ = formal_chain
|
||
detail = client.get(f"/api/v1/business/businesses/{business['id']}", headers=headers).json()[
|
||
"data"
|
||
]
|
||
ids = [row["id"] for row in detail["items"]]
|
||
base = "/api/v1/base/products/extract"
|
||
plan = client.post(base + "/preview", headers=headers, json={"item_ids": ids})
|
||
assert plan.status_code == 200, plan.text
|
||
data = plan.json()["data"]
|
||
applied = client.post(
|
||
base, headers=headers, json={"item_ids": ids, "fingerprint": data["fingerprint"]}
|
||
)
|
||
assert applied.status_code == 200, applied.text
|
||
target = applied.json()["data"]["rows"][0]["target_id"]
|
||
assert target
|
||
again = client.post(base + "/preview", headers=headers, json={"item_ids": ids}).json()["data"]
|
||
assert again["rows"][0]["action"] == "LINK"
|
||
assert again["rows"][0]["target_id"] == target
|
||
|
||
|
||
def test_无绩效试用期工资及冻结分项导出(client, admin, monkeypatch):
|
||
monkeypatch.setattr(get_settings(), "payroll_allow_legacy_calculation", False)
|
||
headers = admin["headers"]
|
||
roles = client.get("/api/v1/system/business-roles", headers=headers).json()["data"]
|
||
if isinstance(roles, dict):
|
||
roles = roles.get("list", [])
|
||
role = next(row for row in roles if row["role_code"] == "salesperson")
|
||
person = client.post(
|
||
"/api/v1/base/personnel",
|
||
headers=headers,
|
||
json={
|
||
"personnel_name": "t_无绩效基本工资",
|
||
"business_role_id": role["id"],
|
||
"salary_standard": "3000",
|
||
"hire_date": "2026-01-01",
|
||
"probation_start_date": "2026-01-01",
|
||
"regularization_date": "2027-01-01",
|
||
"is_probation": 1,
|
||
},
|
||
)
|
||
assert person.status_code == 200, person.text
|
||
generated = client.post(
|
||
"/api/v1/settlements/generate", headers=headers, json={"month": "2026-11"}
|
||
)
|
||
assert generated.status_code == 200, generated.text
|
||
item = next(
|
||
row
|
||
for row in generated.json()["data"]["items"]
|
||
if row["personnel_id"] == person.json()["data"]["id"]
|
||
)
|
||
assert Decimal(item["net_amount"]) == Decimal("3000")
|
||
assert Decimal(item["normal_payable_sum"]) == 0
|
||
assert (
|
||
client.post(
|
||
"/api/v1/settlements/generate", headers=headers, json={"month": "2026-11"}
|
||
).status_code
|
||
== 400
|
||
)
|
||
export = client.get("/api/v1/payroll/salary-export/2026-11", headers=headers)
|
||
assert export.status_code == 200
|
||
assert "工资分项" in load_workbook(BytesIO(export.content)).sheetnames
|
||
with get_session_factory()() as session:
|
||
assert session.scalar(select(func.count()).select_from(ProbationPayment)) == 0
|
||
|
||
|
||
def test_只读账号不能维护系数或执行提取(client, normal):
|
||
headers = normal["headers"]
|
||
response = client.post("/api/v1/base/products/extract", headers=headers, json={"item_ids": [1]})
|
||
assert response.status_code == 403
|
||
response = client.put("/api/v1/system/coefficients", headers=headers, json={})
|
||
assert response.status_code == 403
|
||
|
||
|
||
def test_已生成工资的年度续付走当月调整且防重(client, admin, formal_chain, monkeypatch):
|
||
from app.models import PerformanceLedger
|
||
from app.services import payroll_months
|
||
|
||
monkeypatch.setattr(get_settings(), "payroll_allow_legacy_calculation", False)
|
||
monkeypatch.setattr(get_settings(), "payroll_approval_enabled", False)
|
||
monkeypatch.setattr(payroll_months, "current_month", lambda: "2027-01")
|
||
headers = admin["headers"]
|
||
business, _, _, owner, _ = formal_chain
|
||
response = client.post(
|
||
f"/api/v1/payroll/businesses/{business['id']}/calculate",
|
||
headers=headers,
|
||
json={"month": "2026-03"},
|
||
)
|
||
assert response.status_code == 200, response.text
|
||
# 隔离合成台账：年度计提20000、年底仅1000符合业务支付条件。
|
||
with get_session_factory()() as session:
|
||
row = session.scalar(
|
||
select(PerformanceLedger).where(PerformanceLedger.business_id == business["id"])
|
||
)
|
||
row.is_probation = 1
|
||
row.performance_amount = Decimal("20000")
|
||
row.payable_amount = Decimal("1000")
|
||
session.commit()
|
||
december = client.post(
|
||
"/api/v1/settlements/generate", headers=headers, json={"month": "2026-12"}
|
||
)
|
||
assert december.status_code == 200, december.text
|
||
first = next(
|
||
row for row in december.json()["data"]["items"] if row["personnel_id"] == owner["id"]
|
||
)
|
||
assert Decimal(first["probation_payment_sum"]) == Decimal("1000")
|
||
january = client.post(
|
||
"/api/v1/settlements/generate", headers=headers, json={"month": "2027-01"}
|
||
)
|
||
assert january.status_code == 200, january.text
|
||
# 工资生成后确认额外满足支付条件的1000，原年度权益只剩1000。
|
||
with get_session_factory()() as session:
|
||
row = session.scalar(
|
||
select(PerformanceLedger).where(
|
||
PerformanceLedger.business_id == business["id"], PerformanceLedger.calc_type == 1
|
||
)
|
||
)
|
||
row.payable_amount = Decimal("3000")
|
||
session.commit()
|
||
preview = client.get(
|
||
"/api/v1/payroll/probation",
|
||
headers=headers,
|
||
params={"month": "2027-01", "personnel_id": owner["id"]},
|
||
)
|
||
assert preview.status_code == 200, preview.text
|
||
balance = preview.json()["data"]["rows"][0]
|
||
assert Decimal(balance["delta"]) == Decimal("1000")
|
||
payload = {
|
||
"personnel_id": owner["id"],
|
||
"year": "2026",
|
||
"fingerprint": balance["fingerprint"],
|
||
"reason": "隔离样本：原年度后续支付条件满足",
|
||
}
|
||
adjustment = client.post("/api/v1/payroll/probation/adjustments", headers=headers, json=payload)
|
||
assert adjustment.status_code == 200, adjustment.text
|
||
assert Decimal(adjustment.json()["data"]["amount"]) == Decimal("1000")
|
||
assert (
|
||
client.post(
|
||
"/api/v1/payroll/probation/adjustments", headers=headers, json=payload
|
||
).status_code
|
||
== 400
|
||
)
|
||
final = client.get(
|
||
"/api/v1/payroll/probation",
|
||
headers=headers,
|
||
params={"month": "2027-01", "personnel_id": owner["id"]},
|
||
).json()["data"]["rows"][0]
|
||
assert Decimal(final["paid"]) == Decimal("2000")
|
||
assert Decimal(final["delta"]) == 0
|