shanxia2.0/backend/tests/test_pl_mobile_product_identity.py

"""Pure request identity tests: no database, application startup, network or browser."""
from pathlib import Path
from types import SimpleNamespace as Row
import importlib.util
import pytest

PATH = Path(__file__).parents[1] / 'app/services/pl_mobile_product_identity.py'

def service():
    assert PATH.exists(), 'mobile request identity service exists'
    spec = importlib.util.spec_from_file_location('mobile_identity_test_module', PATH)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module

def test_request_keys_bind_new_same_product_rows_independently():
    m = service()
    products = [Row(client_key='a', id=None, product_id='p'), Row(client_key='b', id=None, product_id='p')]
    stops = [Row(plan_product_client_key='a', plan_product_id=None, product_id='p'), Row(plan_product_client_key='b', plan_product_id=None, product_id='p')]
    m.validate_mobile_product_links(products, stops)
    rows = [Row(id='uuid-a', product_id='p'), Row(id='uuid-b', product_id='p')]
    for row, product in zip(rows, products): m.bind_mobile_product_key(row, product)
    assert [m.resolve_mobile_product_link(stop, rows) for stop in stops] == ['uuid-a','uuid-b']

def test_legacy_no_key_path_is_untouched():
    m = service()
    m.validate_mobile_product_links([Row(id='duplicated'), Row(id='duplicated')], [Row()])
    assert m.resolve_mobile_product_link(Row(plan_product_id='legacy'), []) is None

@pytest.mark.parametrize('products,stops', [
    ([Row(client_key='a',id=None,product_id='p'),Row(client_key='a',id=None,product_id='p')],[]),
    ([Row(client_key='a',id='old',product_id='p'),Row(client_key='b',id='old',product_id='p')],[]),
    ([Row(client_key='a',id=None,product_id='p')],[Row(plan_product_client_key='unknown')]),
    ([Row(client_key='a',id='old',product_id='p')],[Row(plan_product_client_key='a',plan_product_id='other',product_id='p')]),
    ([Row(client_key='a',id='old',product_id='p')],[Row(plan_product_client_key='a',plan_product_id='old',product_id='other')])
])
def test_invalid_request_links_reject_before_writing(products, stops):
    m = service()
    with pytest.raises(Exception) as error: m.validate_mobile_product_links(products,stops)
    assert error.value.status_code == 422

def test_runtime_explicit_and_key_conflict_is_rejected():
    m = service()
    row=Row(id='a',product_id='p');m.bind_mobile_product_key(row,Row(client_key='key'))
    with pytest.raises(Exception) as error:
        m.resolve_mobile_product_link(Row(plan_product_client_key='key',plan_product_id='b',product_id='p'),[row])
    assert error.value.status_code == 422


def production_function(path, names, bindings):
    import ast
    source = ast.parse((Path(__file__).parents[1] / path).read_text())
    nodes = [node for node in source.body if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) and node.name in names]
    scope = {'__builtins__': __builtins__, **bindings}
    exec(compile(ast.Module(body=nodes, type_ignores=[]), path, 'exec'), scope)
    return scope


def test_actual_build_apply_and_stop_resolver_keep_new_and_existing_keys():
    from typing import Optional
    from decimal import Decimal
    m = service()
    common = dict(Optional=Optional, PlanProductCreate=Row, PlPlanProduct=Row,
                  Decimal=Decimal, generate_uuid=lambda:'server-generated',
                  bind_mobile_product_key=m.bind_mobile_product_key)
    scope = production_function('app/services/pl_plan_service.py', ['_product_quantity','_estimate_qty_fields','build_plan_product','apply_plan_product_fields'], common)
    fields = dict(product_id='p',product_no='p',product_name='alias-a',brand_mode='required',brand_id='brand-a',brand_name='A',quantity=Decimal('1.125'))
    for name in ['inbound_estimate_qty','outbound_estimate_qty','tax_price','model','unit','category_name','package_spec','delivery_method','actual_qty','tax_amount','tax_subtotal','is_explosive','is_toxic','purchase_cert_url','transport_cert_url','legal_use_cert_url','remark']:
        fields.setdefault(name,None)
    prod = Row(**fields,client_key='new')
    row = scope['build_plan_product'](tenant_no='t',plan_id='plan',plan_type='outbound',prod=prod,order_num=0,user_id='u',dept_id='d')
    assert row.id == 'server-generated'
    assert row._mobile_request_client_key == 'new'
    assert (row.quantity,row.brand_id,row.product_name) == (Decimal('1.125'),'brand-a','alias-a')
    prod.client_key='existing'
    scope['apply_plan_product_fields'](row,plan_type='outbound',prod=prod,order_num=0,user_id='u')
    assert row._mobile_request_client_key == 'existing'
    from fastapi import HTTPException
    stop_scope = production_function('app/services/pl_plan_stop_service.py', ['_resolve_plan_product_id'], dict(PlanStopProductBody=Row, resolve_mobile_product_link=m.resolve_mobile_product_link, HTTPException=HTTPException))
    wrong = Row(id='first-same-product',product_id='p',product_no='p')
    stop=Row(plan_product_client_key='existing',plan_product_id=None,product_id='p',product_no='p',unit='吨')
    assert stop_scope['_resolve_plan_product_id'](stop,[wrong,row]) == row.id

@pytest.mark.parametrize('field', ['package_spec','model','unit'])
def test_mobile_request_rejects_incompatible_package_model_unit(field):
    product=Row(client_key='a',id='pp',product_id='p',package_spec='桶',model='M',unit='吨')
    stop=Row(plan_product_client_key='a',plan_product_id='pp',product_id='p',package_spec='桶',model='M',unit='吨')
    setattr(stop,field,'different')
    with pytest.raises(Exception) as error:service().validate_mobile_product_links([product],[stop])
    assert error.value.status_code==422


def test_new_mobile_plan_cannot_claim_existing_product_id():
    with pytest.raises(Exception) as error:
        service().validate_mobile_product_links([Row(client_key='a',id='fake')],[],creating=True)
    assert error.value.status_code==422

@pytest.mark.parametrize('route_name',['create_plan_full','update_plan'])
def test_actual_route_rejects_unknown_keys_before_any_write(route_name):
    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==route_name)
    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 NoWrite:
        def add(self,*a):raise AssertionError('must validate before write')
        def commit(self):raise AssertionError('must never commit invalid request')
    payload=Row(mobile_product_meta=False,products=[Row(client_key='known',id=None,product_id='p')],load=Row(products=[Row(plan_product_client_key='unknown',product_id='p')]),deliveries=[])
    scope=dict(get_tenant_no=lambda *a:'t',_get_plan=lambda *a:Row(follow_status='draft'),settings=Row(PL_MOBILE_PLAN_PRODUCT_META_ENABLED=False),validate_mobile_meta_request=lambda *a:None,validate_mobile_product_links=service().validate_mobile_product_links,HTTPException=HTTPException)
    exec(compile(ast.Module(body=[node],type_ignores=[]),str(path),'exec'),scope)
    args=dict(payload=payload,db=NoWrite(),user=Row(to_tenant_user_dict=lambda:{}))
    if route_name=='update_plan':args['id']='plan'
    with pytest.raises(HTTPException) as error:scope[route_name](**args)
    assert error.value.status_code==422


def test_real_schema_accepts_aliases_but_keys_are_not_orm_fields():
    import ast
    from typing import Optional
    from decimal import Decimal
    from pydantic import BaseModel,Field
    path=Path(__file__).parents[1]/'app/schemas/pl/plan.py'
    names={'PlanProductBase','PlanProductCreate','PlanStopProductMeta','PlanStopProductBody'}
    nodes=[n for n in ast.parse(path.read_text()).body if isinstance(n,ast.ClassDef) and n.name in names]
    scope=dict(BaseModel=BaseModel,Field=Field,Optional=Optional,Decimal=Decimal)
    exec(compile(ast.Module(body=nodes,type_ignores=[]),str(path),'exec'),scope)
    product=scope['PlanProductCreate'](clientKey='client',productId='p',productNo='P',productName='alias')
    stop=scope['PlanStopProductBody'](planProductClientKey='client',productId='p',productNo='P',productName='alias',mobileProductMeta={'aliasName':'alias','brandMode':'required','brandId':'brand'})
    assert product.client_key==stop.plan_product_client_key=='client'
    assert stop.mobile_product_meta.brand_id=='brand'
    model=(Path(__file__).parents[1]/'app/models/pl_models.py').read_text()
    assert 'client_key' not in model and 'mobile_product_meta' not in model


def test_unlinked_draft_is_explicit_and_cannot_bypass_formal_or_claim_an_id():
    m=service();stop=Row(unlinked_draft=True,plan_product_id=None,plan_product_client_key=None,product_id='p')
    m.validate_mobile_product_links([], [stop], allow_unlinked_draft=True)
    with pytest.raises(Exception) as error:m.validate_mobile_product_links([], [stop])
    assert error.value.status_code==422
    stop.plan_product_id='fake'
    with pytest.raises(Exception):m.validate_mobile_product_links([], [stop],allow_unlinked_draft=True)


def test_actual_stop_resolver_persists_empty_pending_reference_only_when_explicitly_allowed():
    from fastapi import HTTPException
    scope=production_function('app/services/pl_plan_stop_service.py',['_resolve_plan_product_id'],dict(PlanStopProductBody=Row,resolve_mobile_product_link=service().resolve_mobile_product_link,HTTPException=HTTPException))
    stop=Row(unlinked_draft=True,plan_product_id=None,plan_product_client_key=None)
    assert scope['_resolve_plan_product_id'](stop,[],True)==''
    with pytest.raises(HTTPException):scope['_resolve_plan_product_id'](stop,[])


def test_default_stop_resolver_keeps_legacy_web_first_product_mapping():
    from fastapi import HTTPException
    scope=production_function('app/services/pl_plan_stop_service.py',['_resolve_plan_product_id'],dict(PlanStopProductBody=Row,resolve_mobile_product_link=service().resolve_mobile_product_link,HTTPException=HTTPException))
    rows=[Row(id='a',product_id='p'),Row(id='b',product_id='p')]
    assert scope['_resolve_plan_product_id'](Row(plan_product_id=None,product_id='p'),rows)=='a'


def test_existing_pending_web_draft_matches_only_its_original_segment_and_contract_line():
    m=service()
    assert hasattr(m,'is_existing_pending_delivery')
    old=Row(delivery_id='A',contract_id='c',contract_product_id='line',product_id='p',package_spec='bag',model='M',unit='t')
    product=Row(plan_product_id=None,plan_product_client_key=None,contract_id='c',contract_product_id='line',product_id='p',package_spec='bag',model='M',unit='t')
    assert m.is_existing_pending_delivery(product,'A',[old])
    assert not m.is_existing_pending_delivery(product,'B',[old])
    product.contract_product_id='replacement'
    assert not m.is_existing_pending_delivery(product,'A',[old])