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_mobile_save_does_not_require_side_table_flag():
    m=service()
    assert m.assert_mobile_meta_enabled(False, True) is None
    plan=Row(client_key='a', id='pp', product_name='标准名', brand_mode=None, brand_id=None, brand_name=None)
    meta=Row(alias_name='工业别名', brand_mode='required', brand_id='b1', brand_name='品牌一')
    stop=Row(plan_product_client_key='a', plan_product_id='pp', product_name='标准名', mobile_product_meta=meta)
    payload=Row(mobile_product_meta=True, products=[plan], load=Row(products=[stop]), deliveries=[])
    m.fold_mobile_selection_into_web_fields(payload)
    assert stop.product_name=='工业别名'
    assert plan.product_name=='工业别名'
    assert plan.brand_mode=='required' and plan.brand_id=='b1' and plan.brand_name=='品牌一'


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):
    # Exercise real ORM boundaries instead of extracting a function with stale collaborator stubs.
    from sqlalchemy import create_engine
    from sqlalchemy.orm import Session
    from app.models import pl_models as models
    from app.models.pl_mobile_stop_meta import PlPlanStopProductMeta
    from app.schemas.pl.plan import PlanLoadBody, PlanDeliveryBody, PlanStopProductBody
    from app.services.pl_plan_stop_service import replace_plan_stops

    engine = create_engine('sqlite://')
    for table in models.PlPlan.metadata.sorted_tables:
        if table.name.startswith('pl_'):
            table.create(engine, checkfirst=True)
    with Session(engine, autoflush=False) as db:
        pp = models.PlPlanProduct(id='pp', tenant_no='t', plan_id='plan', product_id='p',
                                  product_no='no', product_name='产品')
        if not pending:
            db.add(pp)
            db.flush()
        product = PlanStopProductBody(productId='p', productNo='no', productName='产品',
            planProductId=None if pending else 'pp', unlinkedDraft=pending, planQty=3)
        load = PlanLoadBody(products=[] if pending else [product], contracts=[])
        delivery = PlanDeliveryBody(products=[product], contracts=[])
        replace_plan_stops(db, 't', 'plan', load, [delivery], 'u', 'd',
                           [] if pending else [pp], allow_unlinked_draft=pending)
        db.flush()
        assert db.query(PlPlanStopProductMeta).count() == 0
        assert db.query(models.PlPlanProduct).count() == (0 if pending else 1)
        assert sum(row.plan_qty for row in db.query(models.PlPlanLoadProduct).all()) == (0 if pending else 3)
        rows = db.query(models.PlPlanDeliveryProduct).all()
        assert sum(row.plan_qty for row in rows) == 3
        if pending:
            assert rows[0].plan_product_id == ''
    engine.dispose()

@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:{})
        state.update(validation_load=load, validation_deliveries=[])
        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

def test_actual_capability_route_does_not_require_side_table():
    import ast
    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)
    scope={}
    exec(compile(ast.Module(body=[node],type_ignores=[]),str(path),'exec'),scope)
    assert scope[node.name](Row())=={'mobileProductMetaVersion':1}


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:{})
    scope.update(validation_load=load, validation_deliveries=deliveries)
    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']==[]