77 lines
3.9 KiB
Python
77 lines
3.9 KiB
Python
"""日历名称只查询已授权计划引用的客户，简称缺失时显示全称。"""
|
|
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"] == "送货全称"
|