276 lines
11 KiB
Python
276 lines
11 KiB
Python
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.sales import create_sales_order, list_sales_orders # 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 # noqa: E402
|
|
from app.models.sales import Customer # noqa: E402
|
|
from app.schemas.database import SalesOrderCreate, SalesOrderItemCreate # noqa: E402
|
|
from app.schemas.document_archives import DocumentArchiveGenerateResult # noqa: E402
|
|
|
|
|
|
class SalesOrderDocumentArchiveTest(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.customer = Customer(
|
|
id=1,
|
|
customer_code="CUS-001",
|
|
customer_name="测试客户",
|
|
short_name="客户",
|
|
address="上海市测试路 1 号",
|
|
credit_days=30,
|
|
status="ACTIVE",
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
self.product = Item(
|
|
id=10,
|
|
item_code="FG-001",
|
|
item_name="测试成品",
|
|
item_type="成品",
|
|
specification="A-100",
|
|
unit_weight_kg=1.25,
|
|
status="ACTIVE",
|
|
created_at=now,
|
|
updated_at=now,
|
|
)
|
|
self.db.add_all([self.customer, self.product])
|
|
self.db.commit()
|
|
|
|
def tearDown(self) -> None:
|
|
self.db.close()
|
|
|
|
def _payload(self) -> SalesOrderCreate:
|
|
return SalesOrderCreate(
|
|
customer_id=self.customer.id,
|
|
sales_employee_id=None,
|
|
promised_date=date(2026, 6, 20),
|
|
delivery_address="上海市测试路 2 号",
|
|
tax_rate=0.13,
|
|
remark="归档测试",
|
|
items=[
|
|
SalesOrderItemCreate(
|
|
product_item_id=self.product.id,
|
|
order_qty=2,
|
|
unit_price=100,
|
|
promised_date=date(2026, 6, 20),
|
|
customer_part_no="CPN-001",
|
|
)
|
|
],
|
|
)
|
|
|
|
def test_create_sales_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,
|
|
)
|
|
)
|
|
db.commit()
|
|
return DocumentArchiveGenerateResult(
|
|
business_id=business_id,
|
|
document_type=document_type,
|
|
document_no="销售2026-00001",
|
|
archive_status="已归档",
|
|
archive_version=1,
|
|
archive_error_message=None,
|
|
)
|
|
|
|
with (
|
|
patch("app.api.routes.sales.ensure_employee_has_permission"),
|
|
patch("app.api.routes.sales.generate_document_archive", side_effect=fake_generate_archive) as archive_mock,
|
|
):
|
|
result = create_sales_order(self._payload(), self.db)
|
|
|
|
archive_mock.assert_called_once_with(self.db, "销售订单", int(result.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.order_id))
|
|
self.assertIsNotNone(saved_archive)
|
|
|
|
def test_list_sales_orders_exposes_latest_archive_fields(self) -> None:
|
|
with (
|
|
patch("app.api.routes.sales.ensure_employee_has_permission"),
|
|
patch(
|
|
"app.api.routes.sales.generate_document_archive",
|
|
return_value=DocumentArchiveGenerateResult(
|
|
business_id=1,
|
|
document_type="销售订单",
|
|
document_no="销售2026-00001",
|
|
archive_status="已归档",
|
|
archive_version=1,
|
|
archive_error_message=None,
|
|
),
|
|
),
|
|
):
|
|
result = create_sales_order(self._payload(), self.db)
|
|
|
|
created_at = datetime(2026, 6, 11, 10, 30, 0)
|
|
self.db.add_all(
|
|
[
|
|
DocumentArchive(
|
|
document_type="销售订单",
|
|
business_id=result.order_id,
|
|
document_no=result.order_no,
|
|
archive_version=1,
|
|
template_version="单据纸面V1",
|
|
file_format="PDF",
|
|
file_name="v1.pdf",
|
|
file_path="/tmp/v1.pdf",
|
|
file_hash="v1",
|
|
status="已归档",
|
|
error_message=None,
|
|
created_by=None,
|
|
created_at=created_at,
|
|
updated_at=created_at,
|
|
),
|
|
DocumentArchive(
|
|
document_type="销售订单",
|
|
business_id=result.order_id,
|
|
document_no=result.order_no,
|
|
archive_version=2,
|
|
template_version="单据纸面V1",
|
|
file_format="PDF",
|
|
file_name="v2.pdf",
|
|
file_path="/tmp/v2.pdf",
|
|
file_hash="v2",
|
|
status="归档失败",
|
|
error_message="render failed",
|
|
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_sales_orders(limit=200, db=self.db)
|
|
|
|
self.assertEqual(len(rows), 1)
|
|
self.assertEqual(rows[0].archive_status, "归档失败")
|
|
self.assertEqual(rows[0].archive_version, 2)
|
|
self.assertEqual(rows[0].archive_created_at, datetime(2026, 6, 11, 10, 35, 0))
|
|
self.assertEqual(rows[0].archive_error_message, "render failed")
|
|
|
|
def test_list_sales_orders_uses_pdf_archive_when_same_version_has_other_format(self) -> None:
|
|
with (
|
|
patch("app.api.routes.sales.ensure_employee_has_permission"),
|
|
patch(
|
|
"app.api.routes.sales.generate_document_archive",
|
|
return_value=DocumentArchiveGenerateResult(
|
|
business_id=1,
|
|
document_type="销售订单",
|
|
document_no="销售2026-00001",
|
|
archive_status="已归档",
|
|
archive_version=1,
|
|
archive_error_message=None,
|
|
),
|
|
),
|
|
):
|
|
result = create_sales_order(self._payload(), self.db)
|
|
|
|
self.db.add_all(
|
|
[
|
|
DocumentArchive(
|
|
document_type="销售订单",
|
|
business_id=result.order_id,
|
|
document_no=result.order_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.order_id,
|
|
document_no=result.order_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_sales_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_create_sales_order_returns_failed_archive_metadata_when_archive_generation_raises(self) -> None:
|
|
with (
|
|
patch("app.api.routes.sales.ensure_employee_has_permission"),
|
|
patch("app.api.routes.sales.generate_document_archive", side_effect=RuntimeError("renderer exploded")),
|
|
):
|
|
result = create_sales_order(self._payload(), self.db)
|
|
|
|
self.assertIsNotNone(result.order_id)
|
|
self.assertTrue(result.order_no)
|
|
self.assertEqual(result.archive_status, "归档失败")
|
|
self.assertIsNone(result.archive_version)
|
|
self.assertTrue(result.archive_error_message and result.archive_error_message.startswith("归档失败："))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|