388 lines
13 KiB
Python
388 lines
13 KiB
Python
from datetime import datetime
|
|
from decimal import Decimal
|
|
|
|
import pytest
|
|
|
|
from app.services.sf_express_settlement import (
|
|
assert_freight_mutable,
|
|
assert_settle_request,
|
|
freight_amount_label,
|
|
plan_fill_amount,
|
|
plan_settle,
|
|
resolve_header_for_save,
|
|
)
|
|
|
|
|
|
def test_new_sf_express_nulls_amount_and_requires_waybill():
|
|
plan = resolve_header_for_save(
|
|
transport_mode="sfkd",
|
|
existing_waybill=None,
|
|
existing_amount=None,
|
|
existing_settle_status=None,
|
|
waybill_supplied=True,
|
|
waybill_no=" SF001 ",
|
|
amount_supplied=True,
|
|
freight_amount=Decimal("88"),
|
|
)
|
|
assert plan.waybill_no == "SF001"
|
|
assert plan.freight_amount is None
|
|
assert plan.settle_status == "wjs"
|
|
assert plan.clear_settle_time is True
|
|
assert freight_amount_label("sfkd", plan.freight_amount) == "月结"
|
|
|
|
|
|
def test_sf_express_rejects_blank_and_long_waybill():
|
|
with pytest.raises(ValueError, match="请填写顺丰单号"):
|
|
resolve_header_for_save(
|
|
transport_mode="sfkd",
|
|
existing_waybill=None,
|
|
existing_amount=None,
|
|
existing_settle_status=None,
|
|
waybill_supplied=True,
|
|
waybill_no=" ",
|
|
amount_supplied=True,
|
|
freight_amount=None,
|
|
)
|
|
with pytest.raises(ValueError, match="顺丰单号不能超过 64 个字符"):
|
|
resolve_header_for_save(
|
|
transport_mode="sfkd",
|
|
existing_waybill=None,
|
|
existing_amount=None,
|
|
existing_settle_status=None,
|
|
waybill_supplied=True,
|
|
waybill_no="S" * 65,
|
|
amount_supplied=False,
|
|
freight_amount=None,
|
|
)
|
|
|
|
|
|
def test_non_sf_requires_amount_and_clears_waybill_and_settlement():
|
|
plan = resolve_header_for_save(
|
|
transport_mode="zxc",
|
|
existing_waybill="SF001",
|
|
existing_amount=None,
|
|
existing_settle_status="wjs",
|
|
waybill_supplied=False,
|
|
waybill_no=None,
|
|
amount_supplied=True,
|
|
freight_amount="12.3",
|
|
)
|
|
assert plan.waybill_no is None
|
|
assert plan.freight_amount == Decimal("12.30")
|
|
assert plan.settle_status is None
|
|
assert plan.clear_settle_time is True
|
|
|
|
|
|
def test_other_mode_named_sf_still_requires_amount():
|
|
plan = resolve_header_for_save(
|
|
transport_mode="sf",
|
|
existing_waybill=None,
|
|
existing_amount=None,
|
|
existing_settle_status=None,
|
|
waybill_supplied=True,
|
|
waybill_no="",
|
|
amount_supplied=True,
|
|
freight_amount="12.30",
|
|
)
|
|
assert plan.freight_amount == Decimal("12.30")
|
|
assert plan.waybill_no is None
|
|
assert plan.settle_status is None
|
|
|
|
|
|
def test_non_sf_rejects_missing_amount_and_negative_amount():
|
|
with pytest.raises(ValueError, match="请填写运费"):
|
|
resolve_header_for_save(
|
|
transport_mode="zxc",
|
|
existing_waybill=None,
|
|
existing_amount=None,
|
|
existing_settle_status=None,
|
|
waybill_supplied=True,
|
|
waybill_no="",
|
|
amount_supplied=True,
|
|
freight_amount=None,
|
|
)
|
|
with pytest.raises(ValueError, match="运费不能为负数"):
|
|
resolve_header_for_save(
|
|
transport_mode="zxc",
|
|
existing_waybill=None,
|
|
existing_amount=None,
|
|
existing_settle_status=None,
|
|
waybill_supplied=True,
|
|
waybill_no="",
|
|
amount_supplied=True,
|
|
freight_amount="-1",
|
|
)
|
|
|
|
|
|
def test_settled_header_cannot_change():
|
|
with pytest.raises(ValueError, match="已结算的运费不能修改运输方式或顺丰单号"):
|
|
resolve_header_for_save(
|
|
transport_mode="sfkd",
|
|
existing_waybill="SF001",
|
|
existing_amount=None,
|
|
existing_settle_status="yjs",
|
|
waybill_supplied=True,
|
|
waybill_no="SF002",
|
|
amount_supplied=False,
|
|
freight_amount=None,
|
|
)
|
|
|
|
|
|
def test_blank_settle_keeps_null_and_historical_amount():
|
|
now = datetime(2026, 10, 21, 12, 0, 0)
|
|
assert plan_settle(current_amount=None, submitted_amount=" ", now=now) == (
|
|
None,
|
|
"yjs",
|
|
now,
|
|
)
|
|
assert plan_settle(current_amount=Decimal("10.00"), submitted_amount=None, now=now) == (
|
|
Decimal("10.00"),
|
|
"yjs",
|
|
now,
|
|
)
|
|
amount, status, settle_time = plan_settle(
|
|
current_amount=Decimal("10.00"), submitted_amount="20", now=now
|
|
)
|
|
assert amount == Decimal("20.00")
|
|
assert status == "yjs"
|
|
assert settle_time == now
|
|
|
|
|
|
def test_batch_settle_rejects_amount_and_fill_keeps_only_monthly_rows():
|
|
assert_settle_request(["a", "b"], None)
|
|
with pytest.raises(ValueError, match="批量结算不能填写运费"):
|
|
assert_settle_request(["a", "b"], "10")
|
|
with pytest.raises(ValueError, match="请选择运费批次"):
|
|
assert_settle_request([], None)
|
|
assert plan_fill_amount(
|
|
current_amount=None, settle_status="yjs", submitted_amount="6.5"
|
|
) == Decimal("6.50")
|
|
with pytest.raises(ValueError, match="只有已结算且仍为月结的运费可以补填金额"):
|
|
plan_fill_amount(
|
|
current_amount=Decimal("1.00"), settle_status="yjs", submitted_amount="2"
|
|
)
|
|
|
|
|
|
def test_settled_batch_cannot_delete_or_change_samples():
|
|
with pytest.raises(ValueError, match="已结算的运费不能删除"):
|
|
assert_freight_mutable("yjs", "delete")
|
|
with pytest.raises(ValueError, match="已结算的运费不能调整样本"):
|
|
assert_freight_mutable("yjs", "items")
|
|
assert_freight_mutable("wjs", "delete")
|
|
assert freight_amount_label("sfkd", Decimal("8")) == "8.00"
|
|
assert freight_amount_label("zxc", None) == ""
|
|
|
|
|
|
def test_freight_model_allows_null_amount_and_settlement_columns():
|
|
from app.models.models import BizSampleFreight
|
|
|
|
cols = {column.name: column for column in BizSampleFreight.__table__.columns}
|
|
assert cols["freight_amount"].nullable is True
|
|
assert cols["waybill_no"].nullable is True
|
|
assert cols["settle_status"].nullable is True
|
|
assert cols["settle_time"].nullable is True
|
|
|
|
|
|
def test_amount_serializer_keeps_null_instead_of_zero():
|
|
from types import SimpleNamespace
|
|
from app.crud.sample_freight import _row_to_dict
|
|
from app.services.market_freight import _batch_dict
|
|
|
|
obj = SimpleNamespace(
|
|
id="f1",
|
|
freight_no="YF1",
|
|
transport_mode="sfkd",
|
|
freight_amount=None,
|
|
waybill_no=None,
|
|
settle_status="wjs",
|
|
settle_time=None,
|
|
tenant_no="zkxw",
|
|
order_num=0,
|
|
remark="",
|
|
create_by="u1",
|
|
create_time=None,
|
|
update_time=None,
|
|
)
|
|
admin = _row_to_dict(obj, transport_mode_label="顺丰快递", create_by_name="张三")
|
|
market = _batch_dict(obj, 1, "顺丰快递", "张三")
|
|
assert admin["freightAmount"] is None
|
|
assert admin["waybillNo"] == ""
|
|
assert admin["settleStatus"] == "wjs"
|
|
assert admin["freightAmountLabel"] == "月结"
|
|
assert market["freightAmount"] is None
|
|
assert market["freightAmountLabel"] == "月结"
|
|
|
|
|
|
def test_sample_card_payload_keeps_null_amount():
|
|
from app.crud.sample_freight import freight_amount_payload
|
|
|
|
monthly = freight_amount_payload(None, "sfkd", True)
|
|
assert monthly["freightAmount"] is None
|
|
assert monthly["freightAmountLabel"] == "月结"
|
|
assert freight_amount_payload(None, "sfkd", False)["freightAmount"] is None
|
|
assert freight_amount_payload(None, "zxc", True)["freightAmountLabel"] == ""
|
|
|
|
numeric = freight_amount_payload(Decimal("88"), "zxc", True)
|
|
assert numeric["freightAmount"] == "88.00"
|
|
assert numeric["freightAmountLabel"] == "88.00"
|
|
assert freight_amount_payload(Decimal("0"), "zxc", False)["freightAmount"] == 0.0
|
|
|
|
|
|
def test_admin_and_market_bodies_accept_sf_without_amount():
|
|
from app.api.sample_freight import SampleFreightCreateBody, SampleFreightUpdateBody
|
|
from app.schemas.market_mobile import (
|
|
MarketFreightCreateBody,
|
|
MarketFreightUpdateBody,
|
|
MarketProtocolSampleFreightDraftBody,
|
|
)
|
|
|
|
admin = SampleFreightCreateBody.model_validate({
|
|
"transportMode": "sfkd",
|
|
"waybillNo": "SF100",
|
|
"items": [{"sampleSource": "cell_storage", "sampleId": "s1"}],
|
|
})
|
|
assert admin.freight_amount is None
|
|
assert admin.waybill_no == "SF100"
|
|
updated = SampleFreightUpdateBody.model_validate({"waybillNo": "SF101"})
|
|
assert "waybill_no" in updated.model_fields_set
|
|
market = MarketFreightCreateBody.model_validate({
|
|
"transportMode": "sfkd",
|
|
"waybillNo": "SF100",
|
|
"items": [{"sampleSource": "cell_storage", "sampleId": "s1"}],
|
|
})
|
|
assert market.freight_amount is None
|
|
assert market.waybill_no == "SF100"
|
|
draft = MarketProtocolSampleFreightDraftBody.model_validate({
|
|
"clientKey": "fd_1",
|
|
"transportMode": "sfkd",
|
|
"waybillNo": "SF100",
|
|
})
|
|
assert draft.waybill_no == "SF100"
|
|
MarketFreightUpdateBody.model_validate({"waybillNo": "SF100"})
|
|
|
|
|
|
def _protocol_freight_db():
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from app.models.base import Base
|
|
from app.models.models import (
|
|
BizCellSampleType,
|
|
BizCollectionProtocol,
|
|
BizProtocolPartyCustomer,
|
|
BizProtocolSample,
|
|
BizSampleFreight,
|
|
BizSampleFreightItem,
|
|
SysDictData,
|
|
)
|
|
|
|
engine = create_engine("sqlite:///:memory:")
|
|
Base.metadata.create_all(
|
|
engine,
|
|
tables=[
|
|
BizSampleFreight.__table__,
|
|
BizSampleFreightItem.__table__,
|
|
BizProtocolSample.__table__,
|
|
BizCollectionProtocol.__table__,
|
|
BizCellSampleType.__table__,
|
|
BizProtocolPartyCustomer.__table__,
|
|
SysDictData.__table__,
|
|
],
|
|
)
|
|
return sessionmaker(bind=engine)()
|
|
|
|
|
|
def _seed_settled_protocol_freight(db, sample_id="s-old"):
|
|
from app.models.models import BizSampleFreight, BizSampleFreightItem
|
|
|
|
now = datetime(2026, 10, 21, 9, 0, 0)
|
|
freight = BizSampleFreight(
|
|
id="f-yjs",
|
|
freight_no="YF1",
|
|
transport_mode="sfkd",
|
|
freight_amount=None,
|
|
waybill_no="SF001",
|
|
settle_status="yjs",
|
|
tenant_no="tenant-a",
|
|
create_dept_id="",
|
|
create_time=now,
|
|
update_time=now,
|
|
del_flag=0,
|
|
)
|
|
item = BizSampleFreightItem(
|
|
id="fi-1",
|
|
freight_id=freight.id,
|
|
sample_source="cell_storage",
|
|
sample_id=sample_id,
|
|
sample_code="C1",
|
|
protocol_id="p1",
|
|
tenant_no="tenant-a",
|
|
create_dept_id="",
|
|
create_time=now,
|
|
update_time=now,
|
|
del_flag=0,
|
|
)
|
|
db.add_all([freight, item])
|
|
db.commit()
|
|
return freight, item
|
|
|
|
|
|
def test_settled_protocol_save_without_freight_id_keeps_links():
|
|
from types import SimpleNamespace
|
|
|
|
from app.crud.sample_freight import apply_freight_on_protocol_save
|
|
from app.models.models import BizSampleFreight, BizSampleFreightItem
|
|
|
|
db = _protocol_freight_db()
|
|
freight, item = _seed_settled_protocol_freight(db)
|
|
try:
|
|
with pytest.raises(ValueError, match="已结算的运费不能调整样本"):
|
|
apply_freight_on_protocol_save(
|
|
db,
|
|
tenant_no="tenant-a",
|
|
user_id="u1",
|
|
sample_source="cell_storage",
|
|
protocol_id="p1",
|
|
saved_samples=[SimpleNamespace(id="s-new", sample_code="C2")],
|
|
input_samples=[{}],
|
|
)
|
|
db.expire_all()
|
|
item = db.query(BizSampleFreightItem).filter_by(id=item.id).one()
|
|
freight = db.query(BizSampleFreight).filter_by(id=freight.id).one()
|
|
assert item.del_flag == 0
|
|
assert freight.del_flag == 0
|
|
finally:
|
|
db.close()
|
|
|
|
|
|
def test_settled_protocol_save_relinks_same_count_to_new_sample():
|
|
from types import SimpleNamespace
|
|
|
|
from app.crud.sample_freight import apply_freight_on_protocol_save
|
|
from app.models.models import BizSampleFreight, BizSampleFreightItem
|
|
|
|
db = _protocol_freight_db()
|
|
freight, _item = _seed_settled_protocol_freight(db)
|
|
try:
|
|
apply_freight_on_protocol_save(
|
|
db,
|
|
tenant_no="tenant-a",
|
|
user_id="u1",
|
|
sample_source="cell_storage",
|
|
protocol_id="p1",
|
|
saved_samples=[SimpleNamespace(id="s-new", sample_code="C2")],
|
|
input_samples=[{"freightId": freight.id}],
|
|
)
|
|
db.expire_all()
|
|
freight = db.query(BizSampleFreight).filter_by(id=freight.id).one()
|
|
assert freight.del_flag == 0
|
|
assert freight.settle_status == "yjs"
|
|
live = (
|
|
db.query(BizSampleFreightItem)
|
|
.filter_by(freight_id=freight.id, sample_id="s-new", del_flag=0)
|
|
.one()
|
|
)
|
|
assert live.protocol_id == "p1"
|
|
finally:
|
|
db.close()
|