shanxia2.0/backend/tests/test_pl_quota_waiver.py

"""证书齐备只跳过购买证和运输证，不改普通计划的两证校验。"""
from types import SimpleNamespace

from fastapi import HTTPException


def test_ready_keyword_matches_supplier_and_customer(monkeypatch):
    from datetime import date

    from sqlalchemy import create_engine, true
    from sqlalchemy.orm import Session

    from app.models.pl_models import PlPlan
    from app.services import pl_quota_waiver as svc

    monkeypatch.setattr(svc, "_toxic_exists", lambda *args, **kwargs: true())
    monkeypatch.setattr(svc, "apply_pl_scope", lambda query, *args, **kwargs: query)
    monkeypatch.setattr(svc, "_product_names", lambda *args, **kwargs: {})
    engine = create_engine("sqlite://")
    PlPlan.__table__.create(engine)
    db = Session(engine)
    user = SimpleNamespace(tenant_no="t")
    db.add(PlPlan(id="sale", plan_no="PO-S", create_date=date(2026, 11, 1), plan_type="direct",
                  belongs_company="shanxia", tenant_no="t", supplier_name="销售单位甲", customer_name="其他需方"))
    db.add(PlPlan(id="buy", plan_no="PO-B", create_date=date(2026, 11, 2), plan_type="direct",
                  belongs_company="shanxia", tenant_no="t", supplier_name="其他供方", customer_name="购买单位乙"))
    db.commit()
    try:
        assert [row["id"] for row in svc.list_ready_plans(db, user, keyword="销售单位")["list"]] == ["sale"]
        assert [row["id"] for row in svc.list_ready_plans(db, user, keyword="购买单位")["list"]] == ["buy"]
        assert svc.list_ready_plans(db, user, keyword="PO-S")["total"] == 1
        assert svc.list_ready_plans(db, user, keyword="%")["total"] == 0
    finally:
        db.close()
        engine.dispose()


def test_ready_list_orders_by_plan_date_ascending(monkeypatch):
    from app.services import pl_quota_waiver as svc

    monkeypatch.setattr(svc, "_toxic_exists", lambda *args, **kwargs: True)
    monkeypatch.setattr(svc, "apply_pl_scope", lambda query, *args, **kwargs: query)
    monkeypatch.setattr(svc, "_product_names", lambda *args, **kwargs: {})
    orders = []

    class Query:
        def filter(self, *args, **kwargs):
            return self

        def order_by(self, *args):
            orders.extend(args)
            return self

        def count(self):
            return 0

        def offset(self, _n):
            return self

        def limit(self, _n):
            return self

        def all(self):
            return []

    svc.list_ready_plans(SimpleNamespace(query=lambda *_args, **_kwargs: Query()), SimpleNamespace(tenant_no="t"))
    text = " ".join(str(item).lower() for item in orders)
    assert "plan_date asc" in text
    assert "create_time desc" not in text


def test_only_external_toxic_stops_need_both_certs():
    from app.services.pl_quota_waiver import needs_purchase_transport_cert

    assert needs_purchase_transport_cert("客户甲", True) is True
    assert needs_purchase_transport_cert("安庆市山峡化工有限公司", True) is False
    assert needs_purchase_transport_cert(" 安徽云栋化工有限公司 ", True) is False
    assert needs_purchase_transport_cert("客户甲", False) is False


def test_waived_reserve_and_archive_do_not_touch_the_ledger(monkeypatch):
    from app.services.pl_quota_business import apply_plan_quota

    plan = SimpleNamespace(id="p1", quota_cert_waived=1)
    seen = []
    monkeypatch.setattr(
        "app.services.pl_quota_waiver.assert_waived_sources",
        lambda *args, **kwargs: seen.append(kwargs.get("lock")),
    )
    monkeypatch.setattr(
        "app.services.pl_quota_business.execute_usage",
        lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError("不应扣证")),
    )
    db = SimpleNamespace(flush=lambda: None)
    reserve = apply_plan_quota(db, SimpleNamespace(), plan, "reserve", None)
    archive = apply_plan_quota(db, SimpleNamespace(), plan, "settle", None)
    assert reserve == {"operationId": reserve["operationId"], "usageIds": [], "waived": True}
    assert archive["waived"] is True
    assert seen == [True]


def test_mark_ready_allows_weighed_plan_until_quota_is_posted(monkeypatch):
    from fastapi import HTTPException

    from app.models.pl_quota_models import PlPlanCertUsage
    from app.services.pl_quota_waiver import mark_ready

    plan = SimpleNamespace(
        id="p1",
        follow_status="arranged",
        archive_status="draft",
        quota_cert_waived=0,
        quota_cert_waived_by=None,
        quota_cert_waived_by_name=None,
        quota_cert_waived_time=None,
        update_by=None,
        plan_no="PO1",
        plan_date=None,
        license_plate=None,
        supplier_name=None,
        customer_name=None,
    )
    user = SimpleNamespace(id="u1", tenant_no="shanxia2.0", name="归档员", dept_id="")
    usage = SimpleNamespace(usage_status="exception", confirmation_snapshot={"executed": True})

    class Query:
        def __init__(self, rows):
            self.rows = rows

        def filter(self, *args, **kwargs):
            return self

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

        def all(self):
            return list(self.rows)

        def first(self):
            return self.rows[0] if self.rows else None

    class DB:
        def query(self, model):
            if model is PlPlanCertUsage:
                return Query([usage])
            # PlPlan.id 是列对象，齐备判断只关心有没有易制毒装送段。
            return Query([plan])

        def flush(self):
            return None

    monkeypatch.setattr("app.services.pl_quota_planning._plan", lambda *args, **kwargs: plan)
    monkeypatch.setattr("app.services.pl_quota_waiver._toxic_exists", lambda *args, **kwargs: True)
    monkeypatch.setattr("app.services.pl_quota_waiver._product_names", lambda *args, **kwargs: {})
    monkeypatch.setattr("app.services.pl_plan_service.record_plan_change", lambda *args, **kwargs: None)
    monkeypatch.setattr(
        "app.services.pl_quota_business.apply_plan_quota",
        lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError("失败结转不应再释放")),
    )

    ready = mark_ready(DB(), user, "p1")
    assert plan.quota_cert_waived == 1
    assert ready["quotaCertWaived"] is True

    usage.usage_status = "settled"
    plan.quota_cert_waived = 0
    try:
        mark_ready(DB(), user, "p1")
    except HTTPException as exc:
        assert exc.status_code == 409
        assert "已经入账" in exc.detail
    else:
        raise AssertionError("已入账仍允许证书齐备")


def test_waived_arrange_check_keeps_source_errors_and_skips_cert_balance(monkeypatch):
    from app.services.pl_quota_planning import contract_cert_arrange_block

    plan = SimpleNamespace(id="p1", quota_cert_waived=1)
    monkeypatch.setattr("app.services.pl_quota_planning._plan", lambda *args, **kwargs: plan)

    def missing_contract(*args, **kwargs):
        raise HTTPException(409, "装货（客户甲）的「丙酮」已经有产品和计划数量，但这一段还没有关联合同。请先选择合同后再安排运输。")

    monkeypatch.setattr("app.services.pl_quota_waiver.assert_waived_sources", missing_contract)
    monkeypatch.setattr(
        "app.services.pl_quota_usage_service.preview_usage",
        lambda *args, **kwargs: (_ for _ in ()).throw(AssertionError("不应预览证件余额")),
    )
    message = contract_cert_arrange_block(None, None, "p1")
    assert "还没有关联合同" in message

    monkeypatch.setattr("app.services.pl_quota_waiver.assert_waived_sources", lambda *args, **kwargs: None)
    assert contract_cert_arrange_block(None, None, "p1") == ""