from __future__ import annotations import unittest from datetime import date, datetime from unittest.mock import patch from sqlalchemy import BigInteger, create_engine, select from sqlalchemy.ext.compiler import compiles from sqlalchemy.orm import Session, sessionmaker @compiles(BigInteger, "sqlite") def _compile_big_integer_for_sqlite(type_, compiler, **kw) -> str: _ = type_, compiler, kw return "INTEGER" import app.models.document_archive # noqa: E402,F401 import app.models.master_data # noqa: E402,F401 import app.models.miniapp # noqa: E402,F401 import app.models.operations # noqa: E402,F401 import app.models.org # noqa: E402,F401 import app.models.planning # noqa: E402,F401 import app.models.sales # noqa: E402,F401 from app.api.routes.purchase import create_purchase_order, list_purchase_orders, update_purchase_order # noqa: E402 from app.models.base import Base # noqa: E402 from app.models.document_archive import DocumentArchive # noqa: E402 from app.models.master_data import Item, Material # noqa: E402 from app.models.operations import Supplier # noqa: E402 from app.schemas.document_archives import DocumentArchiveGenerateResult # noqa: E402 from app.schemas.operations import PurchaseOrderCreate, PurchaseOrderItemCreate, PurchaseOrderUpdate # noqa: E402 class PurchaseOrderDocumentArchiveTest(unittest.TestCase): def setUp(self) -> None: engine = create_engine("sqlite+pysqlite:///:memory:", future=True) Base.metadata.create_all(engine) self.SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False, future=True) self.db: Session = self.SessionLocal() now = datetime.now() self.supplier = Supplier( id=1, supplier_code="SUP-001", supplier_name="测试供应商", short_name="供应商", address="宁波市测试路 1 号", lead_time_days=0, default_tax_rate=0, status="ACTIVE", created_at=now, updated_at=now, ) self.material_item = Item( id=10, item_code="RM-001", item_name="测试原材料", item_type="RAW_MATERIAL", specification="A-100", material_grade="42CrMo", unit_weight_kg=1, status="ACTIVE", created_at=now, updated_at=now, ) self.material = Material( id=100, item_id=self.material_item.id, default_supplier_id=self.supplier.id, purchase_calc_mode="WEIGHT", min_purchase_qty=0, purchase_multiple_qty=0, purchase_multiple_weight_kg=0, lead_time_days=0, quality_rule=None, created_at=now, updated_at=now, ) self.db.add_all([self.supplier, self.material_item, self.material]) self.db.commit() def tearDown(self) -> None: self.db.close() def _payload(self, *, order_weight_kg: float = 12) -> PurchaseOrderCreate: return PurchaseOrderCreate( supplier_id=self.supplier.id, order_date=date(2026, 6, 11), expected_date=date(2026, 6, 20), warning_lead_days=0, purchaser_employee_id=None, target_warehouse_type="RAW", sales_order_ids=[], tax_rate=0.13, remark="采购归档测试", items=[ PurchaseOrderItemCreate( material_item_id=self.material_item.id, order_weight_kg=order_weight_kg, unit_price=100, remark="测试明细", ) ], ) def _archive_result(self, *, business_id: int, document_no: str, archive_version: int = 1) -> DocumentArchiveGenerateResult: return DocumentArchiveGenerateResult( business_id=business_id, document_type="采购订单", document_no=document_no, archive_status="已归档", archive_version=archive_version, archive_error_message=None, ) def test_create_purchase_order_generates_archive_and_returns_archive_metadata(self) -> None: def fake_generate_archive( db: Session, document_type: str, business_id: int, ) -> DocumentArchiveGenerateResult: db.add( DocumentArchive( document_type=document_type, business_id=business_id, document_no="采购2026-00001", archive_version=1, template_version="单据纸面V1", file_format="PDF", file_name="采购2026-00001.pdf", file_path="/tmp/采购2026-00001.pdf", file_hash="abc123", status="已归档", error_message=None, created_by=None, created_at=datetime(2026, 6, 11, 10, 30, 0), updated_at=datetime(2026, 6, 11, 10, 30, 0), ) ) db.commit() return self._archive_result(business_id=business_id, document_no="采购2026-00001") with ( patch("app.api.routes.purchase.ensure_employee_has_permission"), patch("app.api.routes.purchase.generate_document_archive", side_effect=fake_generate_archive) as archive_mock, ): result = create_purchase_order(self._payload(), self.db) archive_mock.assert_called_once_with(self.db, "采购订单", int(result.purchase_order_id)) self.assertEqual(result.archive_status, "已归档") self.assertEqual(result.archive_version, 1) self.assertIsNone(result.archive_error_message) saved_archive = self.db.scalar(select(DocumentArchive).where(DocumentArchive.business_id == result.purchase_order_id)) self.assertIsNotNone(saved_archive) def test_create_purchase_order_returns_failed_archive_metadata_when_archive_generation_raises(self) -> None: with ( patch("app.api.routes.purchase.ensure_employee_has_permission"), patch("app.api.routes.purchase.generate_document_archive", side_effect=RuntimeError("renderer exploded")), ): result = create_purchase_order(self._payload(), self.db) self.assertIsNotNone(result.purchase_order_id) self.assertTrue(result.po_no) self.assertEqual(result.archive_status, "归档失败") self.assertIsNone(result.archive_version) self.assertTrue(result.archive_error_message and result.archive_error_message.startswith("归档失败：")) def test_list_purchase_orders_uses_pdf_archive_when_same_version_has_other_format_without_duplicates(self) -> None: with ( patch("app.api.routes.purchase.ensure_employee_has_permission"), patch( "app.api.routes.purchase.generate_document_archive", return_value=self._archive_result(business_id=1, document_no="采购2026-00001"), ), ): result = create_purchase_order(self._payload(), self.db) self.db.add_all( [ DocumentArchive( document_type="采购订单", business_id=result.purchase_order_id, document_no=result.po_no, archive_version=1, template_version="单据纸面V1", file_format="PDF", file_name="order.pdf", file_path="/tmp/order.pdf", file_hash="pdf", status="已归档", error_message=None, created_by=None, created_at=datetime(2026, 6, 11, 10, 30, 0), updated_at=datetime(2026, 6, 11, 10, 30, 0), ), DocumentArchive( document_type="采购订单", business_id=result.purchase_order_id, document_no=result.po_no, archive_version=1, template_version="单据纸面V1", file_format="DOCX", file_name="order.docx", file_path="/tmp/order.docx", file_hash="docx", status="DOCX不应显示", error_message="wrong format", created_by=None, created_at=datetime(2026, 6, 11, 10, 35, 0), updated_at=datetime(2026, 6, 11, 10, 35, 0), ), ] ) self.db.commit() rows = list_purchase_orders(limit=200, db=self.db) self.assertEqual(len(rows), 1) self.assertEqual(rows[0].archive_status, "已归档") self.assertEqual(rows[0].archive_version, 1) self.assertIsNone(rows[0].archive_error_message) def test_update_purchase_order_generates_archive_and_returns_archive_metadata(self) -> None: with ( patch("app.api.routes.purchase.ensure_employee_has_permission"), patch( "app.api.routes.purchase.generate_document_archive", return_value=self._archive_result(business_id=1, document_no="采购2026-00001"), ), ): created = create_purchase_order(self._payload(), self.db) update_payload = PurchaseOrderUpdate.model_validate(self._payload(order_weight_kg=18).model_dump()) with ( patch("app.api.routes.purchase.ensure_employee_has_permission"), patch( "app.api.routes.purchase.generate_document_archive", return_value=self._archive_result( business_id=created.purchase_order_id, document_no=created.po_no, archive_version=2, ), ) as archive_mock, ): updated = update_purchase_order(created.purchase_order_id, update_payload, self.db) archive_mock.assert_called_once_with(self.db, "采购订单", int(created.purchase_order_id)) self.assertEqual(updated.archive_status, "已归档") self.assertEqual(updated.archive_version, 2) self.assertEqual(updated.total_order_weight_kg, 18) if __name__ == "__main__": unittest.main()