187 lines
10 KiB
Python
187 lines
10 KiB
Python
from datetime import date
|
|
from decimal import Decimal as D
|
|
import pytest
|
|
from app.services.pl_quota_allocation import (SourceDemand, PermitCandidate, ManualAllocation, AllocationLimits, AllocationInputError, AllocationIncomplete, InsufficientQuota, SupplementalAllocationRequired, allocate_plan, settle_allocations)
|
|
|
|
def source(id='s', product='A', quantity='30', unit='t'):
|
|
return SourceDemand(id, product, D(quantity), unit)
|
|
|
|
def permit(id, quantity='30', products=('A',), refs=(), unit='t', expiry=1):
|
|
return PermitCandidate(id, D(quantity), unit, date(2027, 1, expiry), frozenset(products), frozenset(refs))
|
|
|
|
def quantities(rows):
|
|
return [(r.permit_id, r.source_quantity, r.permit_quantity) for r in rows]
|
|
|
|
def test_two_certificates_each_cover_20_plus_10():
|
|
result = allocate_plan([source()], [permit('g1', '20'), permit('g2', '10')], [permit('t1', '20', refs=('g1',)), permit('t2', '10', refs=('g2',))])
|
|
assert quantities(result.purchase_allocations) == [('g2', D(10), D(10)), ('g1', D(20), D(20))]
|
|
assert sum(r.source_quantity for r in result.transport_allocations) == D(30)
|
|
assert [r.allocation_seq for r in result.purchase_allocations] == [1, 2]
|
|
|
|
def test_global_scarce_product_and_shared_capacity():
|
|
result = allocate_plan([source('a', 'A', '10'), source('b', 'B', '10')], [permit('g1', '10', ('A', 'B')), permit('g2', '10')], [permit('t', '20', ('A', 'B'), ('g1', 'g2'))])
|
|
assert {(r.source_id, r.permit_id) for r in result.purchase_allocations} == {('a', 'g2'), ('b', 'g1')}
|
|
|
|
def test_transport_reference_forces_purchase_reallocation():
|
|
result = allocate_plan([source(quantity='10')], [permit('g1', '10'), permit('g2', '10')], [permit('t', '10', refs=('g2',))])
|
|
assert result.purchase_allocations[0].permit_id == 'g2'
|
|
|
|
def test_manual_different_units_and_auto_rejected():
|
|
args = ([source(quantity='10')], [permit('g', '100', unit='kg')], [permit('t', '10', refs=('g',))])
|
|
with pytest.raises(InsufficientQuota):
|
|
allocate_plan(*args)
|
|
result = allocate_plan(*args, manual_allocations=[ManualAllocation('purchase', 's', 'g', D(10), D(80))])
|
|
assert quantities(result.purchase_allocations) == [('g', D(10), D(80))]
|
|
with pytest.raises(AllocationInputError):
|
|
settle_allocations(result.purchase_allocations, {'s': D(9)})
|
|
settled = settle_allocations(result.purchase_allocations, {'s': D(9)}, manual_actuals=[ManualAllocation('purchase', 's', 'g', D(9), D(70))])
|
|
assert settled[0].actual_permit_quantity == D(70)
|
|
assert settled[0].release_permit_quantity == D(80)
|
|
|
|
def test_insufficient_shared_total():
|
|
with pytest.raises(InsufficientQuota):
|
|
allocate_plan([source('a', quantity='20'), source('b', quantity='20')], [permit('g', '30')], [permit('t', '40', refs=('g',))])
|
|
|
|
@pytest.mark.parametrize('value', ['NaN', 'Infinity', '-1', '0.0000001'])
|
|
def test_invalid_decimal(value):
|
|
with pytest.raises(AllocationInputError):
|
|
allocate_plan([source(quantity=value)], [], [])
|
|
|
|
@pytest.mark.parametrize('actual, expected', [('29', [D(20), D(9)]), ('18', [D(18), D(0)])])
|
|
def test_settlement_preserves_sequence_and_releases_all(actual, expected):
|
|
plan = allocate_plan([source()], [permit('g1', '20', expiry=1), permit('g2', '10', expiry=2)], [permit('t', '30', refs=('g1', 'g2'))])
|
|
rows = settle_allocations(plan.purchase_allocations, {'s': D(actual)})
|
|
assert [r.actual_permit_quantity for r in rows] == expected
|
|
assert [r.release_permit_quantity for r in rows] == [D(20), D(10)]
|
|
with pytest.raises(SupplementalAllocationRequired):
|
|
settle_allocations(plan.purchase_allocations, {'s': D(31)})
|
|
|
|
def test_stable_expiry_capacity_id_order_and_budget():
|
|
demands = [source(quantity='5')]
|
|
purchases = [permit('z', '10'), permit('b', '5'), permit('a', '5')]
|
|
transports = [permit('t', '20', refs=('a', 'b', 'z'))]
|
|
assert allocate_plan(demands, purchases, transports) == allocate_plan(demands, purchases[::-1], transports)
|
|
assert allocate_plan(demands, purchases, transports).purchase_allocations[0].permit_id == 'a'
|
|
with pytest.raises(AllocationIncomplete):
|
|
allocate_plan(demands, purchases, transports, limits=AllocationLimits(max_pivots=0))
|
|
|
|
@pytest.mark.parametrize('limits', [AllocationLimits(max_seconds=float('nan')), AllocationLimits(max_variables=-1), AllocationLimits(max_pivots=1.5)])
|
|
def test_invalid_limits_are_input_errors(limits):
|
|
with pytest.raises(AllocationInputError):
|
|
allocate_plan([], [], [], limits=limits)
|
|
|
|
|
|
def test_decimal_context_cannot_round_allocation():
|
|
from decimal import localcontext
|
|
with localcontext() as ctx:
|
|
ctx.prec = 2
|
|
result = allocate_plan([source(quantity='123.456789')], [permit('g', '123.456789')], [permit('t', '123.456789', refs=('g',))])
|
|
assert result.purchase_allocations[0].permit_quantity == D('123.456789')
|
|
|
|
|
|
def test_manual_quantity_still_consumes_shared_certificate_capacity():
|
|
with pytest.raises(InsufficientQuota):
|
|
allocate_plan([source('a', quantity='10'), source('b', quantity='10')], [permit('g', '100', unit='kg')], [permit('t', '20', refs=('g',))], manual_allocations=[ManualAllocation('purchase', sid, 'g', D(10), D(60)) for sid in ('a', 'b')])
|
|
|
|
|
|
def test_manual_transport_multiple_references_matches_source_amount():
|
|
plan = allocate_plan([source(quantity='30')], [permit('g1', '20'), permit('g2', '10')], [permit('t', '100', refs=('g1', 'g2'), unit='kg')], manual_allocations=[ManualAllocation('transport', 's', 't', D(30), D(90))])
|
|
assert sum(r.source_quantity for r in plan.purchase_allocations) == D(30)
|
|
assert quantities(plan.transport_allocations) == [('t', D(30), D(90))]
|
|
|
|
|
|
@pytest.mark.parametrize('rows', [
|
|
[ManualAllocation('purchase', 's', 'g', D(10), D(9))],
|
|
[ManualAllocation('purchase', 'missing', 'g', D(10), D(10))],
|
|
[ManualAllocation('purchase', 's', 'g', D(10), D(10))] * 2,
|
|
])
|
|
def test_invalid_manual_rows(rows):
|
|
with pytest.raises(AllocationInputError):
|
|
allocate_plan([source(quantity='10')], [permit('g', '10')], [permit('t', '10', refs=('g',))], manual_allocations=rows)
|
|
|
|
|
|
def test_input_scale_limit_is_incomplete_not_insufficient():
|
|
with pytest.raises(AllocationIncomplete):
|
|
allocate_plan([source('a'), source('b')], [], [], limits=AllocationLimits(max_sources=1))
|
|
|
|
|
|
def test_expiry_precedes_small_remaining_and_permit_id():
|
|
plan = allocate_plan([source(quantity='5')], [permit('a', '5', expiry=2), permit('z', '10', expiry=1)], [permit('t', '10', refs=('a', 'z'))])
|
|
assert plan.purchase_allocations[0].permit_id == 'z'
|
|
|
|
|
|
def test_zero_demand_and_empty_plan():
|
|
assert allocate_plan([], [], []).purchase_allocations == ()
|
|
assert allocate_plan([source(quantity='0')], [], []).transport_allocations == ()
|
|
|
|
|
|
def test_one_purchase_and_two_transport_certificates():
|
|
result = allocate_plan(
|
|
[source()], [permit('g', '30')],
|
|
[permit('t1', '20', refs=('g',)), permit('t2', '10', refs=('g',))],
|
|
)
|
|
assert len(result.purchase_allocations) == 1
|
|
assert len(result.transport_allocations) == 2
|
|
assert sum(r.permit_quantity for r in result.transport_allocations) == D(30)
|
|
|
|
|
|
def test_transport_reference_does_not_borrow_purchase_from_other_source():
|
|
with pytest.raises(InsufficientQuota):
|
|
allocate_plan(
|
|
[source('a', 'A', '10'), source('b', 'B', '10')],
|
|
[permit('g1', '10', ('B',)), permit('g2', '10', ('A',))],
|
|
[permit('t1', '10', ('A',), ('g1',)), permit('t2', '10', ('B',), ('g2',))],
|
|
)
|
|
|
|
|
|
def test_fractional_quantities_are_exact():
|
|
result = allocate_plan(
|
|
[source(quantity='0.300001')],
|
|
[permit('g1', '0.1'), permit('g2', '0.200001')],
|
|
[permit('t', '0.300001', refs=('g1', 'g2'))],
|
|
)
|
|
assert quantities(result.purchase_allocations) == [
|
|
('g1', D('0.1'), D('0.1')), ('g2', D('0.200001'), D('0.200001')),
|
|
]
|
|
|
|
|
|
def test_settlement_pairing_failure_requires_manual_adjustment():
|
|
from app.services.pl_quota_allocation import AllocationPairingError, validate_allocation_pairing
|
|
from dataclasses import replace
|
|
transports = [permit('y1', '10', refs=('g2',), expiry=1), permit('y2', '20', refs=('g1',), expiry=2)]
|
|
plan = allocate_plan([source()], [permit('g1', '20', expiry=1), permit('g2', '10', expiry=2)], transports)
|
|
def actual_rows(rows, actual):
|
|
return [replace(r.allocation, source_quantity=r.actual_source_quantity, permit_quantity=r.actual_permit_quantity) for r in settle_allocations(rows, {'s': D(actual)})]
|
|
purchases = actual_rows(plan.purchase_allocations, '18')
|
|
transport_rows = actual_rows(plan.transport_allocations, '18')
|
|
with pytest.raises(AllocationPairingError, match='人工调整'):
|
|
validate_allocation_pairing([source(quantity='18')], purchases, transport_rows, transports)
|
|
assert [r.source_quantity for r in purchases] == [D(18), D(0)]
|
|
assert [r.source_quantity for r in transport_rows] == [D(10), D(8)]
|
|
|
|
|
|
def test_settlement_pairing_29_consistent_passes():
|
|
from app.services.pl_quota_allocation import validate_allocation_pairing
|
|
from dataclasses import replace
|
|
transports = [permit('y1', '20', refs=('g1',), expiry=1), permit('y2', '10', refs=('g2',), expiry=2)]
|
|
plan = allocate_plan([source()], [permit('g1', '20', expiry=1), permit('g2', '10', expiry=2)], transports)
|
|
def actual_rows(rows):
|
|
return [replace(r.allocation, source_quantity=r.actual_source_quantity, permit_quantity=r.actual_permit_quantity) for r in settle_allocations(rows, {'s': D(29)})]
|
|
assert validate_allocation_pairing([source(quantity='29')], actual_rows(plan.purchase_allocations), actual_rows(plan.transport_allocations), transports) is None
|
|
|
|
|
|
@pytest.mark.parametrize('manual_values', [(8,10,0),(18,0,0)])
|
|
def test_zero_capacity_foreign_row_cannot_unlock_manual_reordering(manual_values):
|
|
from app.services.pl_quota_allocation import Allocation
|
|
rows=[Allocation('transport','s',pid,D(qty),D(qty),'t',unit,seq)
|
|
for pid,qty,unit,seq in [('t1',20,'t',1),('t2',10,'t',2),('t3',0,'kg',3)]]
|
|
manual=[ManualAllocation('transport','s',row.permit_id,D(qty),D(qty)) for row,qty in zip(rows,manual_values)]
|
|
if manual_values==(8,10,0):
|
|
with pytest.raises(AllocationInputError,match='顺序'):
|
|
settle_allocations(rows,{'s':D(18)},manual_actuals=manual)
|
|
else:
|
|
settled=settle_allocations(rows,{'s':D(18)},manual_actuals=manual)
|
|
assert [r.actual_source_quantity for r in settled]==[18,0,0]
|
|
settled=settle_allocations(rows,{'s':D(18)},manual_actuals=[manual[-1]])
|
|
assert [r.actual_source_quantity for r in settled]==[18,0,0]
|