wuzilongandCursor
85f5f29895
证书齐备豁免或已填写完整人工换算的计划不再标「需单位换算」，避免已归档豁免单长期挂在待办列表。 Co-authored-by: Cursor <cursoragent@cursor.com>
140 lines
7.0 KiB
Python
140 lines
7.0 KiB
Python
"""单位换算只在计量段已绑定异单位证件时成立。"""
|
|
from datetime import date, datetime
|
|
from types import SimpleNamespace
|
|
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.pl_models import (
|
|
PlPlan, PlPlanDelivery, PlPlanDeliveryProduct, PlPlanLoad, PlPlanLoadProduct, PlPlanProduct,
|
|
)
|
|
from app.models.pl_quota_models import (
|
|
PlContractCertLink, PlContractCertProduct, PlPlanCertUsage, PlPurchaseCert, PlTransportCert,
|
|
)
|
|
from app.services.pl_menu_badge_service import count_plan_cert_pending
|
|
from app.services.pl_quota_conversion import plan_needs_unit_conversion
|
|
from app.services.pl_quota_overview import list_plans
|
|
|
|
|
|
def line(line_id, side, unit, contract_product_id='cp', customer='客户', toxic=True):
|
|
return {
|
|
'id': line_id, 'planProductId': 'pp', 'side': side, 'unit': unit,
|
|
'contractProductId': contract_product_id, 'customerName': customer, 'isToxic': toxic,
|
|
}
|
|
|
|
|
|
def bind(unit, kind='purchase', cert_id='buy'):
|
|
return {'kind': kind, 'certId': cert_id, 'unit': unit}
|
|
|
|
|
|
def plan(**values):
|
|
return SimpleNamespace(
|
|
cert_configuration=values.get('configuration'),
|
|
transport_status=values.get('transport_status', 'pending'),
|
|
archive_status=values.get('archive_status', 'unarchived'),
|
|
quota_cert_waived=values.get('quota_cert_waived', 0),
|
|
)
|
|
|
|
|
|
def test_same_label_and_code_do_not_need_conversion():
|
|
current = plan()
|
|
assert plan_needs_unit_conversion(current, [line('L', 'load', '吨')], {'cp': [bind('dun')]}, []) is False
|
|
assert plan_needs_unit_conversion(current, [line('L', 'load', '吨', toxic=False)], {'cp': [bind('kg')]}, []) is False
|
|
assert plan_needs_unit_conversion(
|
|
current, [line('L', 'load', '吨', customer='安庆市山峡化工有限公司')], {'cp': [bind('ping')]}, [],
|
|
) is True
|
|
assert plan_needs_unit_conversion(current, [line('L', 'load', '吨')], {'cp': [bind('kg')]}, []) is False
|
|
assert plan_needs_unit_conversion(current, [line('L', 'load', '升')], {'cp': [bind('毫升')]}, []) is False
|
|
|
|
|
|
def test_unbound_and_unselected_legacy_sides_are_display_only():
|
|
both = [line('L', 'load', '吨'), line('D', 'delivery', '吨')]
|
|
arranged = plan(transport_status='arranged')
|
|
assert plan_needs_unit_conversion(arranged, both, {'cp': [bind('kg')]}, []) is False
|
|
chosen = plan(transport_status='arranged', configuration={'sourceMapping': [
|
|
{'sourceStopType': 'load', 'sourceStopProductId': 'L', 'relatedStopProductIds': ['D']},
|
|
]})
|
|
assert plan_needs_unit_conversion(chosen, both, {'cp': [bind('瓶')]}, []) is True
|
|
assert plan_needs_unit_conversion(chosen, both, {'cp': [bind('kg')]}, []) is False
|
|
|
|
|
|
def test_new_plan_counts_either_stop_when_a_bound_certificate_uses_another_unit():
|
|
current = plan()
|
|
assert plan_needs_unit_conversion(
|
|
current, [line('L', 'load', 'dun'), line('D', 'delivery', 'dun')], {'cp': [bind('kg')]}, [],
|
|
) is False
|
|
assert plan_needs_unit_conversion(
|
|
current, [line('L', 'load', 'dun'), line('D', 'delivery', 'dun')], {'cp': [bind('ping')]}, [],
|
|
) is True
|
|
assert plan_needs_unit_conversion(current, [line('L', 'load', 'dun')], {}, []) is False
|
|
|
|
|
|
def test_waived_or_saved_manual_conversion_clears_pending_flag():
|
|
current = plan(quota_cert_waived=1)
|
|
assert plan_needs_unit_conversion(
|
|
current, [line('L', 'load', 'ping')], {'cp': [bind('sheng', cert_id='buy')]}, [],
|
|
) is False
|
|
saved = plan(configuration={
|
|
'manualAllocations': [{
|
|
'permitKind': 'purchase', 'sourceId': 'delivery:D', 'permitId': 'buy',
|
|
'sourceQuantity': '10', 'permitQuantity': '2.5', 'note': '1瓶=0.25升',
|
|
}],
|
|
})
|
|
assert plan_needs_unit_conversion(
|
|
saved, [line('D', 'delivery', 'ping')], {'cp': [bind('sheng', cert_id='buy')]}, [],
|
|
) is False
|
|
incomplete = plan(configuration={
|
|
'manualAllocations': [{
|
|
'permitKind': 'purchase', 'sourceId': 'delivery:D', 'permitId': 'buy',
|
|
'sourceQuantity': '10', 'permitQuantity': '', 'note': '',
|
|
}],
|
|
})
|
|
assert plan_needs_unit_conversion(
|
|
incomplete, [line('D', 'delivery', 'ping')], {'cp': [bind('sheng', cert_id='buy')]}, [],
|
|
) is True
|
|
|
|
|
|
def test_list_and_badge_count_only_plans_with_a_real_unit_mismatch(monkeypatch):
|
|
engine = create_engine('sqlite://')
|
|
models = (
|
|
PlPlan, PlPlanProduct, PlPlanLoad, PlPlanDelivery, PlPlanLoadProduct, PlPlanDeliveryProduct,
|
|
PlContractCertLink, PlContractCertProduct, PlPurchaseCert, PlTransportCert, PlPlanCertUsage,
|
|
)
|
|
for model in models:
|
|
model.__table__.create(engine)
|
|
with Session(engine) as db:
|
|
user = SimpleNamespace(id='u', tenant_no='t', to_user_info=lambda: {'role': 'super_admin', 'user_id': 'u'})
|
|
db.add(PlPlan(id='plain', plan_no='PLAIN', create_date=date(2026, 11, 1), plan_date=date(2026, 11, 1),
|
|
plan_type='direct', belongs_company='shanxia', tenant_no='t',
|
|
create_time=datetime(2026, 11, 1)))
|
|
db.add(PlPlan(id='mix', plan_no='MIX', create_date=date(2026, 11, 2), plan_date=date(2026, 11, 2),
|
|
plan_type='direct', belongs_company='shanxia', tenant_no='t',
|
|
create_time=datetime(2026, 11, 2)))
|
|
db.add(PlPlanProduct(id='pp-plain', plan_id='plain', product_id='sku', product_no='SKU',
|
|
product_name='硫酸', is_toxic=1, tenant_no='t'))
|
|
db.add(PlPlanProduct(id='pp-mix', plan_id='mix', product_id='sku', product_no='SKU',
|
|
product_name='硫酸', is_toxic=1, tenant_no='t'))
|
|
db.add(PlPlanLoad(id='load-mix', plan_id='mix', customer_name='外部客户', tenant_no='t'))
|
|
db.add(PlPlanLoadProduct(
|
|
id='line-mix', load_id='load-mix', plan_id='mix', plan_product_id='pp-mix', product_id='sku',
|
|
product_no='SKU', product_name='硫酸', unit='吨', contract_product_id='cp', tenant_no='t',
|
|
))
|
|
db.add(PlPurchaseCert(
|
|
id='buy', cert_no='BUY', product_name_on_cert='硫酸', total_qty=1, quantity_unit='ping',
|
|
valid_use_text='多次有效', buyer_name='买方', seller_name='卖方', tenant_no='t',
|
|
))
|
|
db.add(PlContractCertLink(
|
|
id='link', contract_id='c', contract_no_snapshot='C-1', cert_type='purchase', cert_id='buy',
|
|
tenant_no='t',
|
|
))
|
|
db.add(PlContractCertProduct(id='scope', contract_cert_link_id='link', contract_product_id='cp', tenant_no='t'))
|
|
db.commit()
|
|
monkeypatch.setattr('app.services.pl_menu_badge_service.apply_pl_scope', lambda query, *args, **kwargs: query)
|
|
monkeypatch.setattr('app.services.pl_quota_overview.apply_pl_scope', lambda query, *args, **kwargs: query)
|
|
listed = {row['planNo']: row['needsUnitConversion'] for row in list_plans(db, user)['list']}
|
|
assert listed == {'MIX': True, 'PLAIN': False}
|
|
only = list_plans(db, user, conversion_only=True)
|
|
assert only['total'] == 1 and only['list'][0]['planNo'] == 'MIX'
|
|
assert count_plan_cert_pending(db, 't', user) == 1
|
|
engine.dispose()
|