diff --git a/backend/app/api/v1/payroll.py b/backend/app/api/v1/payroll.py index 6c85992..b00c972 100644 --- a/backend/app/api/v1/payroll.py +++ b/backend/app/api/v1/payroll.py @@ -18,39 +18,12 @@ def month_state( session: DbSession, _: LoginUser, ): - from sqlalchemy import select - - from app.core.month import validate_month - from app.models import MonthlySettlement, PayrollMonth - - validate_month(month) - control = session.scalar(select(PayrollMonth).where(PayrollMonth.month == month)) - existing = session.scalar( - select(MonthlySettlement.id).where(MonthlySettlement.settlement_month == month) - ) - return ok( - { - "month": month, - "allocation_locked": bool(existing or (control and control.generated_at)), - "maker_name": control.maker_name if control else None, - } - ) + return ok(payroll_months.month_state(session, month)) @router.get("/businesses/{business_id}/input") def input_detail(business_id: int, session: DbSession, _: LoginUser): - record = payroll_calculation.latest_input(session, business_id) - return ok( - None - if record is None - else { - "id": record.id, - "version": record.version, - "source": record.source_database, - "fingerprint": record.fingerprint, - "data": record.data_json, - } - ) + return ok(payroll_inputs.detail(session, business_id)) @router.post("/businesses/{business_id}/input") @@ -90,19 +63,6 @@ def allocations( def adjustment( payload: AllocationAdjustmentIn, session: DbSession, actor: PayrollWriter, meta: RequestMetaDep ): - from app.core.exceptions import BadRequestError - from app.models import PerformanceLedger - from app.schemas.settlement import AdjustmentCreateIn - - old = session.get(PerformanceLedger, payload.ledger_id) - if old is None or old.settlement_id is None: - raise BadRequestError("请选择已生成工资的原台账") - data = AdjustmentCreateIn( - settlement_month=payroll_months.current_month(), - personnel_id=payload.personnel_id, - direction=payload.direction, - amount=payload.amount, - reason=payload.reason, - source_ledger_id=old.id, + return ok( + settlement_service.create_allocation_adjustment(session, payload, operator=actor, meta=meta) ) - return ok(settlement_service.create_adjustment(session, data, operator=actor, meta=meta)) diff --git a/backend/app/api/v1/system/tplus_workbench.py b/backend/app/api/v1/system/tplus_workbench.py index b2abeec..859428a 100644 --- a/backend/app/api/v1/system/tplus_workbench.py +++ b/backend/app/api/v1/system/tplus_workbench.py @@ -3,17 +3,12 @@ from typing import Annotated from fastapi import APIRouter, Path, Query -from sqlalchemy import func, select from starlette.responses import StreamingResponse from app.api.deps import DbSession, PayrollWriter, WorkbenchReader from app.core.response import ApiResponse, PageQuery, PageResult, ok -from app.models import TPlusSyncRecord, TPlusWorkLine from app.schemas.tplus_workbench import FactIn, SourceChoiceQuery, WorkCreateIn, WorkQuery from app.services.tplus import workbench -from app.services.tplus.report_labels import ACTION_LABELS, reason_label, spreadsheet_value -from app.services.tplus.snapshots import source_statement -from app.services.tplus.task_service import _source_database router = APIRouter() RunNo = Annotated[str, Path(pattern=r"^[A-Za-z0-9_-]+$", max_length=32)] @@ -36,33 +31,7 @@ def fact_history( session: DbSession, _: WorkbenchReader, fact_id: int, query: Annotated[PageQuery, Query()] ): """按事实身份分页查看修改历史，不允许跨账套读取。""" - from app.core.exceptions import NotFoundError - from app.models import TPlusFact - - fact = session.get(TPlusFact, fact_id) - if fact is None or fact.source_database != _source_database(): - raise NotFoundError("补充资料不存在") - conditions = [ - TPlusFact.source_database == fact.source_database, - TPlusFact.kind == fact.kind, - TPlusFact.fact_key == fact.fact_key, - ] - total = session.scalar(select(func.count()).select_from(TPlusFact).where(*conditions)) - rows = session.scalars( - select(TPlusFact) - .where(*conditions) - .order_by(TPlusFact.version.desc()) - .offset(query.offset) - .limit(query.page_size) - ) - return ok( - PageResult( - records=[workbench.fact_out(f) for f in rows], - total=total or 0, - page_num=query.page_num, - page_size=query.page_size, - ) - ) + return ok(workbench.fact_history(session, fact_id, query)) @router.post("/work-runs", response_model=ApiResponse[dict]) @@ -98,126 +67,14 @@ def source_choices( query: Annotated[SourceChoiceQuery, Query()], ): """仅提供挑选资料所需的源编号和名称，不返回电话、地址和原始载荷。""" - from app.core.exceptions import BadRequestError - - try: - statement = source_statement(session, _source_database(), query.entity) - except workbench.WorkbenchError: - raise BadRequestError("请先完整读取相关源数据") from None - if query.search: - escaped = query.search.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") - statement = statement.where(TPlusSyncRecord.source_no.like(f"%{escaped}%", escape="\\")) - total = session.scalar(select(func.count()).select_from(statement.subquery())) - rows = session.scalars( - statement.order_by(TPlusSyncRecord.source_id).offset(query.offset).limit(query.page_size) - ) - records = [ - { - "source_id": r.source_id, - "source_no": r.source_no, - "name": str(r.payload_json.get("name") or ""), - "parent_id": str(r.payload_json.get("source_parent_id") or ""), - "date": str( - r.payload_json.get("voucherdate") or r.payload_json.get("voucherDate") or "" - )[:10], - } - for r in rows - ] - return ok( - PageResult( - records=records, total=total or 0, page_num=query.page_num, page_size=query.page_size - ) - ) + return ok(workbench.source_choices(session, query)) @router.get("/work-runs/{run_no}/report") def work_report(session: DbSession, _: WorkbenchReader, run_no: RunNo): """导出全部预演明细，防止源单号被电子表格当作公式执行。""" - import csv - import io - - workbench.get_run(session, run_no) - - def rows(): - buffer = io.StringIO() - writer = csv.writer(buffer) - yield "\ufeff" - writer.writerow( - [ - "运行编号", - "行号", - "类型", - "源ID或平账组", - "处理结果", - "源编号", - "名称", - "回款单号", - "关联凭证示例", - "发货单号", - "匹配金额", - "分配后该明细剩余应收", - "回款天数", - "指导成本分摊", - "折算利润", - "利润门槛（百分数）", - "计提比例", - "计提金额", - "归属工号", - "原因", - "说明", - ] - ) - yield buffer.getvalue() - buffer.seek(0) - buffer.truncate(0) - statement = ( - select(TPlusWorkLine) - .where(TPlusWorkLine.run_no == run_no) - .order_by(TPlusWorkLine.line_no) - ) - for row in session.scalars(statement).yield_per(200): - data = row.data_json - values = data.get("values") or {} - fields = [ - run_no, - row.line_no, - row.entity_type, - row.source_id, - ACTION_LABELS.get(row.action, row.action), - data.get("source_no", ""), - values.get("personnel_name") - or values.get("product_name") - or values.get("company_name") - or data.get("source_name") - or "", - *[ - data.get(k, "") - for k in ( - "receipt_no", - "accounting_voucher_code", - "shipment_no", - "amount", - "shipment_remaining_after", - "days", - "allocated_cost", - "converted_profit", - "profit_threshold", - "rate_percent", - "performance", - "personnel_id", - ) - ], - reason_label(row.reason) - or ("经理本人业务，已排除绩效汇总" if data.get("excluded_manager_own") else ""), - "仅供核对，未写入正式工资", - ] - writer.writerow([spreadsheet_value(v) for v in fields]) - yield buffer.getvalue() - buffer.seek(0) - buffer.truncate(0) - return StreamingResponse( - rows(), + workbench.report_csv(session, run_no), media_type="text/csv; charset=utf-8", headers={"Content-Disposition": f'attachment; filename="{run_no}-preview.csv"'}, ) diff --git a/backend/app/services/payroll_inputs.py b/backend/app/services/payroll_inputs.py index 59f942c..5ea11a0 100644 --- a/backend/app/services/payroll_inputs.py +++ b/backend/app/services/payroll_inputs.py @@ -22,6 +22,20 @@ from app.services.payroll_calculation import latest_input from app.services.tplus.workbench import digest +def detail(session, business_id): + """返回最新核算资料；未准备资料的业务保持空结果。""" + record = latest_input(session, business_id) + if record is None: + return None + return { + "id": record.id, + "version": record.version, + "source": record.source_database, + "fingerprint": record.fingerprint, + "data": record.data_json, + } + + def setup(session, business_id, payload, operator, meta=None): with audit_service.audit( session, diff --git a/backend/app/services/payroll_months.py b/backend/app/services/payroll_months.py index 602cd61..73e1b7e 100644 --- a/backend/app/services/payroll_months.py +++ b/backend/app/services/payroll_months.py @@ -15,6 +15,20 @@ def current_month() -> str: return datetime.now(timezone(timedelta(hours=8))).strftime("%Y-%m") +def month_state(session, month): + """查询生成锁及制作人，兼容升级前已存在的月结记录。""" + validate_month(month) + control = session.scalar(select(PayrollMonth).where(PayrollMonth.month == month)) + existing = session.scalar( + select(MonthlySettlement.id).where(MonthlySettlement.settlement_month == month) + ) + return { + "month": month, + "allocation_locked": bool(existing or (control and control.generated_at)), + "maker_name": control.maker_name if control else None, + } + + def lock_month(session, month, operator): """INSERT 的唯一键串行化首次建行，避免两个请求同时认为月份不存在。""" ensure_payroll_writer(operator) diff --git a/backend/app/services/settlement_service.py b/backend/app/services/settlement_service.py index ca1c237..a9db1ba 100644 --- a/backend/app/services/settlement_service.py +++ b/backend/app/services/settlement_service.py @@ -534,6 +534,26 @@ def _detail_of(session: Session, settlement: MonthlySettlement) -> SettlementDet return detail +def create_allocation_adjustment(session, payload, *, operator, meta=None): + """校验已生成工资的原台账，并将分配差额转换为当月调整单。""" + from app.models import PerformanceLedger + from app.schemas.settlement import AdjustmentCreateIn + from app.services import payroll_months + + old = session.get(PerformanceLedger, payload.ledger_id) + if old is None or old.settlement_id is None: + raise BadRequestError("请选择已生成工资的原台账") + data = AdjustmentCreateIn( + settlement_month=payroll_months.current_month(), + personnel_id=payload.personnel_id, + direction=payload.direction, + amount=payload.amount, + reason=payload.reason, + source_ledger_id=old.id, + ) + return create_adjustment(session, data, operator=operator, meta=meta) + + def create_adjustment( session: Session, payload, diff --git a/backend/app/services/tplus/workbench.py b/backend/app/services/tplus/workbench.py index 2929e3b..9068c85 100644 --- a/backend/app/services/tplus/workbench.py +++ b/backend/app/services/tplus/workbench.py @@ -12,8 +12,9 @@ from sqlalchemy.orm import defer from app.core.exceptions import BadRequestError, NotFoundError from app.core.response import PageResult from app.core.tplus_lock import TPlusBusyError, sync_lock -from app.models import TPlusFact, TPlusWorkLine, TPlusWorkRun +from app.models import TPlusFact, TPlusSyncRecord, TPlusWorkLine, TPlusWorkRun from app.schemas.tplus_workbench import WorkCreateIn +from app.services.tplus.report_labels import ACTION_LABELS, reason_label, spreadsheet_value from app.services.tplus.task_service import _source_database, new_batch_no RULE_VERSION = "2026-10-30.2" @@ -244,6 +245,152 @@ def list_lines(session, number, query): ) +def fact_history(session, fact_id, query): + """按事实身份分页查看修改历史，不允许跨账套读取。""" + fact = session.get(TPlusFact, fact_id) + if fact is None or fact.source_database != _source_database(): + raise NotFoundError("补充资料不存在") + conditions = [ + TPlusFact.source_database == fact.source_database, + TPlusFact.kind == fact.kind, + TPlusFact.fact_key == fact.fact_key, + ] + total = session.scalar(select(func.count()).select_from(TPlusFact).where(*conditions)) + rows = session.scalars( + select(TPlusFact) + .where(*conditions) + .order_by(TPlusFact.version.desc()) + .offset(query.offset) + .limit(query.page_size) + ) + return PageResult( + records=[fact_out(f) for f in rows], + total=total or 0, + page_num=query.page_num, + page_size=query.page_size, + ) + + +def source_choices(session, query): + """仅提供挑选资料所需的源编号和名称，不返回电话、地址和原始载荷。""" + from app.services.tplus.snapshots import source_statement + + try: + statement = source_statement(session, _source_database(), query.entity) + except WorkbenchError: + raise BadRequestError("请先完整读取相关源数据") from None + if query.search: + escaped = query.search.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") + statement = statement.where(TPlusSyncRecord.source_no.like(f"%{escaped}%", escape="\\")) + total = session.scalar(select(func.count()).select_from(statement.subquery())) + rows = session.scalars( + statement.order_by(TPlusSyncRecord.source_id).offset(query.offset).limit(query.page_size) + ) + records = [ + { + "source_id": r.source_id, + "source_no": r.source_no, + "name": str(r.payload_json.get("name") or ""), + "parent_id": str(r.payload_json.get("source_parent_id") or ""), + "date": str( + r.payload_json.get("voucherdate") or r.payload_json.get("voucherDate") or "" + )[:10], + } + for r in rows + ] + return PageResult( + records=records, total=total or 0, page_num=query.page_num, page_size=query.page_size + ) + + +def report_csv(session, run_no): + """导出全部预演明细，防止源单号被电子表格当作公式执行。""" + import csv + import io + + get_run(session, run_no) + + def rows(): + buffer = io.StringIO() + writer = csv.writer(buffer) + yield "\ufeff" + writer.writerow( + [ + "运行编号", + "行号", + "类型", + "源ID或平账组", + "处理结果", + "源编号", + "名称", + "回款单号", + "关联凭证示例", + "发货单号", + "匹配金额", + "分配后该明细剩余应收", + "回款天数", + "指导成本分摊", + "折算利润", + "利润门槛（百分数）", + "计提比例", + "计提金额", + "归属工号", + "原因", + "说明", + ] + ) + yield buffer.getvalue() + buffer.seek(0) + buffer.truncate(0) + statement = ( + select(TPlusWorkLine) + .where(TPlusWorkLine.run_no == run_no) + .order_by(TPlusWorkLine.line_no) + ) + for row in session.scalars(statement).yield_per(200): + data = row.data_json + values = data.get("values") or {} + fields = [ + run_no, + row.line_no, + row.entity_type, + row.source_id, + ACTION_LABELS.get(row.action, row.action), + data.get("source_no", ""), + values.get("personnel_name") + or values.get("product_name") + or values.get("company_name") + or data.get("source_name") + or "", + *[ + data.get(k, "") + for k in ( + "receipt_no", + "accounting_voucher_code", + "shipment_no", + "amount", + "shipment_remaining_after", + "days", + "allocated_cost", + "converted_profit", + "profit_threshold", + "rate_percent", + "performance", + "personnel_id", + ) + ], + reason_label(row.reason) + or ("经理本人业务，已排除绩效汇总" if data.get("excluded_manager_own") else ""), + "仅供核对，未写入正式工资", + ] + writer.writerow([spreadsheet_value(v) for v in fields]) + yield buffer.getvalue() + buffer.seek(0) + buffer.truncate(0) + + return rows() + + def execute_next(session): """调用方持有原有全局锁；中断任务保留明细，禁止不明状态自动应用。""" for old in session.scalars(select(TPlusWorkRun).where(TPlusWorkRun.status == "RUNNING")): diff --git a/backend/tests/README.md b/backend/tests/README.md index f6651ee..ac174e5 100644 --- a/backend/tests/README.md +++ b/backend/tests/README.md @@ -1,5 +1,15 @@ # 后端测试 +## 审查问题修复核验（2026-11-02） + +- 修复正式核算起付比例的日期范围：2026-03-01 前的发货由期初应收承接，不再计入新业务销售额分母；与 FIFO 的启用边界保持一致。 +- 新增 2 项历史发货回归（含 2 月末边界）。修复前均因起付门槛误判而失败；修复后，期初 100 元、旧发货 100 元、新发货 100 元、回款 200 元的样本正确支付新业务全部绩效。 +- 将工资月份状态、核算资料出参、分配调整编排，以及 T+ 事实历史、源记录选择、CSV 生成迁入 service；路由的路径、参数、响应模型和权限依赖保持一致。 +- 新增 9 项内存库回归，覆盖月份锁定及旧月结兼容、历史分页与账套隔离、搜索通配符转义、报告提前校验、资料与当月调整归属、CSV 前导零和公式防护。 +- 验证：`backend/.venv-workbench/Scripts/python.exe -m pytest tests/unit -q --tb=short -p no:cacheprovider`，**804 passed**；受影响文件 Ruff 检查和差异空白检查通过。保留现有 Starlette TestClient 弃用提醒。 +- 提交前从暂存区导出独立目录验证：**797 项单元测试通过**（未包含另一路首页的 7 项单元测试），暂存 Python 文件 Ruff、前端类型检查和生产构建通过；保留既有大分块提醒。独立目录仅使用临时测试配置，未复制真实凭据。 +- 相关 11 项 MySQL 集成用例已通过收集检查，但本轮未执行；未运行数据库迁移或清理，不将内存库结果记为目标 MySQL 或生产验收。前端无改动，未重复构建。 + > AI 摘要：2026-10-28 的 P0.5 连接修复已通过 19 项定点测试和 1001 项完整回归；使用本轮独立 MySQL/Redis 容器，未连接客户 T+ 源库。真实备份源 10.50.1600 验收仍未执行。 ## 连接策略修复核验（2026-10-28） diff --git a/backend/tests/integration/test_tplus_workbench.py b/backend/tests/integration/test_tplus_workbench.py index 1e85782..f86b94f 100644 --- a/backend/tests/integration/test_tplus_workbench.py +++ b/backend/tests/integration/test_tplus_workbench.py @@ -41,7 +41,6 @@ def workspace(monkeypatch): ) monkeypatch.setattr("app.services.tplus.task_service.get_settings", lambda: settings) monkeypatch.setattr(workbench, "_source_database", lambda: source) - monkeypatch.setattr("app.api.v1.system.tplus_workbench._source_database", lambda: source) with get_session_factory()() as db: yield db, source db.rollback() diff --git a/backend/tests/unit/test_payroll_workbench_api.py b/backend/tests/unit/test_payroll_workbench_api.py new file mode 100644 index 0000000..97fff64 --- /dev/null +++ b/backend/tests/unit/test_payroll_workbench_api.py @@ -0,0 +1,213 @@ +"""用独立内存库守住路由分层调整前后的分页、账套隔离和工资状态契约。""" + +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]["说明"] == "仅供核对，未写入正式工资"