shanxia2.0/backend/tests/test_pl_plan_stop_service.py

"""运输计划多段：装送货产品与计划产品行匹配"""
from decimal import Decimal
from types import SimpleNamespace

import pytest
from fastapi import HTTPException
from sqlalchemy.orm import Session

from app.schemas.pl.plan import PlanStopProductBody
from app.services.pl_plan_stop_service import (
    _resolve_plan_product_id,
    resolve_stop_address_from_library,
)


def _plan_product_row(
    row_id: str,
    product_id: str,
    product_no: str = "P001",
):
    return SimpleNamespace(id=row_id, product_id=product_id, product_no=product_no)


def test_resolve_plan_product_id_by_explicit_id():
    rows = [_plan_product_row("pp-1", "crm-1")]
    prod = PlanStopProductBody(
        planProductId="pp-1",
        productId="crm-1",
        productNo="P001",
        productName="别名1111",
    )
    assert _resolve_plan_product_id(prod, rows) == "pp-1"


def test_resolve_plan_product_id_rejects_explicit_stale_id():
    rows = [_plan_product_row("pp-new", "crm-1")]
    prod = PlanStopProductBody(
        planProductId="pp-old",
        productId="crm-1",
        productNo="P001",
        productName="别名1111",
    )
    with pytest.raises(HTTPException) as error:
        _resolve_plan_product_id(prod, rows)
    assert error.value.status_code == 422


def test_resolve_plan_product_id_matches_unique_product_when_id_omitted():
    rows = [_plan_product_row("pp-new", "crm-1", "P001")]
    prod = PlanStopProductBody(
        productId="crm-1",
        productNo="P001",
        productName="别名1111",
    )
    assert _resolve_plan_product_id(prod, rows) == "pp-new"


def test_resolve_plan_product_id_raises_when_unmatched():
    rows = [_plan_product_row("pp-new", "crm-2", "P002")]
    prod = PlanStopProductBody(
        planProductId="pp-old",
        productId="crm-1",
        productNo="P001",
        productName="别名1111",
    )
    with pytest.raises(HTTPException) as exc_info:
        _resolve_plan_product_id(prod, rows)
    assert exc_info.value.status_code == 422
    assert "别名1111" in exc_info.value.detail


def test_resolve_stop_address_from_library_returns_existing_text():
    db = SimpleNamespace()
    assert (
        resolve_stop_address_from_library(db, "shanxia2.0", "已有地址", None, None)
        == "已有地址"
    )


def test_resolve_stop_address_from_library_looks_up_by_lat():
    row = SimpleNamespace(
        full_address="高新区霞虹路5号安庆市山峡化工有限公司",
        lng=Decimal("116.990968"),
        lat=Decimal("30.549796"),
    )

    class FakeCol:
        @staticmethod
        def desc():
            return FakeCol()

    class FakeAddressModel:
        tenant_no = None
        del_flag = None
        status = None
        lat = None
        lng = None
        create_time = FakeCol()

    class FakeQuery:
        def filter(self, *args, **kwargs):
            return self

        def order_by(self, *args, **kwargs):
            return self

        def all(self):
            return [row]

    class FakeSession:
        def query(self, model):
            return FakeQuery()

    import app.services.pl_plan_stop_service as mod

    original = mod.PlTransportAddress
    mod.PlTransportAddress = FakeAddressModel
    try:
        resolved = resolve_stop_address_from_library(
            FakeSession(),
            "shanxia2.0",
            "",
            None,
            Decimal("30.549796"),
        )
    finally:
        mod.PlTransportAddress = original

    assert resolved == "高新区霞虹路5号安庆市山峡化工有限公司"