1e53792ad6
打通 T+ 快照、业务、发货、回款和核算准备链路，保留来源差异并提供待补状态。 新增业务自动接入任务、跨平台读取服务托管、权限复核、来源关联和前端核对页面。 补充迁移、隔离集成测试、浏览器模拟验收及角色授权边界修复。
107 lines
5.8 KiB
Python
107 lines
5.8 KiB
Python
"""核算资料参照必须能完整遍历大于单页的发货和人员。"""
|
|
|
|
from datetime import date
|
|
from decimal import Decimal
|
|
|
|
import pytest
|
|
from sqlalchemy import select
|
|
|
|
from app.models import (
|
|
Business,
|
|
BusinessRole,
|
|
Personnel,
|
|
Shipment,
|
|
ShipmentItem,
|
|
TPlusFact,
|
|
TPlusSourceLink,
|
|
User,
|
|
)
|
|
from app.services import authorization_metadata, authorization_service
|
|
from tests.unit.test_tplus_business_sync import dataset, sync
|
|
from tests.unit.test_tplus_workbench import work_session as work_session
|
|
|
|
|
|
def test_核算参照按页返回超过二百条的发货与人员(work_session, monkeypatch):
|
|
session = work_session
|
|
dataset(session, shipping=True)
|
|
sync(session)
|
|
business = session.scalar(select(Business))
|
|
role = session.scalar(select(BusinessRole).where(BusinessRole.role_code == "salesperson"))
|
|
for number in range(205):
|
|
shipment = Shipment(business_id=business.id, shipment_no=f"P{number:04}",
|
|
ship_date=date(2026, 3, 1), total_amount=Decimal("1"))
|
|
session.add(shipment)
|
|
session.flush()
|
|
session.add(ShipmentItem(shipment_id=shipment.id, product_name="测试产品", spec_model="",
|
|
quantity=Decimal("1"), amount=Decimal("1")))
|
|
session.add(Personnel(personnel_name=f"测试人员{number:04}", business_role_id=role.id))
|
|
session.flush()
|
|
monkeypatch.setattr(authorization_service, "require_permission", lambda *args: None)
|
|
first = authorization_metadata.payroll_options(session, session.get(User, 1), business.id, page_num=1, page_size=200)
|
|
second = authorization_metadata.payroll_options(session, session.get(User, 1), business.id, page_num=2, page_size=200)
|
|
assert first["business_key"] == second["business_key"] == "customer:8"
|
|
assert first["shipments_total"] == second["shipments_total"] == 206
|
|
assert first["personnel_total"] == second["personnel_total"] == 205
|
|
assert (len(first["shipments"]), len(second["shipments"])) == (200, 6)
|
|
assert (len(first["personnel"]), len(second["personnel"])) == (200, 5)
|
|
assert len({row["id"] for row in first["shipments"] + second["shipments"]}) == 206
|
|
|
|
|
|
def test_来源人员只按已确认身份关联预填且未匹配保持空值(work_session, monkeypatch):
|
|
session = work_session
|
|
dataset(session, shipping=True)
|
|
sync(session)
|
|
business = session.scalar(select(Business))
|
|
monkeypatch.setattr(authorization_service, "require_permission", lambda *args: None)
|
|
before = authorization_metadata.payroll_options(session, session.get(User, 1), business.id)
|
|
assert before["shipments"][0]["source_personnel_name"] == "来源员工"
|
|
assert before["shipments"][0]["matched_personnel_id"] is None
|
|
role = session.scalar(select(BusinessRole).where(BusinessRole.role_code == "salesperson"))
|
|
person = Personnel(personnel_name="来源员工", business_role_id=role.id)
|
|
session.add(person)
|
|
session.flush()
|
|
# 相同姓名本身不是身份关联。
|
|
unmatched = authorization_metadata.payroll_options(session, session.get(User, 1), business.id)
|
|
assert unmatched["shipments"][0]["matched_personnel_id"] is None
|
|
other_source = TPlusSourceLink(source_database="other_source", entity_type="personnel", source_id="1", target_id=person.id)
|
|
session.add(other_source)
|
|
session.flush()
|
|
assert authorization_metadata.payroll_options(session, session.get(User, 1), business.id)["shipments"][0]["matched_personnel_id"] is None
|
|
session.delete(other_source)
|
|
session.flush()
|
|
session.add(TPlusSourceLink(source_database="test_source", entity_type="personnel", source_id="1", target_id=person.id))
|
|
session.flush()
|
|
matched = authorization_metadata.payroll_options(session, session.get(User, 1), business.id)
|
|
assert matched["shipments"][0]["matched_personnel_id"] == person.id
|
|
|
|
|
|
def test_核算参照复用同来源已封存的单位指导成本(work_session, monkeypatch):
|
|
session = work_session
|
|
dataset(session, shipping=True)
|
|
sync(session)
|
|
business = session.scalar(select(Business))
|
|
item = session.scalar(select(ShipmentItem))
|
|
monkeypatch.setattr(authorization_service, "require_permission", lambda *args: None)
|
|
initial = authorization_metadata.payroll_options(session, session.get(User, 1), business.id)["shipments"][0]
|
|
assert initial["guidance_status"] == "MISSING" and initial["guidance_unit_cost"] is None
|
|
data = {"product_id": "9", "month": "2026-03", "basis": "UNIT",
|
|
"amount": "2.1234", "unit": item.unit, "is_finalized": True}
|
|
session.add(TPlusFact(source_database="test_source", kind="GUIDANCE", fact_key="price9-03",
|
|
version=1, data_json=data, evidence="历史记录", requested_by=1))
|
|
session.flush()
|
|
ready = authorization_metadata.payroll_options(session, session.get(User, 1), business.id)["shipments"][0]
|
|
assert (ready["guidance_status"], ready["guidance_unit_cost"]) == ("FINALIZED_UNIT", "2.1234")
|
|
session.add(TPlusFact(source_database="test_source", kind="GUIDANCE", fact_key="duplicate9-03",
|
|
version=1, data_json=data, evidence="历史重复记录", requested_by=1))
|
|
session.flush()
|
|
conflict = authorization_metadata.payroll_options(session, session.get(User, 1), business.id)["shipments"][0]
|
|
assert conflict["guidance_status"] == "AMBIGUOUS" and conflict["guidance_unit_cost"] is None
|
|
|
|
|
|
@pytest.mark.parametrize("page_num,page_size", [(0, 200), (1, 0), (1, 201)])
|
|
def test_核算参照拒绝无效页码与页大小(work_session, monkeypatch, page_num, page_size):
|
|
monkeypatch.setattr(authorization_service, "require_permission", lambda *args: None)
|
|
with pytest.raises(ValueError):
|
|
authorization_metadata.payroll_options(work_session, work_session.get(User, 1), 1,
|
|
page_num=page_num, page_size=page_size)
|