299 lines
19 KiB
Python
299 lines
19 KiB
Python
"""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']==[]
|