shanxia2.0/backend/tests/test_pl_plan_product_certs.py

"""证书保存状态保护：仅使用数据库边界替身，不连接业务数据库。"""
from types import SimpleNamespace
from unittest.mock import MagicMock

import pytest
from fastapi import HTTPException

from app.services import pl_plan_service as svc


@pytest.fixture
def cert_case(monkeypatch):
    plan = SimpleNamespace(
        archive_status="pending", follow_status="pending_arrange",
        update_by="original", hazard_cert_ready=0,
    )
    product = SimpleNamespace(
        id="product-1", product_id="crm-1", product_name="测试危化品",
        purchase_cert_url="old.pdf", transport_cert_url=None,
        legal_use_cert_url=None, update_by="original",
    )
    db = MagicMock()
    plan_query, product_query = MagicMock(), MagicMock()
    plan_query.filter.return_value.first.return_value = plan
    product_query.filter.return_value.first.return_value = product
    product_query.filter.return_value.all.return_value = [product]
    db.query.side_effect = lambda model: plan_query if model is svc.PlPlan else product_query
    monkeypatch.setattr(svc, "fetch_hazard_flags_map", lambda *args: {})
    monkeypatch.setattr(svc, "resolve_hazard_flags_from_crm", lambda *args, **kwargs: (True, False))
    auto_ready = MagicMock()
    monkeypatch.setattr(svc, "maybe_auto_set_hazard_cert_ready", auto_ready)
    item = SimpleNamespace(
        id=product.id, purchase_cert_url=" new.pdf,other.pdf ",
        transport_cert_url=" ", legal_use_cert_url=None,
    )
    def save():
        return svc.update_plan_product_certs(
            db, tenant_no="test-tenant", plan_id="plan-1", items=[item], user_id="editor"
        )
    return SimpleNamespace(
        plan=plan, product=product, db=db, save=save,
        auto_ready=auto_ready, plan_query=plan_query,
    )


@pytest.mark.parametrize("state", ["cancelled", "archived", "missing"])
def test_invalid_plan_rejected_before_any_certificate_write(cert_case, state):
    c = cert_case
    if state == "cancelled":
        c.plan.follow_status = "cancelled"
    elif state == "archived":
        c.plan.archive_status = "archived"
    else:
        c.plan_query.filter.return_value.first.return_value = None
    with pytest.raises(HTTPException) as error:
        c.save()
    assert error.value.status_code == (404 if state == "missing" else 400)
    if state == "cancelled":
        assert "已作废" in error.value.detail
    assert c.product.purchase_cert_url == "old.pdf"
    assert c.product.update_by == c.plan.update_by == "original"
    assert c.plan.hazard_cert_ready == 0
    c.db.query.assert_called_once_with(svc.PlPlan)
    c.db.commit.assert_not_called()
    c.auto_ready.assert_not_called()


def test_active_plan_can_save_and_clear_certificate_urls(cert_case):
    c = cert_case
    assert c.save() == [c.product]
    assert c.product.purchase_cert_url == "new.pdf,other.pdf"
    assert c.product.transport_cert_url is None
    assert c.product.legal_use_cert_url is None
    assert c.product.update_by == c.plan.update_by == "editor"
    c.db.commit.assert_called_once()
    c.db.refresh.assert_called_once_with(c.product)
    c.auto_ready.assert_called_once()