71b7846a79
将月份状态、资料输出、调整单编排及工作区查询、 CSV 生成移至现有服务层，路由仅绑定参数并返回响应。 保持接口路径、参数、响应模型和权限依赖一致， 补充九项内存库行为回归并同步测试记录。
214 lines
7.3 KiB
Python
214 lines
7.3 KiB
Python
"""用独立内存库守住路由分层调整前后的分页、账套隔离和工资状态契约。"""
|
|
|
|
import csv
|
|
import io
|
|
from datetime import datetime
|
|
from types import SimpleNamespace
|
|
from uuid import uuid4
|
|
|
|
import pytest
|
|
|
|
from app.api.v1 import payroll as payroll_api
|
|
from app.api.v1.system import tplus_workbench as workbench_api
|
|
from app.core.config import get_settings
|
|
from app.core.exceptions import NotFoundError
|
|
from app.core.response import PageQuery
|
|
from app.models import (
|
|
AdjustmentOrder,
|
|
MonthlySettlement,
|
|
PayrollMonth,
|
|
PerformanceLedger,
|
|
TPlusFact,
|
|
TPlusWorkLine,
|
|
TPlusWorkRun,
|
|
User,
|
|
)
|
|
from app.schemas.payroll import AllocationAdjustmentIn
|
|
from app.schemas.tplus_workbench import FactIn, SourceChoiceQuery
|
|
from app.services import payroll_calculation, payroll_months, settlement_service
|
|
from app.services.tplus import task_service, workbench
|
|
from tests.unit.test_formal_payroll import imported
|
|
from tests.unit.test_formal_payroll import payroll_case as payroll_case
|
|
from tests.unit.test_tplus_workbench import fact, source
|
|
from tests.unit.test_tplus_workbench import work_session as work_session
|
|
|
|
|
|
@pytest.mark.parametrize("state", ["empty", "draft", "generated", "legacy"])
|
|
def test_月份状态兼容未生成和历史工资(work_session, state):
|
|
s = work_session
|
|
user = User(username="month_maker", password="test_hash_only", status=1)
|
|
s.add(user)
|
|
s.flush()
|
|
if state in ("draft", "generated"):
|
|
s.add(
|
|
PayrollMonth(
|
|
month="2026-03",
|
|
maker_id=user.id,
|
|
maker_name=user.username,
|
|
generated_at=datetime(2026, 3, 31) if state == "generated" else None,
|
|
)
|
|
)
|
|
elif state == "legacy":
|
|
s.add(
|
|
MonthlySettlement(
|
|
settlement_month="2026-03", submitter_id=user.id, submitter_name=user.username
|
|
)
|
|
)
|
|
s.commit()
|
|
|
|
response = payroll_api.month_state("2026-03", s, user)
|
|
|
|
assert response.data == {
|
|
"month": "2026-03",
|
|
"allocation_locked": state in ("generated", "legacy"),
|
|
"maker_name": user.username if state in ("draft", "generated") else None,
|
|
}
|
|
|
|
|
|
@pytest.fixture
|
|
def source_config(monkeypatch):
|
|
monkeypatch.setattr(
|
|
task_service,
|
|
"resolve_tplus_settings",
|
|
lambda _settings: SimpleNamespace(t_plus_database="test_source"),
|
|
)
|
|
|
|
|
|
def test_事实历史分页保持倒序且隔离账套(work_session, source_config):
|
|
s = work_session
|
|
first = fact(s, "PRODUCT", "9", {"category": "municipal_water", "unit": "件"})
|
|
workbench.save_fact(
|
|
s,
|
|
FactIn(
|
|
kind="PRODUCT",
|
|
fact_key="9",
|
|
expected_version=1,
|
|
evidence="修改分类的测试依据",
|
|
data={"category": "rail_transit", "unit": "件"},
|
|
),
|
|
1,
|
|
)
|
|
foreign = TPlusFact(
|
|
source_database="other_source",
|
|
kind="PRODUCT",
|
|
fact_key="9",
|
|
version=99,
|
|
data_json={"category": "rail_transit", "unit": "件"},
|
|
evidence="其他账套资料",
|
|
requested_by=1,
|
|
)
|
|
s.add(foreign)
|
|
s.commit()
|
|
|
|
page = workbench_api.fact_history(s, None, first["id"], PageQuery(pageSize=1)).data
|
|
next_page = workbench_api.fact_history(
|
|
s, None, first["id"], PageQuery(pageNum=2, pageSize=1)
|
|
).data
|
|
|
|
assert page.total == next_page.total == 2
|
|
assert [row["version"] for row in page.records] == [2]
|
|
assert [row["version"] for row in next_page.records] == [1]
|
|
with pytest.raises(NotFoundError, match="补充资料不存在"):
|
|
workbench_api.fact_history(s, None, foreign.id, PageQuery())
|
|
|
|
|
|
def test_源选择按字面搜索且只返回必要字段(work_session, source_config):
|
|
s = work_session
|
|
source(
|
|
s,
|
|
"products",
|
|
[
|
|
{"id": 1, "code": "CODE_%1", "name": "目标产品", "unit_name": "件"},
|
|
{"id": 2, "code": "CODE_X1", "name": "其他产品", "unit_name": "件"},
|
|
],
|
|
)
|
|
|
|
page = workbench_api.source_choices(
|
|
s, None, SourceChoiceQuery(entity="products", search="CODE_%", pageSize=1)
|
|
).data
|
|
|
|
assert page.total == 1
|
|
assert page.records == [
|
|
{"source_id": "1", "source_no": "CODE_%1", "name": "目标产品", "parent_id": "", "date": ""}
|
|
]
|
|
|
|
|
|
def test_报告在开始流式响应前校验运行存在及账套(work_session, source_config):
|
|
with pytest.raises(NotFoundError, match="工作区运行不存在"):
|
|
workbench_api.work_report(work_session, None, "missing-run")
|
|
|
|
|
|
def test_正式输入及分配调整接口保留原台账与当月归属(payroll_case, monkeypatch):
|
|
s, user, _ = payroll_case
|
|
monkeypatch.setattr(get_settings(), "payroll_allow_legacy_calculation", False)
|
|
monkeypatch.setattr(get_settings(), "payroll_approval_enabled", False)
|
|
monkeypatch.setattr(payroll_months, "current_month", lambda: "2026-04")
|
|
assert payroll_api.input_detail(99999, s, user).data is None
|
|
business = imported(payroll_case)
|
|
detail = payroll_api.input_detail(business.id, s, user).data
|
|
assert detail["version"] == 1 and detail["source"] == "test_source"
|
|
assert detail["data"]["maps"]["receipts"]
|
|
result = payroll_calculation.calculate(s, business.id, "2026-03", operator=user)
|
|
original = s.get(PerformanceLedger, result["ledger_ids"][0])
|
|
settlement = settlement_service.generate(s, "2026-03", operator=user)
|
|
|
|
response = payroll_api.adjustment(
|
|
AllocationAdjustmentIn(
|
|
ledger_id=original.id,
|
|
personnel_id=original.personnel_id,
|
|
direction=1,
|
|
amount="1.00",
|
|
reason="核对原工资分配差额",
|
|
),
|
|
s,
|
|
user,
|
|
None,
|
|
)
|
|
|
|
order = s.get(AdjustmentOrder, response.data["id"])
|
|
assert response.data["status"] == 2
|
|
assert order.settlement_month == "2026-04" and order.source_ledger_id == original.id
|
|
assert order.attribution_json["manager_id"] == original.snapshot_json["manager_id"]
|
|
assert s.get(MonthlySettlement, settlement.id).total_payable == settlement.total_payable
|
|
|
|
|
|
def test_报告保留完整字段前导零与公式防护(work_session):
|
|
s = work_session
|
|
s.add(
|
|
TPlusWorkRun(
|
|
run_no="report-run",
|
|
request_key=str(uuid4()),
|
|
source_database="test_source",
|
|
requested_by=1,
|
|
kind="CALC_PREVIEW",
|
|
status="SUCCEEDED",
|
|
input_json={},
|
|
)
|
|
)
|
|
s.add(
|
|
TPlusWorkLine(
|
|
run_no="report-run",
|
|
line_no=1,
|
|
entity_type="group",
|
|
source_id="group-1",
|
|
action="READY",
|
|
data_json={
|
|
"receipt_no": "00043",
|
|
"shipment_no": "=1+1",
|
|
"amount": "100.00",
|
|
"rate_percent": "0.2%",
|
|
"performance": "0.20",
|
|
},
|
|
)
|
|
)
|
|
s.commit()
|
|
|
|
content = "".join(workbench.report_csv(s, "report-run"))
|
|
rows = list(csv.DictReader(io.StringIO(content.lstrip("\ufeff"))))
|
|
|
|
assert content.startswith("\ufeff") and len(rows) == 1
|
|
assert len(rows[0]) == 21
|
|
assert rows[0]["回款单号"] == "'00043" and rows[0]["发货单号"] == "'=1+1"
|
|
assert rows[0]["匹配金额"] == "100.00" and rows[0]["计提比例"] == "0.2%"
|
|
assert rows[0]["计提金额"] == "0.20" and rows[0]["说明"] == "仅供核对，未写入正式工资"
|