638 lines
23 KiB
Python
638 lines
23 KiB
Python
"""正式链路使用独立内存库，验证金额、防重、生成锁与历史归属。"""
|
||
|
||
import copy
|
||
from datetime import date
|
||
from decimal import Decimal
|
||
from types import SimpleNamespace
|
||
from uuid import uuid4
|
||
|
||
import pytest
|
||
from sqlalchemy import func, select
|
||
|
||
from app.core.exceptions import BadRequestError, ForbiddenError
|
||
from app.models import (
|
||
Business,
|
||
BusinessRole,
|
||
Customer,
|
||
PayrollAllocation,
|
||
PayrollInput,
|
||
PayrollMonth,
|
||
PerformanceLedger,
|
||
Personnel,
|
||
Product,
|
||
Receipt,
|
||
Shipment,
|
||
ShipmentItem,
|
||
TPlusSourceLink,
|
||
TPlusWorkRun,
|
||
User,
|
||
)
|
||
from app.services import manager_share, payroll_calculation, payroll_months
|
||
from app.services.payroll_access import can_write_payroll
|
||
from app.services.tplus.business_import import execute_business
|
||
from app.services.tplus.workbench import digest
|
||
from tests.unit.test_tplus_workbench import apply, fact, prepare_calc, preview
|
||
from tests.unit.test_tplus_workbench import source as publish_source
|
||
from tests.unit.test_tplus_workbench import work_session as work_session
|
||
|
||
|
||
@pytest.fixture
|
||
def payroll_case(work_session, monkeypatch):
|
||
s = work_session
|
||
user = User(username="test_maker", password="test_hash_only", status=1)
|
||
s.add(user)
|
||
role = BusinessRole(
|
||
role_name="经理",
|
||
role_code="sales_manager",
|
||
calc_template="SALES_MANAGER",
|
||
status=1,
|
||
sort_order=2,
|
||
)
|
||
s.add(role)
|
||
s.flush()
|
||
manager = Personnel(
|
||
personnel_name="测试经理", business_role_id=role.id, status=1, source_no="90001"
|
||
)
|
||
s.add(manager)
|
||
s.commit()
|
||
from app.models import Role, UserRole
|
||
|
||
maker_role = s.scalar(select(Role).where(Role.role_code == "payroll_maker"))
|
||
user.user_roles.append(UserRole(role_id=maker_role.id))
|
||
s.commit()
|
||
prepare_calc(s)
|
||
apply(s, preview(s, ["personnel"]))
|
||
product = Product(
|
||
product_name="测试产品",
|
||
spec_model="测试规格",
|
||
unit="件",
|
||
product_category="municipal_water",
|
||
status=1,
|
||
)
|
||
customer = Customer(company_name="测试客户", owner_name="负责人", customer_type="personal")
|
||
s.add_all([product, customer])
|
||
s.flush()
|
||
s.add_all(
|
||
[
|
||
TPlusSourceLink(
|
||
source_database="test_source",
|
||
entity_type="products",
|
||
source_id="9",
|
||
target_id=product.id,
|
||
),
|
||
TPlusSourceLink(
|
||
source_database="test_source",
|
||
entity_type="customers",
|
||
source_id="8",
|
||
target_id=customer.id,
|
||
),
|
||
]
|
||
)
|
||
s.commit()
|
||
s.add(
|
||
TPlusSourceLink(
|
||
source_database="test_source",
|
||
entity_type="personnel",
|
||
source_id="90001",
|
||
target_id=manager.id,
|
||
)
|
||
)
|
||
s.commit()
|
||
fact(
|
||
s,
|
||
"PERSONNEL",
|
||
"90001",
|
||
{
|
||
"code": "90001",
|
||
"name": manager.personnel_name,
|
||
"role_code": "sales_manager",
|
||
"status": 1,
|
||
},
|
||
)
|
||
fact(
|
||
s,
|
||
"MANAGER",
|
||
"manager-biz",
|
||
{"business_key": "biz", "manager_id": "90001", "effective_date": "2026-03-01"},
|
||
)
|
||
from app.models import TPlusFact
|
||
|
||
guidance = s.scalar(select(TPlusFact).where(TPlusFact.kind == "GUIDANCE"))
|
||
guidance.data_json = {**guidance.data_json, "is_finalized": True}
|
||
s.commit()
|
||
return s, user, manager
|
||
|
||
|
||
def business_work(s, user, kind, previous=None):
|
||
run = TPlusWorkRun(
|
||
run_no=uuid4().hex,
|
||
request_key=str(uuid4()),
|
||
source_database="test_source",
|
||
requested_by=user.id,
|
||
kind=kind,
|
||
status="RUNNING",
|
||
input_json={"preview_no": previous.run_no}
|
||
if previous
|
||
else {"month": "2026-03", "group_keys": ["g"]},
|
||
)
|
||
s.add(run)
|
||
s.flush()
|
||
execute_business(s, run)
|
||
s.commit()
|
||
return run
|
||
|
||
|
||
def confirm_test_receipts(s, user):
|
||
"""金额回归的明确前置：经独立确认服务确认测试回款，不关闭默认门槛。"""
|
||
from app.schemas.receipt import ReceiptConfirmIn
|
||
from app.services import receipt_confirmation, receipt_service
|
||
|
||
for row in list(s.scalars(select(Receipt))):
|
||
receipt_service.confirm_receipt(s, row.id, ReceiptConfirmIn(
|
||
expected_revision=receipt_confirmation.revision(row),
|
||
), operator=user)
|
||
|
||
|
||
def imported(case, *, confirmed=True):
|
||
s, user, _manager = case
|
||
pre = business_work(s, user, "BUSINESS_PREVIEW")
|
||
assert pre.status == "SUCCEEDED", pre.snapshot_json["plan"]
|
||
assert pre.summary_json["counts"] == {"CREATE": 1}
|
||
run = business_work(s, user, "BUSINESS_APPLY", pre)
|
||
assert run.status == "SUCCEEDED", run.summary_json
|
||
if confirmed:
|
||
confirm_test_receipts(s, user)
|
||
return s.scalar(select(Business))
|
||
|
||
|
||
def test_正式导入不自动确认_保存资料也不能代替确认(payroll_case):
|
||
s, user, _ = payroll_case
|
||
business = imported(payroll_case, confirmed=False)
|
||
assert s.scalar(select(Receipt)).confirmation_status == 0
|
||
with pytest.raises(BadRequestError, match="人工确认"):
|
||
payroll_calculation.calculate(s, business.id, "2026-03", operator=user)
|
||
|
||
|
||
def test_工资角色可写且未授权对象不能冒用身份(payroll_case):
|
||
_, user, _ = payroll_case
|
||
assert can_write_payroll(user)
|
||
assert not can_write_payroll(SimpleNamespace(id=user.id + 1, status=1))
|
||
assert not can_write_payroll(None)
|
||
|
||
|
||
def test_正式导入防重与来源可追溯(payroll_case):
|
||
s, user, _ = payroll_case
|
||
business = imported(payroll_case)
|
||
assert s.scalar(select(func.count()).select_from(Receipt)) == 1
|
||
assert s.scalar(select(func.count()).select_from(ShipmentItem)) == 1
|
||
again = business_work(s, user, "BUSINESS_PREVIEW")
|
||
assert again.summary_json["counts"] == {"SKIP": 1}
|
||
business_work(s, user, "BUSINESS_APPLY", again)
|
||
assert s.scalar(select(func.count()).select_from(Business)) == 1
|
||
record = payroll_calculation.latest_input(s, business.id)
|
||
assert record.data_json["source_facts"]
|
||
|
||
|
||
@pytest.fixture
|
||
def correction_case(payroll_case, monkeypatch):
|
||
from contextlib import nullcontext
|
||
|
||
from app.models import TPlusSyncRecord
|
||
from app.schemas.tplus_workbench import CorrectionPairIn
|
||
from app.services.tplus import corrections
|
||
from app.services.tplus.snapshots import source_records
|
||
|
||
s, user, _ = payroll_case
|
||
original = source_records(s, "test_source", "shipments")[0].payload_json
|
||
original_line = source_records(s, "test_source", "shipment_items")[0].payload_json
|
||
# 收款ID与负更正发货ID相同，必须由实体前缀区分。
|
||
publish_source(s, "shipments", [original,
|
||
{**original, "id": 30, "code": "A-NEG"},
|
||
{**original, "id": 40, "code": "B-POS", "idcustomer": 88}])
|
||
publish_source(s, "shipment_items", [original_line,
|
||
{**original_line, "id": 31, "source_parent_id": 30, "taxAmount": "-100", "quantity": "-2"},
|
||
{**original_line, "id": 41, "source_parent_id": 40}])
|
||
publish_source(s, "receipts", [{"id": 30, "code": "B-CASH", "idpartner": 88,
|
||
"voucherDate": "2026-03-10", "isReceiveFlag": 1, "voucherstate": 189, "amount": "100",
|
||
"accounting_voucher_count": 1, "valid_accounting_voucher_count": 1,
|
||
"accounting_voucher_id": 123, "accounting_voucher_code": "TEST-GL"}])
|
||
monkeypatch.setattr(corrections, "sync_lock", lambda db: nullcontext())
|
||
payload = CorrectionPairIn(negative_shipment_id="30", positive_shipment_id="40",
|
||
receipt_ids=["30"], target_group_key="g", evidence="隔离转客户例单核对")
|
||
before = s.scalar(select(func.count()).select_from(TPlusSyncRecord))
|
||
return s, user, corrections, payload, before
|
||
|
||
|
||
def test_更正对贯通正式导入与FIFO_保留日期金额且重复不增加(payroll_case, correction_case):
|
||
from app.models import TPlusFact, TPlusSyncRecord
|
||
|
||
s, user, service, payload, before = correction_case
|
||
first = service.save_pair(s, payload, user.id)
|
||
again = service.save_pair(s, payload, user.id)
|
||
assert first["pair_key"] == again["pair_key"]
|
||
assert s.scalar(select(func.count()).select_from(TPlusFact).where(TPlusFact.kind == "CORRECTION")) == 3
|
||
business = imported(payroll_case)
|
||
assert business.total_sale_amount == Decimal("100")
|
||
cash = s.scalar(select(Receipt))
|
||
assert cash.source_no == "B-CASH" and cash.receipt_date == date(2026, 3, 10)
|
||
assert cash.amount == Decimal("100") and cash.business_id == business.id
|
||
assert s.scalar(select(func.count()).select_from(Shipment)) == 1
|
||
assert s.scalar(select(func.count()).select_from(TPlusSyncRecord)) == before
|
||
result = payroll_calculation.calculate(s, business.id, "2026-03", operator=user)
|
||
assert result["performance"] == "0.20" and result["payable"] == "0.20"
|
||
repeated = business_work(s, user, "BUSINESS_PREVIEW")
|
||
assert repeated.summary_json["counts"] == {"SKIP": 1}
|
||
with pytest.raises(BadRequestError):
|
||
payroll_calculation.calculate(s, business.id, "2026-03", operator=user)
|
||
|
||
|
||
@pytest.mark.parametrize("change", ["wrong_sign", "wrong_group", "wrong_receipt", "missing_source"])
|
||
def test_无效更正对原子拒绝不留下半边事实(correction_case, change):
|
||
from app.models import TPlusFact
|
||
|
||
s, user, service, payload, _ = correction_case
|
||
updates = {"wrong_sign": {"negative_shipment_id": "40", "positive_shipment_id": "30"},
|
||
"wrong_group": {"target_group_key": "unknown"},
|
||
"wrong_receipt": {"receipt_ids": ["999"]},
|
||
"missing_source": {"positive_shipment_id": "999"}}
|
||
with pytest.raises(BadRequestError):
|
||
service.save_pair(s, payload.model_copy(update=updates[change]), user.id)
|
||
assert s.scalar(select(func.count()).select_from(TPlusFact).where(TPlusFact.kind == "CORRECTION")) == 0
|
||
|
||
|
||
def test_更正对后来源变化阻止静默排除(correction_case):
|
||
from app.services.tplus.snapshots import source_records
|
||
from app.services.tplus.workbench import latest_facts
|
||
|
||
s, user, service, payload, _ = correction_case
|
||
service.save_pair(s, payload, user.id)
|
||
rows = [r.payload_json for r in source_records(s, "test_source", "shipments")]
|
||
publish_source(s, "shipments", [{**r, "code": "CHANGED"} if r["id"] == 30 else r for r in rows])
|
||
record = source_records(s, "test_source", "shipments", ids=["30"])[0]
|
||
with pytest.raises(ValueError, match="CORRECTION_RECONFIRM_REQUIRED"):
|
||
service.verified_correction(s, "test_source", latest_facts(s, "test_source"), "SHIPMENT", record)
|
||
|
||
|
||
def test_抵顶预览显示未自动折算且不读取折扣参数():
|
||
from app.services.param_service import _preview_barter
|
||
|
||
def forbidden(_):
|
||
raise AssertionError("线下抵顶不得读取自动折扣参数")
|
||
result = _preview_barter(SimpleNamespace(payload=SimpleNamespace(barter=True), take=forbidden), Decimal("100"))
|
||
assert result.base_amount == Decimal("100")
|
||
assert result.applied is False and result.after_barter is None
|
||
|
||
|
||
def test_正式FIFO台账金额与逐段过程守恒(payroll_case):
|
||
s, user, manager = payroll_case
|
||
business = imported(payroll_case)
|
||
result = payroll_calculation.calculate(s, business.id, "2026-03", operator=user)
|
||
assert result["performance"] == "0.20"
|
||
assert result["payable"] == "0.20"
|
||
ledger = s.get(PerformanceLedger, result["ledger_ids"][0])
|
||
assert ledger.snapshot_json["manager_id"] == manager.id
|
||
assert s.scalar(select(PayrollAllocation)).amount == Decimal("100.00")
|
||
with pytest.raises(BadRequestError, match="已有核算"):
|
||
payroll_calculation.calculate(s, business.id, "2026-03", operator=user)
|
||
|
||
|
||
def test_输入变化使旧预览失效(payroll_case):
|
||
s, user, _ = payroll_case
|
||
pre = business_work(s, user, "BUSINESS_PREVIEW")
|
||
product = s.scalar(select(Product))
|
||
product.product_category = "rail_transit"
|
||
s.commit()
|
||
with pytest.raises(ValueError, match="PREVIEW_STALE"):
|
||
business_work(s, user, "BUSINESS_APPLY", pre)
|
||
s.rollback()
|
||
assert s.scalar(select(func.count()).select_from(Business)) == 0
|
||
|
||
|
||
def test_经理汇总读取固化业务分管(payroll_case):
|
||
s, user, manager = payroll_case
|
||
business = imported(payroll_case)
|
||
payroll_calculation.calculate(s, business.id, "2026-03", operator=user)
|
||
rows = list(s.scalars(select(PerformanceLedger)))
|
||
owner = s.get(Personnel, rows[0].personnel_id)
|
||
owner.manager_id = None
|
||
s.commit()
|
||
buckets = manager_share.collect_team_buckets(s, rows, False)
|
||
assert buckets[manager.id]["performance"] == Decimal("0.20")
|
||
|
||
|
||
def test_生成标记独立于审批状态(payroll_case):
|
||
s, user, _ = payroll_case
|
||
control = payroll_months.lock_month(s, "2026-03", user)
|
||
from datetime import datetime
|
||
|
||
control.generated_at = datetime(2026, 3, 31)
|
||
s.commit()
|
||
with pytest.raises(BadRequestError, match="已锁定"):
|
||
payroll_months.ensure_allocation_editable(s, "2026-03", user)
|
||
s.rollback()
|
||
assert payroll_months.ensure_allocation_editable(s, "2026-04", user).generated_at is None
|
||
|
||
|
||
def test_生成失败不留下月份锁(payroll_case):
|
||
s, user, _ = payroll_case
|
||
from app.services.settlement_service import generate
|
||
|
||
with pytest.raises(BadRequestError, match=r"无可(?:结算|发薪)"):
|
||
generate(s, "2026-03", operator=user)
|
||
assert s.scalar(select(PayrollMonth)) is None
|
||
|
||
|
||
def test_非制作人无法生成工资(payroll_case):
|
||
s, user, _ = payroll_case
|
||
with pytest.raises(ForbiddenError):
|
||
payroll_months.lock_month(s, "2026-03", SimpleNamespace(id=user.id + 1, status=1))
|
||
|
||
|
||
def replace_input(s, record, values):
|
||
newer = PayrollInput(
|
||
business_id=record.business_id,
|
||
version=record.version + 1,
|
||
source_database=record.source_database,
|
||
group_key=record.group_key,
|
||
fingerprint=digest(values),
|
||
data_json=values,
|
||
requested_by=record.requested_by,
|
||
)
|
||
s.add(newer)
|
||
s.commit()
|
||
confirm_test_receipts(s, s.get(User, newer.requested_by))
|
||
return newer
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"amount,performance,payable,start",
|
||
[("50", "0.10", "0.00", False), ("60", "0.12", "0.04", True), ("100", "0.20", "0.20", True)],
|
||
)
|
||
def test_起付百分之六十与平方权重(payroll_case, amount, performance, payable, start):
|
||
s, user, _ = payroll_case
|
||
business = imported(payroll_case)
|
||
record = payroll_calculation.latest_input(s, business.id)
|
||
data = copy.deepcopy(record.data_json)
|
||
data["receipts"][0]["amount"] = amount
|
||
receipt = s.scalar(select(Receipt))
|
||
receipt.amount = Decimal(amount)
|
||
replace_input(s, record, data)
|
||
result = payroll_calculation.calculate(s, business.id, "2026-03", operator=user)
|
||
assert (result["performance"], result["payable"], result["start_payment_ok"]) == (
|
||
performance,
|
||
payable,
|
||
start,
|
||
)
|
||
|
||
|
||
@pytest.mark.parametrize("legacy_date", [date(2026, 2, 10), date(2026, 2, 28)])
|
||
def test_历史发货不重复计入新业务起付分母(payroll_case, legacy_date):
|
||
"""旧应收先冲抵，启用前发货不得再次压低新业务的回款比例。"""
|
||
s, user, _ = payroll_case
|
||
business = imported(payroll_case)
|
||
record = payroll_calculation.latest_input(s, business.id)
|
||
data = copy.deepcopy(record.data_json)
|
||
legacy = Shipment(
|
||
business_id=business.id,
|
||
shipment_no="SH-LEGACY",
|
||
ship_date=legacy_date,
|
||
total_amount=Decimal("100"),
|
||
freight_amount=Decimal("0"),
|
||
)
|
||
s.add(legacy)
|
||
s.flush()
|
||
original = s.get(ShipmentItem, data["maps"]["shipment_items"][data["shipments"][0]["line_id"]])
|
||
item = ShipmentItem(
|
||
shipment_id=legacy.id,
|
||
product_id=original.product_id,
|
||
product_name=original.product_name,
|
||
spec_model=original.spec_model,
|
||
unit=original.unit,
|
||
quantity=Decimal("2"),
|
||
sale_price=Decimal("50"),
|
||
amount=Decimal("100"),
|
||
)
|
||
s.add(item)
|
||
s.flush()
|
||
data["shipments"].append(
|
||
{
|
||
**data["shipments"][0],
|
||
"source_id": "99",
|
||
"line_id": "99",
|
||
"date": legacy_date.isoformat(),
|
||
}
|
||
)
|
||
data["maps"]["shipments"]["99"] = legacy.id
|
||
data["maps"]["shipment_items"]["99"] = item.id
|
||
data["group"]["opening_balance"] = "100"
|
||
data["receipts"][0]["amount"] = "200"
|
||
s.get(Receipt, data["maps"]["receipts"][data["receipts"][0]["source_id"]]).amount = Decimal(
|
||
"200"
|
||
)
|
||
replace_input(s, record, data)
|
||
|
||
result = payroll_calculation.calculate(s, business.id, "2026-03", operator=user)
|
||
|
||
assert result["start_payment_ok"] is True
|
||
assert result["performance"] == result["payable"] == "0.20"
|
||
allocations = list(s.scalars(select(PayrollAllocation)))
|
||
assert sum(row.amount for row in allocations) == Decimal("100.00")
|
||
assert all(row.data_json["shipment_date"] == "2026-03-10" for row in allocations)
|
||
|
||
|
||
def test_跨月续算仅计应付差额且离职未发给接手人(payroll_case):
|
||
s, user, _manager = payroll_case
|
||
business = imported(payroll_case)
|
||
record = payroll_calculation.latest_input(s, business.id)
|
||
data = copy.deepcopy(record.data_json)
|
||
receipt = s.scalar(select(Receipt))
|
||
receipt.amount = Decimal("80")
|
||
data["receipts"][0]["amount"] = "80"
|
||
record = replace_input(s, record, data)
|
||
march = payroll_calculation.calculate(s, business.id, "2026-03", operator=user)
|
||
assert march["payable"] == "0.10"
|
||
old = s.get(PerformanceLedger, march["ledger_ids"][0])
|
||
new_person = Personnel(
|
||
personnel_name="接手业务员",
|
||
business_role_id=s.get(Personnel, old.personnel_id).business_role_id,
|
||
status=1,
|
||
)
|
||
s.add(new_person)
|
||
s.flush()
|
||
second = Receipt(
|
||
business_id=business.id,
|
||
receipt_batch_no="HK-NEXT",
|
||
receipt_date=date(2026, 4, 10),
|
||
amount=Decimal("20"),
|
||
is_barter=0,
|
||
settlement_status=0,
|
||
)
|
||
s.add(second)
|
||
s.flush()
|
||
data = copy.deepcopy(record.data_json)
|
||
data["receipts"].append(
|
||
{
|
||
"source_id": "11",
|
||
"source_no": "HK-NEXT",
|
||
"date": "2026-04-10",
|
||
"group_key": "g",
|
||
"amount": "20",
|
||
"voucher_confirmed": True,
|
||
}
|
||
)
|
||
data["maps"]["receipts"]["11"] = second.id
|
||
data["maps"]["personnel"]["next-person"] = new_person.id
|
||
data["facts"]["personnel"]["next-person"] = {"role_code": "salesperson", "status": 1}
|
||
data["facts"]["handovers"] = [
|
||
{
|
||
"mode": "RESIGNATION",
|
||
"business_key": "biz",
|
||
"from_personnel_id": "00043",
|
||
"to_personnel_id": "next-person",
|
||
"effective_date": "2026-04-01",
|
||
}
|
||
]
|
||
replace_input(s, record, data)
|
||
april = payroll_calculation.calculate(s, business.id, "2026-04", operator=user)
|
||
assert april["payable"] == "0.10"
|
||
assert old.payable_amount == Decimal("0.10") and old.personnel_id != new_person.id
|
||
assert all(
|
||
s.get(PerformanceLedger, lid).personnel_id == new_person.id for lid in april["ledger_ids"]
|
||
)
|
||
assert sum(s.scalars(select(PerformanceLedger.payable_amount))) == Decimal("0.20")
|
||
rebuilt = payroll_calculation.calculate(
|
||
s, business.id, "2026-03", operator=user, reason="复核分次回款"
|
||
)
|
||
assert rebuilt["recalculated_months"] == ["2026-03", "2026-04"]
|
||
active = list(s.scalars(select(PerformanceLedger).where(PerformanceLedger.is_active == 1)))
|
||
assert sum(r.payable_amount for r in active) == Decimal("0.20")
|
||
assert all(r.calc_reason == "复核分次回款" for r in active)
|
||
|
||
|
||
def test_已核算源金额变化不静默改工资(payroll_case):
|
||
s, user, _ = payroll_case
|
||
business = imported(payroll_case)
|
||
receipt = s.scalar(select(Receipt))
|
||
receipt.amount = Decimal("99")
|
||
s.commit()
|
||
with pytest.raises(BadRequestError, match="不一致"):
|
||
payroll_calculation.calculate(s, business.id, "2026-03", operator=user)
|
||
assert s.scalar(select(func.count()).select_from(PerformanceLedger)) == 0
|
||
|
||
|
||
def test_同客户不同合同合并平账_一款对应多发货(payroll_case):
|
||
from tests.unit.test_tplus_workbench import source
|
||
|
||
s, user, _ = payroll_case
|
||
source(
|
||
s,
|
||
"shipments",
|
||
[
|
||
{
|
||
"id": 20,
|
||
"code": "SH-ONE",
|
||
"idcustomer": 8,
|
||
"voucherdate": "2026-03-10",
|
||
"voucherState": 189,
|
||
},
|
||
{
|
||
"id": 22,
|
||
"code": "SH-TWO",
|
||
"idcustomer": 8,
|
||
"voucherdate": "2026-03-11",
|
||
"voucherState": 189,
|
||
},
|
||
],
|
||
)
|
||
source(
|
||
s,
|
||
"shipment_items",
|
||
[
|
||
{
|
||
"id": 21,
|
||
"source_parent_id": 20,
|
||
"idinventory": 9,
|
||
"taxAmount": "100",
|
||
"quantity": "2",
|
||
},
|
||
{
|
||
"id": 23,
|
||
"source_parent_id": 22,
|
||
"idinventory": 9,
|
||
"taxAmount": "100",
|
||
"quantity": "2",
|
||
},
|
||
],
|
||
)
|
||
source(
|
||
s,
|
||
"receipts",
|
||
[
|
||
{
|
||
"id": 10,
|
||
"code": "RC-ONE",
|
||
"idpartner": 8,
|
||
"voucherDate": "2026-03-12",
|
||
"isReceiveFlag": 1,
|
||
"voucherstate": 189,
|
||
"amount": "200",
|
||
}
|
||
],
|
||
)
|
||
fact(
|
||
s,
|
||
"SHIPMENT",
|
||
"23",
|
||
{
|
||
"sales_amount_excludes_freight": True,
|
||
"business_key": "second-contract",
|
||
"personnel_id": "00043",
|
||
},
|
||
)
|
||
fact(
|
||
s,
|
||
"MANAGER",
|
||
"second-manager",
|
||
{"business_key": "second-contract", "manager_id": "90001", "effective_date": "2026-03-01"},
|
||
)
|
||
business = imported(payroll_case)
|
||
assert s.scalar(select(func.count()).select_from(Business)) == 1
|
||
result = payroll_calculation.calculate(s, business.id, "2026-03", operator=user)
|
||
assert len(result["ledger_ids"]) == 2
|
||
assert sum(s.scalars(select(PayrollAllocation.amount))) == Decimal("200.00")
|
||
record = payroll_calculation.latest_input(s, business.id)
|
||
assert record.data_json["group"]["shipment_ids"] == ["20", "22"]
|
||
|
||
|
||
def test_封存发货月指导成本不随后续回款改变(payroll_case):
|
||
from app.models import TPlusFact
|
||
from app.schemas.tplus_workbench import FactIn
|
||
from app.services.tplus import workbench
|
||
|
||
s, user, _ = payroll_case
|
||
cost = s.scalar(select(TPlusFact).where(TPlusFact.kind == "GUIDANCE"))
|
||
with pytest.raises(BadRequestError, match="已封存"):
|
||
workbench.save_fact(
|
||
s,
|
||
FactIn(
|
||
kind="GUIDANCE",
|
||
fact_key=cost.fact_key,
|
||
expected_version=cost.version,
|
||
evidence="试图修改已封存月份",
|
||
data={**cost.data_json, "amount": "30"},
|
||
),
|
||
user.id,
|
||
)
|
||
|
||
|
||
def test_同客户旧项目期初不一致必须先合并事实(payroll_case):
|
||
from app.models import TPlusFact
|
||
|
||
s, user, _ = payroll_case
|
||
group = s.scalar(select(TPlusFact).where(TPlusFact.kind == "GROUP")).data_json
|
||
fact(s, "GROUP", "old-contract-group", {**group, "opening_balance": "10"})
|
||
run = business_work(s, user, "BUSINESS_PREVIEW")
|
||
assert run.status == "PARTIAL"
|
||
assert run.summary_json["counts"] == {"PENDING": 1}
|
||
assert s.scalar(select(func.count()).select_from(Business)) == 0
|