shanxia2.0/backend/tests/test_pl_calendar_customer_names.py

"""日历名称只查询已授权计划引用的客户，简称缺失时显示全称。"""
from types import SimpleNamespace as NS

from sqlalchemy import create_engine, text
from sqlalchemy.orm import Session

from app.services.pl_calendar_customer_names import attach_calendar_routes


def test_short_names_full_names_stop_order_and_tenant_isolation():
    engine = create_engine("sqlite://")
    with engine.begin() as conn:
        conn.execute(text("CREATE TABLE crm_customer (id TEXT, customer_name TEXT, short_name TEXT, tenant_no TEXT, del_flag INTEGER)"))
        conn.execute(text("INSERT INTO crm_customer VALUES (:id, :name, :short, :tenant, :deleted)"), [
            dict(id="load", name="山峡化工有限公司", short=" 山峡 ", tenant="t1", deleted=0),
            dict(id="full", name="永恒有限公司", short="  ", tenant="t1", deleted=0),
            dict(id="other", name="其他租户", short="不应泄漏", tenant="t2", deleted=0),
            dict(id="deleted", name="已删除", short="不应显示", tenant="t1", deleted=1),
            dict(id="unused", name="无关客户", short="无关", tenant="t1", deleted=0),
        ])
    plan = NS(load=NS(customer_id="load", customer_name="山峡旧全称"), deliveries=[
        NS(seq=3, customer_id="deleted", customer_name="删除前全称"),
        NS(seq=1, customer_id="full", customer_name="永恒旧全称"),
        NS(seq=2, customer_id="other", customer_name="计划留存全称"),
    ])
    with Session(engine) as db:
        attach_calendar_routes(db, tenant_no="t1", plans=[plan])
    assert plan.calendar_route == "山峡 → 永恒有限公司 → 计划留存全称 → 删除前全称"
    assert plan.load.calendar_customer_name == "山峡"
    assert [stop.calendar_customer_name for stop in plan.deliveries] == ["删除前全称", "永恒有限公司", "计划留存全称"]
    assert plan.load.customer_name == "山峡旧全称"


def test_legacy_plan_and_address_fallback_without_customer_ids():
    class NoQuery:
        def query(self, *args):
            raise AssertionError("没有客户 ID 时不应查询客户表")
    plans = [
        NS(supplier_name="装货全称", customer_name="送货全称"),
        NS(load=NS(address="装货地址"), deliveries=[NS(seq=1, address="送货地址")]),
        NS(),
    ]
    attach_calendar_routes(NoQuery(), tenant_no="t1", plans=plans)
    assert [p.calendar_route for p in plans] == ["装货全称 → 送货全称", "装货地址 → 送货地址", "去向待定"]


def test_load_name_without_id_does_not_get_replaced_by_supplier():
    class SupplierQuery:
        def filter(self, *args):
            return self

        def all(self):
            return [NS(id="supplier", short_name="供方简称", customer_name="供方全称")]

    class SupplierDb:
        def query(self, *args):
            return SupplierQuery()

    plan = NS(supplier_id="supplier", supplier_name="供方全称", load=NS(customer_name="实际装货站点"))
    attach_calendar_routes(SupplierDb(), tenant_no="t1", plans=[plan])
    assert plan.calendar_route == "实际装货站点"


def test_read_schemas_serialize_calendar_stop_names():
    from app.schemas.pl.plan import PlanLoadRead, PlanDeliveryRead

    class NoQuery:
        def query(self, *args):
            raise AssertionError("无关联客户不查询")

    fields = dict(planId="p1", tenantNo="t1", createTime="2026-10-25T00:00:00", updateTime="2026-10-25T00:00:00")
    load = PlanLoadRead.model_validate(dict(id="load", customerName="装货全称", **fields))
    delivery = PlanDeliveryRead.model_validate(dict(id="delivery", customerName="送货全称", **fields))
    attach_calendar_routes(NoQuery(), tenant_no="t1", plans=[NS(load=load, deliveries=[delivery])])
    assert load.model_dump(by_alias=True)["calendarCustomerName"] == "装货全称"
    assert delivery.model_dump(by_alias=True)["calendarCustomerName"] == "送货全称"