shanxia2.0/backend/tests/test_pl_quota_diagnostics.py

from datetime import date
from decimal import Decimal as D

from app.services.pl_quota_allocation import SourceDemand, PermitCandidate


def permit(id, qty, products, unit='吨', refs=frozenset()):
    return PermitCandidate(id, D(qty), unit, date.max, frozenset(products), refs)


def test_shared_capacity_is_counted_once_per_kind_and_unit():
    from app.services.pl_quota_diagnostics import allocation_diagnostics
    sources = [SourceDemand('a', 'a', D('20'), '吨'), SourceDemand('b', 'b', D('10'), '吨')]
    rows = allocation_diagnostics(sources, [permit('p', '25', ['a', 'b'])], [])
    purchase = rows[0]
    assert purchase['demand'] == '30' and purchase['availableUpperBound'] == '25'
    assert purchase['gapLowerBound'] == '5' and purchase['unit'] == '吨'
    assert rows[1]['gapLowerBound'] == '30'


def test_standard_units_convert_before_the_gap_and_packing_units_do_not():
    from app.services.pl_quota_diagnostics import allocation_diagnostics
    mass = allocation_diagnostics([SourceDemand('a', 'a', D('30'), '吨')], [permit('p', '1000', ['a'], '千克')], [])
    assert mass[0]['availableUpperBound'] == '1' and mass[0]['gapLowerBound'] == '29'
    assert mass[0]['reason'] == '余额不足，请补充该证种额度'
    packed = allocation_diagnostics([SourceDemand('a', 'a', D('30'), '吨')], [permit('p', '1000', ['a'], '瓶')], [])
    assert packed[0]['availableUpperBound'] is None and packed[0]['gapLowerBound'] is None
    assert '人工' in packed[0]['reason']


def test_sufficient_totals_do_not_claim_joint_feasibility():
    from app.services.pl_quota_diagnostics import allocation_diagnostics
    rows = allocation_diagnostics([SourceDemand('a', 'a', D('30'), '吨')], [permit('p', '40', ['a'])], [permit('t', '40', ['a'])])
    assert rows[0]['gapLowerBound'] == '0'
    assert all('引用' in row['reason'] for row in rows)


def test_missing_purchase_link_names_both_certificates():
    from app.services.pl_quota_diagnostics import missing_purchase_link_message
    source = SourceDemand('a', 'a', D('30'), '吨')
    message = missing_purchase_link_message(
        [source],
        [permit('buy', '100', ['a'])],
        [permit('move', '100', ['a'], refs=frozenset())],
        {('purchase', 'buy'): 'G34260485131004', ('transport', 'move'): 'Y34265019298950'},
    )
    assert '运输证 Y34265019298950' in message
    assert '购买证 G34260485131004' in message
    assert '对应购买证档案' in message
    linked = missing_purchase_link_message(
        [source],
        [permit('buy', '100', ['a'])],
        [permit('move', '100', ['a'], refs=frozenset({'buy'}))],
        {('purchase', 'buy'): 'G34260485131004', ('transport', 'move'): 'Y34265019298950'},
    )
    assert linked == ''


def test_foreign_unit_message_names_plan_unit_and_certificates():
    from app.services.pl_quota_diagnostics import foreign_unit_message
    source = SourceDemand('a', 'a', D('1'), 'ping')
    message = foreign_unit_message(
        [source],
        [permit('buy', '100', ['a'], 'dun')],
        [permit('move', '100', ['a'], '吨', refs=frozenset({'buy'}))],
        {('purchase', 'buy'): 'G34260485131004', ('transport', 'move'): 'Y34265019298950'},
    )
    assert '计划产品单位是瓶' in message
    convertible = foreign_unit_message(
        [SourceDemand('a', 'a', D('1'), 'kg')],
        [permit('buy', '100', ['a'], 'dun')],
        [permit('move', '100', ['a'], '吨', refs=frozenset({'buy'}))],
        {('purchase', 'buy'): 'G34260485131004', ('transport', 'move'): 'Y34265019298950'},
    )
    assert convertible == ''
    assert '购买证 G34260485131004（吨）' in message
    assert '运输证 Y34265019298950（吨）' in message
    assert '查看用证' in message
    covered = foreign_unit_message(
        [SourceDemand('a', 'a', D('1'), '吨')],
        [permit('buy', '100', ['a'], '吨')],
        [permit('move', '100', ['a'], '吨', refs=frozenset({'buy'}))],
        {},
    )
    assert covered == ''


def test_archive_shortfall_names_the_certificate_and_the_quantity_gap():
    from app.services.pl_quota_diagnostics import archive_quantity_shortfall_message
    message = archive_quantity_shortfall_message(
        [
            {'certType': 'purchase', 'reason': '总余额仅为上界；请核对每个合同产品范围、运输引用关系及人工分配约束',
             'gapLowerBound': '0', 'availableUpperBound': '1609.999', 'demand': '280', 'unit': '吨'},
            {'certType': 'transport', 'reason': '余额不足，请补充该证种额度',
             'gapLowerBound': '218.001', 'availableUpperBound': '61.999', 'demand': '280', 'unit': '吨'},
        ],
        [{'kind': 'transport', 'certNo': 'Y34265019298950', 'spare': D('61.999'), 'unit': 'dun'},
         {'kind': 'purchase', 'certNo': 'G34260485131004', 'spare': D('1609.999'), 'unit': '吨'}],
        {'delivery': D('300')},
        {'delivery': D('280')},
        {'delivery': 'dun'},
    )
    assert '实际数量已保存，证件没有扣减' in message
    assert '这次实际数量是 300.000 吨' in message
    assert '安排时预占了 20.000 吨' in message
    assert '还要再扣 280.000 吨' in message
    assert '运输证 Y34265019298950 还剩 61.999 吨，不够再扣 280.000 吨' in message
    assert '合同上没有其他运输证' in message
    assert '所以这次最多可以归档 81.999 吨' in message
    assert '购买证余量够用' in message
    assert '请把实际数量改小到上面的数量以内，或先补充运输证，再重新归档' in message


def test_out_of_scope_balance_does_not_inflate_available():
    from app.services.pl_quota_diagnostics import allocation_diagnostics
    rows = allocation_diagnostics([SourceDemand('a', 'a', D('30'), '吨')], [permit('p', '100', ['b'])], [])
    assert rows[0]['availableUpperBound'] == '0'