shanxia2.0/backend/tests/test_pl_mobile_stop_meta.py

"""Pure mobile snapshot preservation tests; no DB imports or connections."""
from pathlib import Path
from types import SimpleNamespace as Row
import importlib.util
import pytest


def service():
    path = Path(__file__).parents[1] / 'app/services/pl_mobile_stop_meta.py'
    assert path.exists()
    spec=importlib.util.spec_from_file_location('mobile_meta_test_module',path)
    module=importlib.util.module_from_spec(spec);spec.loader.exec_module(module)
    return module


def snapshot(**overrides):
    fields=dict(id='meta',tenant_no='t',plan_id='plan',stop_type='delivery',stop_id='stop-a',stop_product_id='old-row',contract_id='c',contract_product_id='line-a',product_id='p',package_spec='桶',model='',unit='吨',alias_name='送货别名',brand_mode='required',brand_id='delivery-brand',brand_name='送货品牌',product_name='送货别名',source_product_name='装货别名',source_brand_mode='any',source_brand_id='',source_brand_name='')
    return Row(**(fields|overrides))


def incoming(**overrides):
    return Row(**(dict(id=None,contract_id='c',contract_product_id='line-a',product_id='p',package_spec='桶',model='',unit='吨',product_name='送货别名',mobile_product_meta=None)|overrides))


def plan_product(**overrides):
    return Row(**(dict(product_name='装货别名',brand_mode='any',brand_id='',brand_name='')|overrides))


def test_disabled_mobile_gate_explains_without_loading_models_or_db():
    m=service()
    m.assert_mobile_meta_enabled(False,False)
    with pytest.raises(Exception) as error:m.assert_mobile_meta_enabled(False,True)
    assert error.value.status_code == 503
    assert '启用' in error.value.detail


def test_web_unrelated_edit_carries_delivery_snapshot_and_consumes_once():
    m=service();used=set()
    result=m.resolve_stop_meta([snapshot()],used,'t','plan','delivery','stop-a',incoming(),plan_product())
    assert result==dict(alias_name='送货别名',brand_mode='required',brand_id='delivery-brand',brand_name='送货品牌')
    assert m.resolve_stop_meta([snapshot()],used,'t','plan','delivery','stop-a',incoming(),plan_product()) is None


@pytest.mark.parametrize('change',[dict(contract_product_id='line-b'),dict(package_spec='散装'),dict(product_id='q')])
def test_replaced_line_does_not_inherit_snapshot(change):
    assert service().resolve_stop_meta([snapshot()],set(),'t','plan','delivery','stop-a',incoming(**change),plan_product()) is None


@pytest.mark.parametrize('tenant,stop', [('other','stop-a'),('t','stop-b')])
def test_cross_tenant_or_stop_cannot_carry(tenant,stop):
    assert service().resolve_stop_meta([snapshot()],set(),tenant,'plan','delivery',stop,incoming(),plan_product()) is None


def test_web_explicit_legal_name_and_brand_changes_win():
    result=service().resolve_stop_meta([snapshot()],set(),'t','plan','delivery','stop-a',incoming(product_name='新名称'),plan_product(brand_mode='required',brand_id='new',brand_name='新品牌'))
    assert result==dict(alias_name=None,brand_mode='required',brand_id='new',brand_name='新品牌')


def test_mobile_explicit_metadata_wins_and_web_ambiguous_mapping_rejects():
    m=service();old=[snapshot(),snapshot(id='m2',stop_product_id='old2')]
    meta=Row(alias_name='独立',brand_mode='any',brand_id='',brand_name='')
    result=m.resolve_stop_meta(old,set(),'t','plan','delivery','stop-a',incoming(mobile_product_meta=meta),plan_product())
    assert result['alias_name']=='独立'
    with pytest.raises(Exception) as error:m.resolve_stop_meta(old,set(),'t','plan','delivery','stop-a',incoming(),plan_product())
    assert error.value.status_code==422


def test_disabled_loading_never_uses_database():
    class NoDatabase:
        def query(self,*args):raise AssertionError('old Web must not query side table')
    assert service().load_stop_meta(NoDatabase(),'t','plan',False)==[]


@pytest.mark.parametrize("pending", [False, True])
def test_actual_replace_writes_metadata_for_both_sides_without_extra_plan_products(pending):
    import ast
    from typing import Optional
    from datetime import datetime
    path=Path(__file__).parents[1]/'app/services/pl_plan_stop_service.py'
    tree=ast.parse(path.read_text())
    fn=next(n for n in tree.body if isinstance(n,ast.FunctionDef) and n.name=='replace_plan_stops')
    class Field:
        def __eq__(self,value):return self
        def in_(self,value):return self
    class Delivery:
        plan_id=tenant_no=Field()
        def __init__(self,**kw):self.__dict__.update(kw)
    class DB:
        def __init__(self):self.added=[]
        def add(self,row):self.added.append(row)
        def query(self,*args):return self
        def filter(self,*args):return self
        def all(self):return []
    db=DB();calls=[];old=snapshot()
    scope=dict(Session=object,Optional=Optional,PlanLoadBody=Row,PlanDeliveryBody=Row,
      datetime=datetime,settings=Row(PL_MOBILE_PLAN_PRODUCT_META_ENABLED=True),
      load_stop_meta=lambda *args:[old],write_stop_meta=lambda *args:calls.append(args),
      _soft_delete_load_products_and_stop_contracts=lambda *a:None,
      _resolve_or_create_plan_load=lambda *a:'load',_resolve_plan_product_id=lambda p,rows,*args:p.plan_product_id or "",
      generate_uuid=lambda:'new-stop-product',PlPlanLoadProduct=Row,PlPlanDeliveryProduct=Row,PlPlanDelivery=Delivery,
      _write_stop_contracts=lambda *a:0,PLAN_CONTRACT_SIDE_LOAD='load',PLAN_CONTRACT_SIDE_DELIVERY='delivery',
      normalize_stop_dibang_fields=lambda *a:(0,'none'),resolve_delivery_id_for_upsert=lambda **kw:('stop-a',None),
      _soft_delete_delivery_products=lambda *a:None,is_existing_pending_delivery=lambda *a:False)
    exec(compile(ast.Module(body=[fn],type_ignores=[]),str(path),'exec'),scope)
    prod=incoming(plan_product_id=None if pending else 'pp',unlinked_draft=pending,product_no='no',delivery_method='',plan_qty=3,remark='')
    load=Row(products=[] if pending else [prod],contracts=[])
    delivery=Row(id=None,seq=1,customer_id='c',customer_name='C',address='',lng=None,lat=None,delivery_time=None,contact='',notes='',generate_dibang=0,dibang_direction='none',remark='',contracts=[],products=[prod])
    scope['replace_plan_stops'](db,'t','plan',load,[delivery],'u','d',[] if pending else [Row(id='pp')],allow_unlinked_draft=pending)
    assert len(calls)==(1 if pending else 2)
    assert [args[5] for args in calls]==(['delivery'] if pending else ['load','delivery'])
    if pending:
        assert calls[0][9] is None
        assert next(row for row in db.added if hasattr(row,'delivery_id')).plan_product_id == ''
    assert old.del_flag==1
    assert sum(row.plan_qty for row in db.added if hasattr(row,'load_id'))==(0 if pending else 3)
    assert sum(row.plan_qty for row in db.added if hasattr(row,'delivery_id'))==3

@pytest.mark.parametrize('route_name',['create_plan_full','update_plan'])
def test_actual_mobile_draft_route_branch_preserves_partial_stops_and_does_not_swallow_error(route_name):
    import ast
    from fastapi import HTTPException
    path=Path(__file__).parents[1]/'app/api/pl/plan.py'
    fn=next(n for n in ast.parse(path.read_text()).body if isinstance(n,ast.FunctionDef) and n.name==route_name)
    branch=next(n for n in fn.body if isinstance(n,ast.If) and ast.unparse(n.test)=='is_draft' and 'validate_plan_stops_write' in ast.unparse(n))
    load=Row(products=[incoming()],contracts=[])
    def strict(*args,**kwargs):raise HTTPException(422,'missing required fields')
    def run(light):
        state=dict(PlanContractLink=Row,is_draft=True,multi_stop=True,payload=Row(mobile_product_meta=True,belongs_company='x',create_date='date',load=load,deliveries=[],products=[]),db=object(),tenant_no='t',plan=Row(),id='p',validate_draft_min_fields=lambda **kw:None,validate_plan_stops_write=strict,validate_mobile_draft_stops=light,collect_stop_qty_warnings=lambda *a:[],sync_plan_legacy_from_stops=lambda *a:None,HTTPException=HTTPException,load_plan_contracts_grouped=lambda *a:{})
        exec(compile(ast.Module(body=[branch],type_ignores=[]),str(path),'exec'),state)
        return state
    result=run(lambda *args:(load,[]))
    assert result['load_body'] is load
    with pytest.raises(HTTPException):run(strict)


def test_mobile_draft_accepts_missing_fields_zero_and_validates_present_sources():
    import ast
    from typing import Optional
    from decimal import Decimal
    from fastapi import HTTPException
    path=Path(__file__).parents[1]/'app/services/pl_plan_stop_service.py'
    fn=next((n for n in ast.parse(path.read_text()).body if isinstance(n,ast.FunctionDef) and n.name=='validate_mobile_draft_stops'),None)
    assert fn is not None
    calls=[]
    scope=dict(Session=object,Optional=Optional,PlanLoadBody=Row,PlanDeliveryBody=Row,Decimal=Decimal,HTTPException=HTTPException,
       validate_mobile_stop_sources=lambda *args:calls.append(args))
    exec(compile(ast.Module(body=[fn],type_ignores=[]),str(path),'exec'),scope)
    load=Row(products=[Row(plan_qty=0)],customer_id='',customer_name='',contracts=[])
    assert scope['validate_mobile_draft_stops'](object(),'t',load,[],[Row(quantity=0)])==(load,[])
    assert len(calls)==1
    for qty in [-1,Decimal('NaN'),Decimal('Infinity')]:
        load.products[0].plan_qty=qty
        with pytest.raises(HTTPException):scope['validate_mobile_draft_stops'](object(),'t',load,[],[])


def test_opt_in_requires_complete_keys_and_nonnegative_finite_quantities():
    m=service()
    payload=Row(mobile_product_meta=True,products=[Row(client_key='a',quantity=0)],load=Row(products=[Row(plan_product_client_key='a',plan_qty=0,mobile_product_meta=Row())]),deliveries=[])
    m.validate_mobile_meta_request(payload,True)
    payload.load.products[0].plan_product_client_key=''
    with pytest.raises(Exception) as error:m.validate_mobile_meta_request(payload,True)
    assert error.value.status_code==422
    payload.load.products[0].plan_product_client_key='a'
    for qty in [-1,float('nan'),float('inf')]:
        payload.load.products[0].plan_qty=qty
        with pytest.raises(Exception) as error:m.validate_mobile_meta_request(payload,True)
        assert error.value.status_code==422


@pytest.mark.parametrize('alias', [None, '', '送货别名'])
def test_actual_snapshot_writer_preserves_same_packaging_contract_lines_across_web_rebuild(monkeypatch, alias):
    import sys
    import types
    from datetime import datetime
    model=types.ModuleType('app.models.pl_mobile_stop_meta');model.PlPlanStopProductMeta=Row
    utils=types.ModuleType('app.core.utils');sequence=iter(['new-meta-2','new-meta-1']);utils.generate_uuid=lambda:next(sequence)
    monkeypatch.setitem(sys.modules,'app.models.pl_mobile_stop_meta',model)
    monkeypatch.setitem(sys.modules,'app.core.utils',utils)
    added=[];db=Row(add=added.append);m=service();used=set()
    old=[snapshot(id='m1',contract_product_id='line1',brand_id='b1',alias_name=alias),snapshot(id='m2',contract_product_id='line2',brand_id='b2',alias_name=alias)]
    for line in ['line2','line1']:
        m.write_stop_meta(db,old,used,'t','plan','delivery','stop-a','new-'+line,incoming(contract_product_id=line),plan_product(),'u','d',datetime.now())
    assert [(row.stop_product_id,row.brand_id) for row in added]==[('new-line2','b2'),('new-line1','b1')]
    assert all(row.tenant_no=='t' and row.plan_id=='plan' for row in added)
    assert used=={'m1','m2'}
    assert all(row.alias_name == alias for row in added)


def test_sql_and_model_match_and_only_new_side_table_declares_metadata_fields():
    import ast,re
    root=Path(__file__).parents[2]
    path=root/'backend/app/models/pl_mobile_stop_meta.py'
    cls=next(n for n in ast.parse(path.read_text()).body if isinstance(n,ast.ClassDef))
    columns=[n.targets[0].id for n in cls.body if isinstance(n,ast.Assign) and isinstance(n.value,ast.Call) and isinstance(n.value.func,ast.Name) and n.value.func.id=='Column']
    sql=(root/'docs/sql/20261023_mobile_plan_stop_product_meta.sql').read_text()
    sql_columns=re.findall(r'^  (\w+) (?:VARCHAR|INT|DATETIME|SMALLINT)\b',sql,re.M)
    assert columns==sql_columns
    assert columns[-9:]==['tenant_no','order_num','remark','create_by','create_dept_id','create_time','update_by','update_time','del_flag']
    assert 'ALTER TABLE' not in sql
    alias_column = next(n.value for n in cls.body if isinstance(n,ast.Assign) and n.targets[0].id == 'alias_name')
    assert {kw.arg: ast.literal_eval(kw.value) for kw in alias_column.keywords} == {'nullable': True}
    assert "alias_name VARCHAR(200) DEFAULT NULL" in sql

@pytest.mark.parametrize('enabled',[False,True])
def test_actual_capability_route_requires_flag_and_readable_table(enabled):
    import ast
    from fastapi import HTTPException
    path=Path(__file__).parents[1]/'app/api/pl/plan.py'
    node=next(n for n in ast.parse(path.read_text()).body if isinstance(n,ast.FunctionDef) and n.name=='get_mobile_product_capability')
    node.decorator_list=[];node.returns=None
    for arg in node.args.args:arg.annotation=None
    node.args.defaults=[ast.Constant(None) for _ in node.args.defaults];ast.fix_missing_locations(node)
    reads=[];m=service()
    scope=dict(get_tenant_no=lambda *a:'tenant',settings=Row(PL_MOBILE_PLAN_PRODUCT_META_ENABLED=enabled),assert_mobile_meta_enabled=m.assert_mobile_meta_enabled,load_stop_meta=lambda *a:reads.append(a))
    exec(compile(ast.Module(body=[node],type_ignores=[]),str(path),'exec'),scope)
    if enabled:
        assert scope[node.name](object(),Row(to_tenant_user_dict=lambda:{}))=={'mobileProductMetaVersion':1}
        assert reads[0][1:]==('tenant','',True)
    else:
        with pytest.raises(HTTPException) as error:scope[node.name](object(),Row(to_tenant_user_dict=lambda:{}))
        assert error.value.status_code==503
        assert reads==[]


def test_actual_arrange_route_rejects_pending_stop_before_vehicle_checks_or_writes():
    import ast
    from fastapi import HTTPException
    path=Path(__file__).parents[1]/'app/api/pl/transport.py'
    node=next(n for n in ast.parse(path.read_text()).body if isinstance(n,ast.FunctionDef) and n.name=='arrange_transport')
    node.decorator_list=[];node.returns=None
    for arg in node.args.args:arg.annotation=None
    node.args.defaults=[ast.Constant(None) for _ in node.args.defaults];ast.fix_missing_locations(node)
    class Field:
        def __eq__(self,value):return self
    model=Row(id=Field(),tenant_no=Field(),del_flag=Field())
    class DB:
        def query(self,*a):return self
        def filter(self,*a):return self
        def first(self):return Row(transport_id=None)
        def add(self,*a):raise AssertionError('no write')
        def commit(self):raise AssertionError('no commit')
    def reject(*args):raise HTTPException(422,'请先补齐待关联产品')
    def vehicle(*args,**kwargs):raise AssertionError('must reject before vehicle arrangement')
    scope=dict(PlPlan=model,get_tenant_no=lambda *a:'tenant',assert_pl_row_in_scope=lambda *a:None,assert_no_pending_stop_products=reject,_is_rearrange=lambda *a:False,_assert_vehicle_carrier_eligible=vehicle,HTTPException=HTTPException)
    exec(compile(ast.Module(body=[node],type_ignores=[]),str(path),'exec'),scope)
    with pytest.raises(HTTPException) as error:scope[node.name]('p',Row(vehicle_id='v'),DB(),Row(to_tenant_user_dict=lambda:{}))
    assert error.value.status_code==422


@pytest.mark.parametrize('alias', [None, ''])
def test_web_unrelated_save_preserves_pending_alias_versus_explicit_mobile_clear(alias):
    result=service().resolve_stop_meta([snapshot(alias_name=alias)],set(),'t','plan','delivery','stop-a',incoming(),plan_product())
    assert result['alias_name'] is alias


def test_mobile_explicit_clear_is_not_pending_web_name_inference():
    meta=Row(alias_name='',brand_mode='any',brand_id='',brand_name='')
    result=service().resolve_stop_meta([snapshot()],set(),'t','plan','delivery','stop-a',incoming(product_name='新名称',mobile_product_meta=meta),plan_product())
    assert result['alias_name']==''


@pytest.mark.parametrize('route_name', ['create_plan_full', 'update_plan'])
@pytest.mark.parametrize('delivery_shape', ['empty', 'omitted', 'present'])
@pytest.mark.parametrize('with_products', [False, True])
def test_mobile_draft_route_uses_real_legacy_sync_with_optional_deliveries(route_name, delivery_shape, with_products):
    import ast
    from decimal import Decimal
    from typing import Optional
    from fastapi import HTTPException
    root=Path(__file__).parents[1]
    service_path=root/'app/services/pl_plan_stop_service.py'
    service_tree=ast.parse(service_path.read_text())
    functions=[n for n in service_tree.body if isinstance(n,ast.FunctionDef) and n.name in ('sync_plan_legacy_from_stops','validate_mobile_draft_stops')]
    source_checks=[]
    scope=dict(PlPlan=Row,PlanLoadBody=Row,PlanDeliveryBody=Row,Session=object,Optional=Optional,
        Decimal=Decimal,HTTPException=HTTPException,validate_mobile_stop_sources=lambda *a:source_checks.append(a))
    exec(compile(ast.Module(body=functions,type_ignores=[]),str(service_path),'exec'),scope)
    load_products=[Row(product_id='p',plan_qty=0)] if with_products else []
    plan_products=[Row(product_id='p',quantity=0)] if with_products else []
    load=Row(products=load_products,contracts=[],customer_id='supplier',customer_name='装货客户',address='装货地址',lng=1,lat=2,load_time='load-time',contact='装货联系',notes='装货备注')
    delivery=Row(products=[],contracts=[],customer_id='customer',customer_name='送货客户',address='送货地址',lng=3,lat=4,delivery_time='delivery-time',contact='送货联系',notes='送货备注')
    # Omitted Optional deliveries deserialize to None in the actual request schema.
    deliveries={'empty':[], 'omitted':None, 'present':[delivery]}[delivery_shape]
    customer_fields=('customer_id','customer_name','destination_id','destination_name','destination_address','destination_lng','destination_lat','delivery_time','delivery_contact','destination_notes')
    plan=Row(**{field:'stale-customer-value' for field in customer_fields},products=plan_products)
    payload=Row(mobile_product_meta=True,save_mode='draft',belongs_company='x',create_date=None,load=load,deliveries=deliveries,products=plan_products)
    path=root/'app/api/pl/plan.py'
    route=next(n for n in ast.parse(path.read_text()).body if isinstance(n,ast.FunctionDef) and n.name==route_name)
    branch=next(n for n in route.body if isinstance(n,ast.If) and ast.unparse(n.test)=='is_draft' and 'validate_plan_stops_write' in ast.unparse(n))
    statements=[branch]
    if route_name=='create_plan_full':
        # Creation syncs after constructing the plan, outside the draft validation branch.
        statements.append(next(n for n in route.body if isinstance(n,ast.If) and ast.unparse(n.test)=='multi_stop and load_body is not None' and 'sync_plan_legacy_from_stops' in ast.unparse(n)))
    def strict(*args,**kwargs):raise AssertionError('mobile incomplete draft must not use strict validation')
    scope.update(PlanContractLink=Row,is_draft=True,multi_stop=True,payload=payload,db=object(),tenant_no='t',plan=plan,id='plan',validate_draft_min_fields=lambda **kw:None,validate_plan_stops_write=strict,collect_stop_qty_warnings=lambda *a:[],load_plan_contracts_grouped=lambda *a:{})
    exec(compile(ast.Module(body=statements,type_ignores=[]),str(path),'exec'),scope)
    assert len(source_checks)==1
    assert scope['load_body'] is load
    assert scope['load_body'].products is load_products
    assert plan.products is plan_products
    assert plan.supplier_id=='supplier' and plan.pickup_address=='装货地址'
    assert plan.pickup_time=='load-time' and plan.pickup_contact=='装货联系'
    assert scope['delivery_bodies']==(deliveries or [])
    if delivery_shape=='present':
        assert [getattr(plan,f) for f in customer_fields]==['customer','送货客户','customer','送货客户','送货地址',3,4,'delivery-time','送货联系','送货备注']
    else:
        assert all(getattr(plan,f) is None for f in customer_fields)
        assert scope['delivery_bodies']==[]