78 lines
3.1 KiB
Python
78 lines
3.1 KiB
Python
"""证书保存状态保护：仅使用数据库边界替身，不连接业务数据库。"""
|
|
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()
|