zkxw/backend/tests/test_sf_express_settlement.py

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()


from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from app.models.base import Base
from app.models.models import BizSampleFreight, SysUser


def _amount_equals(actual, expected: str) -> None:
    """SQLite 可能把 Numeric 读成 float，按数值比较，空金额必须保持 None。"""
    assert actual is not None
    assert Decimal(str(actual)) == Decimal(expected)


@pytest.fixture
def db():
    engine = create_engine("sqlite:///:memory:")
    Base.metadata.create_all(
        engine,
        tables=[SysUser.__table__, BizSampleFreight.__table__],
    )
    session = sessionmaker(bind=engine)()
    now = datetime(2026, 10, 21, 9, 0, 0)
    common = dict(
        tenant_no="tenant-a",
        create_dept_id="d1",
        create_time=now,
        update_time=now,
        del_flag=0,
    )
    session.add(
        SysUser(
            id="u1",
            user_no="u1",
            nick_name="张三",
            email="",
            phone_no="",
            password="",
            status="1",
            dept_id="d1",
            **common,
        )
    )
    session.add_all(
        [
            BizSampleFreight(
                id="sf-null",
                freight_no="YF1",
                transport_mode="sfkd",
                freight_amount=None,
                waybill_no="SF001",
                settle_status="wjs",
                create_by="u1",
                **common,
            ),
            BizSampleFreight(
                id="sf-old",
                freight_no="YF2",
                transport_mode="sfkd",
                freight_amount=Decimal("10.00"),
                waybill_no="",
                settle_status="wjs",
                create_by="u1",
                **common,
            ),
            BizSampleFreight(
                id="line",
                freight_no="YF3",
                transport_mode="zxc",
                freight_amount=Decimal("5.00"),
                create_by="u1",
                **common,
            ),
            BizSampleFreight(
                id="other-tenant",
                freight_no="YF4",
                transport_mode="sfkd",
                freight_amount=None,
                waybill_no="SF999",
                settle_status="wjs",
                create_by="u1",
                tenant_no="tenant-b",
                create_dept_id="d1",
                create_time=now,
                update_time=now,
                del_flag=0,
            ),
        ]
    )
    session.commit()
    try:
        yield session
    finally:
        session.close()


USER = {
    "user_id": "u1",
    "tenant_no": "tenant-a",
    "role": "super_admin",
    "dept_id": "d1",
}


def test_list_returns_only_current_tenant_sf_rows(db):
    from app.crud.sf_express_settlement import list_sf_settlements

    rows, total = list_sf_settlements(
        db, tenant_no="tenant-a", user_info=USER, page=1, page_size=20
    )
    assert total == 2
    assert {row["id"] for row in rows} == {"sf-null", "sf-old"}
    assert all(row["transportMode"] == "sfkd" for row in rows)
    null_row = next(row for row in rows if row["id"] == "sf-null")
    assert null_row["freightAmountLabel"] == "月结"
    assert null_row["createByName"] == "张三"
    assert null_row["waybillNo"] == "SF001"


def test_settle_blank_keeps_null_and_historical_amount(db):
    from app.crud.sf_express_settlement import settle_sf_freights

    settle_sf_freights(
        db,
        tenant_no="tenant-a",
        user_id="u1",
        user_info=USER,
        ids=["sf-null"],
        freight_amount=None,
    )
    settle_sf_freights(
        db,
        tenant_no="tenant-a",
        user_id="u1",
        user_info=USER,
        ids=["sf-old"],
        freight_amount="",
    )
    assert db.query(BizSampleFreight).filter_by(id="sf-null").one().freight_amount is None
    _amount_equals(
        db.query(BizSampleFreight).filter_by(id="sf-old").one().freight_amount, "10"
    )
    assert db.query(BizSampleFreight).filter_by(id="sf-null").one().settle_status == "yjs"
    with pytest.raises(ValueError, match="该运费已结算"):
        settle_sf_freights(
            db,
            tenant_no="tenant-a",
            user_id="u1",
            user_info=USER,
            ids=["sf-null"],
            freight_amount=None,
        )


def test_fill_amount_does_not_change_settle_time(db):
    from app.crud.sf_express_settlement import fill_sf_freight_amount, settle_sf_freights

    settle_sf_freights(
        db,
        tenant_no="tenant-a",
        user_id="u1",
        user_info=USER,
        ids=["sf-null"],
        freight_amount=None,
    )
    settled_at = db.query(BizSampleFreight).filter_by(id="sf-null").one().settle_time
    fill_sf_freight_amount(
        db,
        tenant_no="tenant-a",
        user_id="u1",
        user_info=USER,
        freight_id="sf-null",
        freight_amount="15",
    )
    row = db.query(BizSampleFreight).filter_by(id="sf-null").one()
    _amount_equals(row.freight_amount, "15.00")
    assert row.settle_time == settled_at


def test_export_empty_filter_raises_chinese_message(db):
    from app.crud.sf_express_settlement import list_sf_settlements_for_export

    with pytest.raises(ValueError, match="没有可导出的数据"):
        list_sf_settlements_for_export(
            db, tenant_no="tenant-a", user_info=USER, waybill_no="不存在"
        )


def test_market_router_has_no_settlement_paths():
    from app.api.market_mobile import router as market_router
    from app.api.sample_freight import router as admin_router

    market_paths = [getattr(route, "path", "") for route in market_router.routes]
    admin_paths = [getattr(route, "path", "") for route in admin_router.routes]
    assert all("sf-settlement" not in path for path in market_paths)
    assert "/web/admin/sample-freights/sf-settlements" in admin_paths
    assert "/web/admin/sample-freights/sf-settlements/export" in admin_paths