shanxia2.0/backend/tests/test_pl_customer_cert_quota.py

"""客户+品类证件额度：切点、列表默认过滤、齐备语义、整笔选证。"""
from datetime import date, datetime
from decimal import Decimal
from types import SimpleNamespace
from unittest.mock import MagicMock

import pytest
from fastapi import HTTPException


def test_assert_rule_not_configured_blocks_arrange():
    from app.services import pl_cert_quota_rule_config_service as cfg

    db = MagicMock()
    db.query.return_value.filter_by.return_value.first.return_value = None
    with pytest.raises(HTTPException) as exc:
        cfg.assert_cert_quota_rule_configured(db, 't1')
    assert exc.value.status_code == 409
    assert '系统信息' in str(exc.value.detail)


def test_resolve_rule_version_by_effective_at():
    from app.services import pl_cert_quota_rule_config_service as cfg

    db = MagicMock()
    row = SimpleNamespace(effective_at=datetime(2026, 11, 14, 0, 0, 0))
    db.query.return_value.filter_by.return_value.first.return_value = row
    assert cfg.resolve_rule_version_for_create(db, 't', datetime(2026, 11, 13)) == 'legacy'
    assert cfg.resolve_rule_version_for_create(db, 't', datetime(2026, 11, 14)) == 'customer_category'
    db.query.return_value.filter_by.return_value.first.return_value = None
    assert cfg.resolve_rule_version_for_create(db, 't', datetime(2026, 11, 20)) == 'legacy'


def test_apply_plan_quota_new_rule_waive_skips_reserve_not_settle(monkeypatch):
    from app.services import pl_quota_business as biz

    plan = SimpleNamespace(
        id='p1',
        quota_cert_waived=1,
        quota_cert_pending_fulfill=1,
        cert_quota_rule_version='customer_category',
        cert_configuration={},
    )
    called = {}

    def fake_assert(*a, **k):
        called['assert'] = True

    monkeypatch.setattr('app.services.pl_quota_waiver.assert_waived_sources', fake_assert)
    monkeypatch.setattr(
        'app.services.pl_cert_quota_rule_config_service.plan_uses_customer_category',
        lambda p: True,
    )

    result = biz.apply_plan_quota(MagicMock(), SimpleNamespace(id='u'), plan, 'reserve', None)
    assert result.get('waived') is True
    assert called.get('assert') is True

    with pytest.raises(HTTPException) as exc:
        biz.apply_plan_quota(MagicMock(), SimpleNamespace(id='u'), plan, 'settle', None)
    assert '待完善' in str(exc.value.detail)


def test_list_certs_default_excludes_closed_void(monkeypatch):
    from app.services import pl_quota_cert_service as svc

    class Col:
        def __init__(self, name):
            self.name = name

    class FakeModel:
        __tablename__ = 'pl_purchase_cert'
        tenant_no = SimpleNamespace()
        del_flag = 0
        status = SimpleNamespace(notin_=lambda *a, **k: 'STATUS_FILTER')
        create_time = SimpleNamespace(desc=lambda: 'DESC')
        cert_no = seller_name = buyer_name = SimpleNamespace(contains=lambda *a, **k: True)
        __table__ = SimpleNamespace(columns=[Col('id'), Col('status')])

        @staticmethod
        def id():
            return 'id'

    class Query:
        def __init__(self):
            self.filters = []

        def filter(self, *a, **k):
            self.filters.extend(a)
            return self

        def filter_by(self, **k):
            return self

        def order_by(self, *a):
            return self

        def all(self):
            return []

    q = Query()
    db = MagicMock()
    db.query.return_value = q
    monkeypatch.setattr(svc, '_model', lambda t: FakeModel)
    monkeypatch.setattr(svc, 'apply_pl_scope', lambda query, *a, **k: query)
    monkeypatch.setattr(svc, 'get_cert', lambda *a, **k: {})
    svc.list_certs(db, SimpleNamespace(tenant_no='t'), 'purchase')
    assert any(f == 'STATUS_FILTER' for f in q.filters)


def test_whole_cover_skips_insufficient_earlier_purchase():
    """规则 A：早到期 10 吨不够则跳过，选用晚到期够量的证。"""
    from app.services import pl_quota_customer_select as sel
    from app.schemas.pl.quota_usage import UsageSource

    source = UsageSource(
        source_stop_type='delivery',
        source_stop_id='s1',
        source_stop_product_id='sp1',
        plan_product_id='pp1',
        contract_id='c1',
        contract_product_id='cp1',
        product_id='prod1',
        quantity=Decimal('50'),
        unit='吨',
        fingerprint='x' * 64,
        snapshot={'customerId': 'cust1'},
        requires_transport=False,
    )

    early = SimpleNamespace(
        id='buy-early', customer_id='cust1', category_id='cat1', status='active',
        verified_by='u', verified_time=datetime.now(), valid_from=None, valid_to=date(2026, 12, 1),
        total_qty=Decimal('10'), used_qty=Decimal('0'), reserved_qty=Decimal('0'),
        quantity_unit='吨', opening_single_use_consumed=0,
    )
    late = SimpleNamespace(
        id='buy-late', customer_id='cust1', category_id='cat1', status='active',
        verified_by='u', verified_time=datetime.now(), valid_from=None, valid_to=date(2027, 6, 1),
        total_qty=Decimal('100'), used_qty=Decimal('0'), reserved_qty=Decimal('0'),
        quantity_unit='吨', opening_single_use_consumed=0,
    )

    class PurchaseQuery:
        def filter(self, *a, **k):
            return self

        def order_by(self, *a, **k):
            return self

        def all(self):
            return [early, late]

    class LogQuery:
        def filter_by(self, **k):
            return self

        def first(self):
            return SimpleNamespace(id='opening')

    db = MagicMock()

    def query(model):
        name = getattr(model, '__name__', str(model))
        if 'Purchase' in name or model is early.__class__:
            return PurchaseQuery()
        return LogQuery()

    # route by model class name via side_effect
    from app.models.pl_quota_models import PlPurchaseCert, PlCertQtyLog

    def q_side(model):
        if model is PlPurchaseCert:
            return PurchaseQuery()
        if model is PlCertQtyLog:
            return LogQuery()
        return MagicMock()

    db.query.side_effect = q_side
    user = SimpleNamespace(tenant_no='t')

    import app.services.pl_quota_customer_select as mod
    orig_cat = mod.category_ids_for_product
    mod.category_ids_for_product = lambda db, tenant, pid: {'cat1'}
    try:
        purchase_ids, transport_ids, manuals = sel.select_whole_cover_certs(
            db, user, [source], vehicle_no='皖A12345', on_date=date(2026, 11, 14),
        )
    finally:
        mod.category_ids_for_product = orig_cat

    assert purchase_ids == ['buy-late']
    assert transport_ids == []
    assert len(manuals) == 1
    assert manuals[0].permit_id == 'buy-late'
    assert manuals[0].permit_quantity == Decimal('50')


def test_save_configuration_blocked_for_customer_category(monkeypatch):
    from app.services import pl_quota_planning as planning
    from app.schemas.pl.quota_planning import PlanQuotaConfiguration

    plan = SimpleNamespace(
        id='p1',
        cert_quota_rule_version='customer_category',
        transport_status='pending_arrange',
        archive_status='none',
        cert_configuration={},
    )
    monkeypatch.setattr(planning, '_plan', lambda *a, **k: plan)
    with pytest.raises(HTTPException) as exc:
        planning.save_configuration(
            MagicMock(),
            SimpleNamespace(id='u', tenant_no='t', dept_id=''),
            'p1',
            PlanQuotaConfiguration(),
        )
    assert exc.value.status_code == 409
    assert '自动选证' in str(exc.value.detail)


def test_legacy_waive_still_skips_settle(monkeypatch):
    from app.services import pl_quota_business as biz

    plan = SimpleNamespace(
        id='p1',
        quota_cert_waived=1,
        quota_cert_pending_fulfill=0,
        cert_quota_rule_version='legacy',
        cert_configuration={},
    )
    monkeypatch.setattr(
        'app.services.pl_cert_quota_rule_config_service.plan_uses_customer_category',
        lambda p: False,
    )
    monkeypatch.setattr('app.services.pl_quota_waiver.assert_waived_sources', lambda *a, **k: None)
    result = biz.apply_plan_quota(MagicMock(), SimpleNamespace(id='u'), plan, 'settle', None)
    assert result.get('waived') is True


def test_null_valid_to_sorted_after_dated_certs():
    """无到期日的证不能排在有到期日之前（MySQL NULL ASC 默认会错序）。"""
    from app.services import pl_quota_customer_select as sel
    from app.schemas.pl.quota_usage import UsageSource
    from app.models.pl_quota_models import PlPurchaseCert, PlCertQtyLog

    source = UsageSource(
        source_stop_type='delivery',
        source_stop_id='s1',
        source_stop_product_id='sp1',
        plan_product_id='pp1',
        contract_id='c1',
        contract_product_id='cp1',
        product_id='prod1',
        quantity=Decimal('10'),
        unit='吨',
        fingerprint='z' * 64,
        snapshot={'customerId': 'cust1'},
        requires_transport=False,
    )
    unbounded = SimpleNamespace(
        id='buy-null', customer_id='cust1', category_id='cat1', status='active',
        verified_by='u', verified_time=datetime.now(), valid_from=None, valid_to=None,
        total_qty=Decimal('100'), used_qty=Decimal('0'), reserved_qty=Decimal('0'),
        quantity_unit='吨', opening_single_use_consumed=0,
    )
    dated = SimpleNamespace(
        id='buy-dated', customer_id='cust1', category_id='cat1', status='active',
        verified_by='u', verified_time=datetime.now(), valid_from=None, valid_to=date(2026, 12, 1),
        total_qty=Decimal('100'), used_qty=Decimal('0'), reserved_qty=Decimal('0'),
        quantity_unit='吨', opening_single_use_consumed=0,
    )

    class PurchaseQuery:
        def __init__(self):
            self._order = None

        def filter(self, *a, **k):
            return self

        def order_by(self, *exprs):
            self._order = exprs
            return self

        def all(self):
            # 应先按 valid_to IS NULL 升序（非空在前），再按 valid_to ASC
            assert self._order is not None and len(self._order) >= 2
            return [dated, unbounded]

    class LogQuery:
        def filter_by(self, **k):
            return self

        def first(self):
            return SimpleNamespace(id='opening')

    db = MagicMock()

    def q_side(model):
        if model is PlPurchaseCert:
            return PurchaseQuery()
        if model is PlCertQtyLog:
            return LogQuery()
        return MagicMock()

    db.query.side_effect = q_side
    orig = sel.category_ids_for_product
    sel.category_ids_for_product = lambda db, tenant, pid: {'cat1'}
    try:
        purchase_ids, _, manuals = sel.select_whole_cover_certs(
            db, SimpleNamespace(tenant_no='t'), [source], vehicle_no='皖A1', on_date=date(2026, 11, 14),
        )
    finally:
        sel.category_ids_for_product = orig
    assert purchase_ids == ['buy-dated']
    assert manuals[0].permit_id == 'buy-dated'


def test_same_city_source_skips_transport_requirement():
    """同市来源 requires_transport=False，整笔选证可不带运输证。"""
    from app.services import pl_quota_customer_select as sel
    from app.schemas.pl.quota_usage import UsageSource
    from app.models.pl_quota_models import PlPurchaseCert, PlCertQtyLog

    source = UsageSource(
        source_stop_type='delivery',
        source_stop_id='s1',
        source_stop_product_id='sp1',
        plan_product_id='pp1',
        contract_id='c1',
        contract_product_id='cp1',
        product_id='prod1',
        quantity=Decimal('20'),
        unit='吨',
        fingerprint='y' * 64,
        snapshot={'customerId': 'cust1', 'sameCity': 1},
        requires_transport=False,
    )
    purchase = SimpleNamespace(
        id='buy1', customer_id='cust1', category_id='cat1', status='active',
        verified_by='u', verified_time=datetime.now(), valid_from=None, valid_to=date(2027, 1, 1),
        total_qty=Decimal('100'), used_qty=Decimal('0'), reserved_qty=Decimal('0'),
        quantity_unit='吨', opening_single_use_consumed=0,
    )

    class PurchaseQuery:
        def filter(self, *a, **k):
            return self

        def order_by(self, *a, **k):
            return self

        def all(self):
            return [purchase]

    class LogQuery:
        def filter_by(self, **k):
            return self

        def first(self):
            return SimpleNamespace(id='opening')

    db = MagicMock()

    def q_side(model):
        if model is PlPurchaseCert:
            return PurchaseQuery()
        if model is PlCertQtyLog:
            return LogQuery()
        return MagicMock()

    db.query.side_effect = q_side
    orig = sel.category_ids_for_product
    sel.category_ids_for_product = lambda db, tenant, pid: {'cat1'}
    try:
        purchase_ids, transport_ids, manuals = sel.select_whole_cover_certs(
            db, SimpleNamespace(tenant_no='t'), [source], vehicle_no='皖A1', on_date=date(2026, 11, 14),
        )
    finally:
        sel.category_ids_for_product = orig
    assert purchase_ids == ['buy1']
    assert transport_ids == []
    assert manuals[0].permit_id == 'buy1'


def test_candidate_context_customer_rule_skips_contract_scope(monkeypatch):
    """新规则账本候选不以合同绑证 scope 判定适用，否则预占会误报人工分配不适用。"""
    from datetime import date
    from decimal import Decimal
    from types import SimpleNamespace
    from app.services import pl_quota_usage_service as usage
    from app.schemas.pl.quota_usage import UsageSource

    plan = SimpleNamespace(id='plan', cert_quota_rule_version='customer_category', plan_no='P1')
    source = UsageSource(
        source_stop_type='delivery',
        source_stop_id='s1',
        source_stop_product_id='sp1',
        plan_product_id='pp1',
        contract_id='c1',
        contract_product_id='cp1',
        product_id='prod1',
        quantity=Decimal('50'),
        unit='dun',
        fingerprint='f' * 64,
        snapshot={'customerId': 'cust1'},
        requires_transport=True,
    )
    request = SimpleNamespace(
        sources=[source],
        on_date=date(2026, 11, 15),
        vehicle_no='皖A9TEST1',
        manual_allocations=[],
    )
    cert = SimpleNamespace(
        id='buy1', status='active', verified_time=date(2026, 11, 1), verified_by='u',
        valid_from=date(2026, 11, 1), valid_to=date(2027, 1, 1), valid_use_text='多次有效',
        total_qty=Decimal('100'), used_qty=Decimal('0'), reserved_qty=Decimal('0'),
        quantity_unit='dun', vehicle_numbers_json=['皖A9TEST1'], opening_single_use_consumed=0,
        revision=1, cert_no='G1',
    )

    class Q:
        def __init__(self, row=None, rows=None):
            self._row = row
            self._rows = rows or ([] if row is None else [row])

        def first(self):
            return self._row

        def all(self):
            return self._rows

    def fake_query(db, user, model, lock=False, **filters):
        name = getattr(model, '__name__', '')
        if name == 'PlCertQtyLog':
            return Q(SimpleNamespace(id='opening'))
        if name == 'PlTransportCertPurchaseRef':
            return Q(rows=[])
        if name in ('PlContractCertLink', 'PlContractCertProduct'):
            raise AssertionError('customer_category 不应查询合同绑证 scope')
        return Q()

    monkeypatch.setattr(usage, '_query', fake_query)
    monkeypatch.setattr(usage, '_contract', lambda *a, **k: SimpleNamespace(status='executing', contract_type='sale'))
    monkeypatch.setattr(
        'app.services.pl_cert_quota_rule_config_service.plan_uses_customer_category',
        lambda p: True,
    )
    # product ownership check uses _query for CrmContractProduct
    def fake_query2(db, user, model, lock=False, **filters):
        name = getattr(model, '__name__', '')
        if name == 'CrmContractProduct':
            return Q(SimpleNamespace(product_id='prod1'))
        return fake_query(db, user, model, lock=lock, **filters)

    monkeypatch.setattr(usage, '_query', fake_query2)

    sources, scopes, candidates = usage._candidate_context(
        MagicMock(), SimpleNamespace(tenant_no='t', id='u'), request, plan,
        {('purchase', 'buy1'): cert}, {}, {}, {('purchase', 'buy1')},
    )
    assert 'delivery:sp1' in sources
    assert scopes[(('purchase', 'buy1'), 'delivery:sp1')] == (None, None)
    assert candidates['purchase'][0].product_ids == frozenset({'delivery:sp1'})


def test_list_certs_filters_by_customer_id(monkeypatch):
    from app.services import pl_quota_cert_service as svc

    class Col:
        def __init__(self, name):
            self.name = name

    class CustomerCol:
        def __eq__(self, other):
            return ('CUSTOMER_FILTER', other)

    class FakeModel:
        __tablename__ = 'pl_purchase_cert'
        tenant_no = SimpleNamespace()
        del_flag = 0
        customer_id = CustomerCol()
        status = SimpleNamespace(notin_=lambda *a, **k: 'STATUS_FILTER')
        create_time = SimpleNamespace(desc=lambda: 'DESC')
        cert_no = seller_name = buyer_name = SimpleNamespace(contains=lambda *a, **k: True)
        __table__ = SimpleNamespace(columns=[Col('id'), Col('status')])

    class Query:
        def __init__(self):
            self.filters = []

        def filter(self, *a, **k):
            self.filters.extend(a)
            return self

        def filter_by(self, **k):
            return self

        def order_by(self, *a):
            return self

        def all(self):
            return []

    q = Query()
    db = MagicMock()
    db.query.return_value = q
    monkeypatch.setattr(svc, '_model', lambda t: FakeModel)
    monkeypatch.setattr(svc, 'apply_pl_scope', lambda query, *a, **k: query)
    monkeypatch.setattr(svc, 'get_cert', lambda *a, **k: {})
    svc.list_certs(db, SimpleNamespace(tenant_no='t'), 'purchase', customer_id='cust-1')
    assert ('CUSTOMER_FILTER', 'cust-1') in q.filters
    assert any(f == 'STATUS_FILTER' for f in q.filters)


def test_is_single_use_consumed_blocks_other_plan_usage():
    from app.services.pl_quota_cert_service import is_single_use_consumed
    from app.models.pl_quota_models import PlPlanCertUsage

    cert = SimpleNamespace(id='buy1', valid_use_text='一次有效', opening_single_use_consumed=0)
    usage_row = SimpleNamespace(
        plan_id='other-plan',
        usage_status='reserved',
        confirmation_snapshot=None,
    )

    class UsageQuery:
        def filter_by(self, **k):
            return self

        def all(self):
            return [usage_row]

    db = MagicMock()

    def q_side(model):
        assert model is PlPlanCertUsage
        return UsageQuery()

    db.query.side_effect = q_side
    user = SimpleNamespace(tenant_no='t')
    assert is_single_use_consumed(db, user, 'purchase', cert) is True
    assert is_single_use_consumed(db, user, 'purchase', cert, exclude_plan_id='other-plan') is False
    multi = SimpleNamespace(id='buy2', valid_use_text='多次有效', opening_single_use_consumed=1)
    assert is_single_use_consumed(db, user, 'purchase', multi) is False


def test_whole_cover_skips_once_used_purchase_with_remaining(monkeypatch):
    """一次有效证已被他计划占用时，即使余量足够也不能再选。"""
    from app.services import pl_quota_customer_select as sel
    from app.schemas.pl.quota_usage import UsageSource

    source = UsageSource(
        source_stop_type='delivery',
        source_stop_id='s1',
        source_stop_product_id='sp1',
        plan_product_id='pp1',
        contract_id='c1',
        contract_product_id='cp1',
        product_id='prod1',
        quantity=Decimal('10'),
        unit='吨',
        fingerprint='y' * 64,
        snapshot={'customerId': 'cust1'},
        requires_transport=False,
    )
    used_once = SimpleNamespace(
        id='buy-once', customer_id='cust1', category_id='cat1', status='active',
        verified_by='u', verified_time=datetime.now(), valid_from=None, valid_to=date(2026, 12, 1),
        total_qty=Decimal('100'), used_qty=Decimal('10'), reserved_qty=Decimal('0'),
        quantity_unit='吨', opening_single_use_consumed=0, valid_use_text='一次有效',
    )
    fresh = SimpleNamespace(
        id='buy-fresh', customer_id='cust1', category_id='cat1', status='active',
        verified_by='u', verified_time=datetime.now(), valid_from=None, valid_to=date(2027, 6, 1),
        total_qty=Decimal('100'), used_qty=Decimal('0'), reserved_qty=Decimal('0'),
        quantity_unit='吨', opening_single_use_consumed=0, valid_use_text='多次有效',
    )

    class PurchaseQuery:
        def filter(self, *a, **k):
            return self

        def order_by(self, *a, **k):
            return self

        def all(self):
            return [used_once, fresh]

    class LogQuery:
        def filter_by(self, **k):
            return self

        def first(self):
            return SimpleNamespace(id='opening')

    from app.models.pl_quota_models import PlPurchaseCert, PlCertQtyLog

    db = MagicMock()

    def q_side(model):
        if model is PlPurchaseCert:
            return PurchaseQuery()
        if model is PlCertQtyLog:
            return LogQuery()
        return MagicMock()

    db.query.side_effect = q_side
    user = SimpleNamespace(tenant_no='t')
    monkeypatch.setattr(sel, 'category_ids_for_product', lambda db, tenant, pid: {'cat1'})
    monkeypatch.setattr(
        sel,
        'is_single_use_consumed',
        lambda db, user, kind, cert, **kw: cert.id == 'buy-once',
    )
    purchase_ids, transport_ids, manuals = sel.select_whole_cover_certs(
        db, user, [source], vehicle_no='皖A12345', on_date=date(2026, 11, 14),
    )
    assert purchase_ids == ['buy-fresh']
    assert transport_ids == []
    assert manuals[0].permit_id == 'buy-fresh'


def test_list_customers_requires_customer_cert_counts(monkeypatch):
    from app.services import pl_quota_statistics as stats

    class Q:
        def __init__(self, total=1, rows=None):
            self._total = total
            self._rows = rows or []

        def join(self, *a, **k):
            return self

        def filter(self, *a, **k):
            return self

        def count(self):
            return self._total

        def with_entities(self, *a, **k):
            return self

        def order_by(self, *a, **k):
            return self

        def offset(self, *a, **k):
            return self

        def limit(self, *a, **k):
            return self

        def all(self):
            return self._rows

    customer = SimpleNamespace(id='cust1', customer_no='C1', customer_name='甲客户', create_time=None)
    monkeypatch.setattr(stats, '_customer_cert_counts', lambda db, user: SimpleNamespace(c=SimpleNamespace(customer_id='id', cert_count='cnt')))
    monkeypatch.setattr(stats, '_customers', lambda db, user: Q(rows=[(customer, 3)]))
    result = stats.list_customers(MagicMock(), SimpleNamespace(tenant_no='t', to_user_info=lambda: {}))
    assert result['total'] == 1
    assert result['list'][0]['customerName'] == '甲客户'
    assert result['list'][0]['certCount'] == 3


def test_get_cert_customer_name_from_crm_not_consignee(monkeypatch):
    """详情/编辑用 CRM 系统客户名，不用证面收货单位。"""
    from app.services import pl_quota_cert_service as svc

    class Col:
        def __init__(self, name):
            self.name = name

    row = SimpleNamespace(
        id='t1',
        cert_no='Y1',
        customer_id='cust-kanghua',
        consignee_name='国能宿州热电有限公司',
        valid_use_text='多次有效',
        total_qty=Decimal('10'),
        used_qty=Decimal('0'),
        reserved_qty=Decimal('0'),
        __table__=SimpleNamespace(
            columns=[
                Col('id'),
                Col('cert_no'),
                Col('customer_id'),
                Col('consignee_name'),
                Col('valid_use_text'),
                Col('total_qty'),
                Col('used_qty'),
                Col('reserved_qty'),
            ]
        ),
    )
    monkeypatch.setattr(svc, '_get_row', lambda *a, **k: row)
    monkeypatch.setattr(svc, 'is_single_use_consumed', lambda *a, **k: False)
    monkeypatch.setattr(
        svc, '_customer_display_name', lambda *a, **k: '滁州康华电子材料有限公司'
    )
    monkeypatch.setattr(svc, '_refs', lambda *a, **k: [])
    monkeypatch.setattr(svc, '_has_history', lambda *a, **k: False)
    monkeypatch.setattr(svc, '_transport_derived_categories', lambda *a, **k: [])

    opening = MagicMock()
    opening.filter_by.return_value.first.return_value = None
    db = MagicMock()
    db.query.return_value = opening

    result = svc.get_cert(db, SimpleNamespace(tenant_no='t'), 'transport', 't1')
    assert result['customerName'] == '滁州康华电子材料有限公司'
    assert result['consigneeName'] == '国能宿州热电有限公司'
    assert result['customerId'] == 'cust-kanghua'