"""用独立内存库守住路由分层调整前后的分页、账套隔离和工资状态契约。""" 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 Settings, 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): settings = Settings( _env_file=None, app_secret_key="x" * 32, t_plus_database="test_source" ) monkeypatch.setattr(task_service, "get_settings", lambda: settings) 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, source_config): 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]["说明"] == "仅供核对，未写入正式工资"