158 lines
8.1 KiB
Python
158 lines
8.1 KiB
Python
from datetime import date
|
|
from decimal import Decimal
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
|
|
@pytest.fixture(params=[False, True])
|
|
def case(request):
|
|
from app.models import pl_models as m
|
|
from app.services import pl_quota_sources as svc
|
|
engine = create_engine('sqlite://')
|
|
for model in (m.PlPlan, m.PlPlanProduct, m.PlPlanLoad, m.PlPlanLoadProduct, m.PlPlanDelivery, m.PlPlanDeliveryProduct):
|
|
model.__table__.create(engine)
|
|
db = Session(engine, autoflush=request.param)
|
|
user = SimpleNamespace(id='u', tenant_no='t', to_user_info=lambda: {'role':'super_admin','user_id':'u'})
|
|
db.add(m.PlPlan(id='plan', plan_no='PLAN', create_date=date.today(), plan_type='direct', belongs_company='shanxia', tenant_no='t'))
|
|
db.add(m.PlPlanProduct(id='pp', plan_id='plan', product_id='sku', product_no='SKU', product_name='硫酸', is_toxic=1, tenant_no='t'))
|
|
db.add(m.PlPlanLoad(id='load', plan_id='plan', tenant_no='t'))
|
|
db.add(m.PlPlanDelivery(id='delivery', plan_id='plan', tenant_no='t'))
|
|
db.add(m.PlPlanLoadProduct(id='lp', load_id='load', plan_id='plan', plan_product_id='pp', product_id='sku', product_no='SKU', product_name='硫酸', plan_qty=30, contract_id='purchase', contract_product_id='cp', tenant_no='t'))
|
|
db.add(m.PlPlanDeliveryProduct(id='dp', delivery_id='delivery', plan_id='plan', plan_product_id='pp', product_id='sku', product_no='SKU', product_name='硫酸', plan_qty=30, contract_id='sale', contract_product_id='cs', tenant_no='t', outbound_qty=0))
|
|
db.commit()
|
|
yield SimpleNamespace(db=db, user=user, m=m, svc=svc)
|
|
db.close(); engine.dispose()
|
|
|
|
|
|
def mapping():
|
|
return [{'sourceStopType':'delivery', 'sourceStopProductId':'dp', 'relatedStopProductIds':['lp']}]
|
|
|
|
|
|
def test_two_sides_require_explicit_pairing_and_count_once(case):
|
|
c = case
|
|
with pytest.raises(HTTPException): c.svc.extract_sources(c.db,c.user,'plan')
|
|
sources, actual = c.svc.extract_sources(c.db,c.user,'plan',mapping(),actual=True)
|
|
assert len(sources)==1 and sources[0].quantity==Decimal('30')
|
|
assert sources[0].contract_id=='sale' and sources[0].related_stop_products==['lp']
|
|
assert actual=={'delivery:dp':Decimal('0')}
|
|
|
|
|
|
def test_missing_actual_not_replaced_by_plan_quantity(case):
|
|
c = case
|
|
c.db.get(c.m.PlPlanDeliveryProduct,'dp').outbound_qty=None
|
|
c.db.flush()
|
|
with pytest.raises(HTTPException) as error:
|
|
c.svc.extract_sources(c.db,c.user,'plan',mapping(),actual=True)
|
|
assert error.value.status_code==409
|
|
|
|
|
|
def test_invalid_or_duplicate_pairing_and_omitted_opposite_rejected(case):
|
|
c = case
|
|
for value in (mapping()*2,[{'sourceStopType':'delivery','sourceStopProductId':'dp','relatedStopProductIds':[]}],
|
|
[{'sourceStopType':'delivery','sourceStopProductId':'dp','relatedStopProductIds':['foreign']}],
|
|
mapping()+[{'sourceStopType':'load','sourceStopProductId':'lp','relatedStopProductIds':['dp']}]):
|
|
with pytest.raises(HTTPException): c.svc.extract_sources(c.db,c.user,'plan',value)
|
|
|
|
|
|
def test_single_side_can_be_derived_and_fingerprint_tracks_plan_facts(case):
|
|
c = case
|
|
c.db.get(c.m.PlPlanLoadProduct,'lp').del_flag=1; c.db.flush()
|
|
first,_=c.svc.extract_sources(c.db,c.user,'plan')
|
|
c.db.get(c.m.PlPlan,'plan').license_plate='皖A12345'; c.db.flush()
|
|
second,_=c.svc.extract_sources(c.db,c.user,'plan')
|
|
assert first[0].fingerprint!=second[0].fingerprint
|
|
c.db.get(c.m.PlPlanDeliveryProduct,'dp').outbound_qty=18; c.db.flush()
|
|
third,_=c.svc.extract_sources(c.db,c.user,'plan',actual=True)
|
|
assert second[0].fingerprint==third[0].fingerprint
|
|
|
|
|
|
def test_explosive_only_needs_no_quota_and_wrong_tenant_rejected(case):
|
|
c = case
|
|
row=c.db.get(c.m.PlPlanProduct,'pp'); row.is_toxic=0; row.is_explosive=1; c.db.flush()
|
|
assert c.svc.extract_sources(c.db,c.user,'plan')==([], {})
|
|
other=SimpleNamespace(**{**c.user.__dict__,'tenant_no':'other'})
|
|
with pytest.raises(HTTPException): c.svc.extract_sources(c.db,other,'plan')
|
|
|
|
|
|
def test_orphan_effective_stop_product_blocks_instead_of_silently_omitted(case):
|
|
c=case
|
|
c.db.add(c.m.PlPlanDeliveryProduct(id='orphan',delivery_id='deleted-stop',plan_id='plan',
|
|
plan_product_id='pp',product_id='sku',product_no='SKU',product_name='硫酸',plan_qty=70,
|
|
contract_id='sale',contract_product_id='cs',tenant_no='t'))
|
|
c.db.flush()
|
|
with pytest.raises(HTTPException) as error:
|
|
c.svc.extract_sources(c.db,c.user,'plan',mapping())
|
|
assert error.value.status_code==409
|
|
|
|
|
|
def test_extraction_does_not_flush_pending_changes(case):
|
|
from sqlalchemy import event
|
|
c=case
|
|
c.db.get(c.m.PlPlan,'plan').remark='未保存'
|
|
statements=[]
|
|
def capture(conn,cursor,statement,parameters,context,executemany):
|
|
statements.append(statement)
|
|
event.listen(c.db.bind,'before_cursor_execute',capture)
|
|
try:
|
|
c.svc.extract_sources(c.db,c.user,'plan',mapping())
|
|
finally:
|
|
event.remove(c.db.bind,'before_cursor_execute',capture)
|
|
assert not any(sql.lstrip().upper().startswith(('UPDATE','INSERT','DELETE')) for sql in statements)
|
|
|
|
|
|
def test_preview_vehicle_has_same_fingerprint_as_formal_facts_without_mutation(case):
|
|
c = case
|
|
plan = c.db.get(c.m.PlPlan, 'plan')
|
|
original = plan.license_plate
|
|
first, _ = c.svc.extract_sources(c.db, c.user, 'plan', mapping(),
|
|
preview_vehicle={'licensePlate': '皖A12345', 'trailerNo': '皖A挂'})
|
|
assert plan.license_plate == original and not c.db.dirty
|
|
plan.license_plate = '皖A12345'; plan.trailer_no = '皖A挂'; c.db.flush()
|
|
second, _ = c.svc.extract_sources(c.db, c.user, 'plan', mapping())
|
|
assert first[0].fingerprint == second[0].fingerprint
|
|
|
|
|
|
def test_planning_context_and_options_return_readonly_authoritative_facts(case):
|
|
from app.services import pl_quota_planning as planning
|
|
from app.models.pl_quota_models import PlPlanCertUsage, PlPlanCertUsageItem
|
|
c = case
|
|
for model in (PlPlanCertUsage, PlPlanCertUsageItem):
|
|
model.__table__.create(c.db.bind)
|
|
info = planning.get_plan_quota(c.db, c.user, 'plan')
|
|
assert len(info['sourceOptions']) == 2
|
|
context = planning.configuration_context(c.db, c.user, 'plan', {'sourceMapping': mapping()},
|
|
{'licensePlate': '皖A12345'})
|
|
assert context['sources'][0]['snapshot']['vehicle'] == '皖A12345'
|
|
assert len(context['sourceFingerprint']) == 64
|
|
assert not c.db.dirty and not c.db.new
|
|
|
|
|
|
def test_overview_filters_real_usage_states_ignores_old_ready_and_other_tenant(case):
|
|
from app.services.pl_quota_overview import list_plans
|
|
from app.models.pl_quota_models import PlPlanCertUsage
|
|
c = case
|
|
PlPlanCertUsage.__table__.create(c.db.bind)
|
|
c.db.get(c.m.PlPlan, 'plan').hazard_cert_ready = 1
|
|
c.db.flush()
|
|
assert list_plans(c.db, c.user)['list'][0]['quotaStatus'] == 'needs_configuration'
|
|
c.db.add(PlPlanCertUsage(id='usage',tenant_no='t',plan_id='plan',plan_no_snapshot='PLAN',cert_type='purchase',cert_id='g',cert_no_snapshot='G',cert_revision=1,quantity_unit='吨',usage_status='reserved'))
|
|
c.db.flush()
|
|
assert list_plans(c.db, c.user, status='reserved')['total'] == 1
|
|
assert list_plans(c.db, c.user, status='needs_configuration')['total'] == 0
|
|
c.db.get(PlPlanCertUsage, 'usage').usage_status = 'settled'; c.db.flush()
|
|
assert list_plans(c.db, c.user)['list'][0]['quotaStatus'] == 'settled'
|
|
c.db.get(PlPlanCertUsage, 'usage').confirmation_snapshot = {'settlementError': '实际纠错失败'}; c.db.flush()
|
|
assert list_plans(c.db, c.user, status='exception')['total'] == 1
|
|
c.db.get(PlPlanCertUsage, 'usage').confirmation_snapshot = {}; c.db.flush()
|
|
assert list_plans(c.db, c.user, status='settled')['total'] == 1
|
|
c.db.get(PlPlanCertUsage, 'usage').usage_status = 'exception'; c.db.flush()
|
|
assert list_plans(c.db, c.user)['list'][0]['quotaStatus'] == 'exception'
|
|
other=SimpleNamespace(**{**c.user.__dict__,'tenant_no':'other'})
|
|
assert list_plans(c.db, other)['total'] == 0
|
|
c.db.get(c.m.PlPlanProduct, 'pp').is_toxic = 0; c.db.flush()
|
|
assert list_plans(c.db, c.user)['list'][0]['quotaStatus'] == 'not_required'
|