from typing import List, Dict, Any from django.contrib.auth import get_user_model from django.test import TestCase, override_settings from django.utils import timezone from unittest.mock import patch, MagicMock from basic_info import models as basic_models from stock import models as stock_models from business import models as business_models, services, tasks def create_basic_fixtures(): merchant = basic_models.Merchant.objects.create( name='测试商户', type=basic_models.MerchantTypeEnum.FACTORY, ) basic_models.MerchantSetting.objects.create( merchant=merchant, key=basic_models.MerchantSettingKeyEnum.AUTO_CREATE_STOCK_CHANGE_TASKS, type=basic_models.MerchantSettingTypeEnum.BOOL, val_bool=True, ) supplier = basic_models.Supplier.objects.create( merchant=merchant, name='测试供应商', ) warehouse_strict = basic_models.WareHouse.objects.create( merchant=merchant, name='严进仓', mode=basic_models.WareHouseModeEnum.RESTRICT_IN, ) warehouse_relaxed = basic_models.WareHouse.objects.create( merchant=merchant, name='宽进仓', mode=basic_models.WareHouseModeEnum.UNRESTRICTED, ) category = basic_models.ProductCategory.objects.create( merchant=merchant, name='面料', product_prefix='FAB', ) product = basic_models.Product.objects.create( merchant=merchant, category=category, name='测试面料', human_id='FAB-001', unit=basic_models.ProductUnitEnum.METER, ) operator = basic_models.Employee.objects.create( merchant=merchant, name='经办人', status=basic_models.EmployeeStatusEnum.ACTIVE, ) return merchant, supplier, warehouse_strict, warehouse_relaxed, product, operator def create_sales_fixtures(): merchant = basic_models.Merchant.objects.create( name='销售商户', type=basic_models.MerchantTypeEnum.FACTORY, ) basic_models.MerchantSetting.objects.create( merchant=merchant, key=basic_models.MerchantSettingKeyEnum.AUTO_CREATE_STOCK_CHANGE_TASKS, type=basic_models.MerchantSettingTypeEnum.BOOL, val_bool=True, ) customer = basic_models.Customer.objects.create( merchant=merchant, name='客户A', mobile='13800000000', created_by=None, ) warehouse_strict = basic_models.WareHouse.objects.create( merchant=merchant, name='销售严进仓', mode=basic_models.WareHouseModeEnum.RESTRICT_IN, ) warehouse_relaxed = basic_models.WareHouse.objects.create( merchant=merchant, name='销售宽进仓', mode=basic_models.WareHouseModeEnum.UNRESTRICTED, ) warehouse_strict_out = basic_models.WareHouse.objects.create( merchant=merchant, name='销售严出仓', mode=basic_models.WareHouseModeEnum.RESTRICT_IN_OUT, ) category = basic_models.ProductCategory.objects.create( merchant=merchant, name='销售品类', product_prefix='SAL', ) product = basic_models.Product.objects.create( merchant=merchant, category=category, name='销售面料', human_id='SAL-001', unit=basic_models.ProductUnitEnum.METER, ) operator = basic_models.Employee.objects.create( merchant=merchant, name='销售员', status=basic_models.EmployeeStatusEnum.ACTIVE, ) return merchant, customer, warehouse_strict, warehouse_relaxed, warehouse_strict_out, product, operator class PurchaseOrderServiceTestCase(TestCase): def setUp(self): ( self.merchant, self.supplier, self.warehouse_strict, self.warehouse_relaxed, self.product, self.operator, ) = create_basic_fixtures() User = get_user_model() self.user = User.objects.create_user(username='creator', password='pass123') self.strict_items: List[Dict[str, Any]] = [ {'product_id': self.product.id, 'numbers': [10, 5], 'price': '12.5', 'unit': '米'} ] self.relaxed_items: List[Dict[str, Any]] = [ {'product_id': self.product.id, 'quantity': 120, 'num_of_rolls': 3, 'price': '10.0', 'unit': '米'} ] def test_create_purchase_order_does_not_trigger_task_immediately(self): with patch('business.services.create_purchase_order_stock_entries.delay') as mock_delay: purchase_order = services.create_purchase_order( merchant=self.merchant, supplier=self.supplier, order_date=timezone.now().date(), warehouse=self.warehouse_strict, operator=self.operator, items=self.strict_items, remarks='自动化测试', created_by=self.user, ) self.assertIsInstance(purchase_order, business_models.PurchaseOrder) self.assertEqual(purchase_order.status, business_models.PurchaseOrderStatusEnum.PENDING) self.assertEqual(purchase_order.items.count(), 1) item = purchase_order.items.first() self.assertEqual(item.quantity_of_rolls, '10,5') mock_delay.assert_not_called() def test_review_purchase_order_approval_triggers_task(self): purchase_order = services.create_purchase_order( merchant=self.merchant, supplier=self.supplier, order_date=timezone.now().date(), warehouse=self.warehouse_strict, operator=self.operator, items=self.strict_items, created_by=self.user, ) with patch('business.services.create_purchase_order_stock_entries.delay') as mock_delay: reviewed = services.review_purchase_order( purchase_order=purchase_order, target_status=business_models.PurchaseOrderStatusEnum.APPROVED, reviewed_by=self.user, ) self.assertEqual(reviewed.status, business_models.PurchaseOrderStatusEnum.APPROVED) mock_delay.assert_called_once_with( purchase_order_id=purchase_order.id, warehouse_id=self.warehouse_strict.id, items=[{'product_id': self.product.id, 'quantities': ['10', '5']}], created_by_id=self.user.id, ) def test_create_purchase_order_without_items_raises(self): with self.assertRaises(ValueError): services.create_purchase_order( merchant=self.merchant, supplier=self.supplier, order_date=timezone.now().date(), warehouse=self.warehouse_strict, operator=self.operator, items=[], ) def test_review_purchase_order_cancel_without_stock(self): purchase_order = services.create_purchase_order( merchant=self.merchant, supplier=self.supplier, order_date=timezone.now().date(), warehouse=self.warehouse_relaxed, operator=self.operator, items=self.relaxed_items, created_by=self.user, ) self.assertIsNone(purchase_order.items.first().quantity_of_rolls) cancelled = services.review_purchase_order( purchase_order=purchase_order, target_status=business_models.PurchaseOrderStatusEnum.CANCELLED, reviewed_by=self.user, ) self.assertEqual(cancelled.status, business_models.PurchaseOrderStatusEnum.CANCELLED) def test_review_purchase_order_cancel_blocked_after_stock_created(self): purchase_order = services.create_purchase_order( merchant=self.merchant, supplier=self.supplier, order_date=timezone.now().date(), warehouse=self.warehouse_relaxed, operator=self.operator, items=self.relaxed_items, created_by=self.user, ) stock_models.StockChangeRecord.objects.create( merchant=self.merchant, type=stock_models.StockChangeTypeEnum.ADD, warehouse=self.warehouse_relaxed, source_type=stock_models.StockChangeSourceEnum.PURCHASE, source_id=purchase_order.id, ) with self.assertRaises(ValueError): services.review_purchase_order( purchase_order=purchase_order, target_status=business_models.PurchaseOrderStatusEnum.CANCELLED, reviewed_by=self.user, ) def test_review_purchase_order_relaxed_mode_approval_payload(self): purchase_order = services.create_purchase_order( merchant=self.merchant, supplier=self.supplier, order_date=timezone.now().date(), warehouse=self.warehouse_relaxed, operator=self.operator, items=self.relaxed_items, created_by=self.user, ) with patch('business.services.create_purchase_order_stock_entries.delay') as mock_delay: services.review_purchase_order( purchase_order=purchase_order, target_status=business_models.PurchaseOrderStatusEnum.APPROVED, reviewed_by=self.user, ) mock_delay.assert_called_once_with( purchase_order_id=purchase_order.id, warehouse_id=self.warehouse_relaxed.id, items=[{'product_id': self.product.id, 'value': '120.00', 'num_of_rolls': 3}], created_by_id=self.user.id, ) class SalesOrderServiceTestCase(TestCase): def setUp(self): ( self.merchant, self.customer, self.warehouse_strict, self.warehouse_relaxed, self.warehouse_strict_out, self.product, self.operator, ) = create_sales_fixtures() User = get_user_model() self.user = User.objects.create_user(username='sales-creator', password='pass123') self.strict_items = [ {'product_id': self.product.id, 'numbers': [6, 4], 'price': '30.5', 'unit': '米'} ] self.relaxed_items = [ {'product_id': self.product.id, 'quantity': 80, 'num_of_rolls': 2, 'price': '28.0', 'unit': '米'} ] self.strict_out_items = [ { 'product_id': self.product.id, 'consume_detail_ids': [1, 2], 'quantity': 40, 'price': '32.0', 'unit': '米', } ] def test_create_sales_order_success(self): with patch('business.services.create_sales_order_stock_entries.delay') as mock_delay: sales_order = services.create_sales_order( merchant=self.merchant, customer=self.customer, order_date=timezone.now().date(), warehouse=self.warehouse_strict, operator=self.operator, items=self.strict_items, remarks='销售测试', created_by=self.user, ) self.assertIsInstance(sales_order, business_models.SalesOrder) self.assertEqual(sales_order.status, business_models.SalesOrderStatusEnum.PENDING) self.assertEqual(sales_order.items.count(), 1) self.assertEqual(sales_order.items.first().quantity_of_rolls, '6,4') mock_delay.assert_not_called() def test_review_sales_order_triggers_task(self): sales_order = services.create_sales_order( merchant=self.merchant, customer=self.customer, order_date=timezone.now().date(), warehouse=self.warehouse_strict, operator=self.operator, items=self.strict_items, created_by=self.user, ) with patch('business.services.create_sales_order_stock_entries.delay') as mock_delay: reviewed = services.review_sales_order( sales_order=sales_order, target_status=business_models.SalesOrderStatusEnum.APPROVED, reviewed_by=self.user, ) self.assertEqual(reviewed.status, business_models.SalesOrderStatusEnum.APPROVED) mock_delay.assert_called_once_with( sales_order_id=sales_order.id, warehouse_id=self.warehouse_strict.id, items=[{'product_id': self.product.id, 'quantities': ['6', '4']}], created_by_id=self.user.id, ) def test_sales_order_cancel_blocked_after_stock_created(self): sales_order = services.create_sales_order( merchant=self.merchant, customer=self.customer, order_date=timezone.now().date(), warehouse=self.warehouse_relaxed, operator=self.operator, items=self.relaxed_items, created_by=self.user, ) stock_models.StockChangeRecord.objects.create( merchant=self.merchant, type=stock_models.StockChangeTypeEnum.REMOVE, warehouse=self.warehouse_relaxed, source_type=stock_models.StockChangeSourceEnum.SALES, source_id=sales_order.id, ) with self.assertRaises(ValueError): services.review_sales_order( sales_order=sales_order, target_status=business_models.SalesOrderStatusEnum.CANCELLED, reviewed_by=self.user, ) def test_sales_order_requires_consume_ids_for_strict_out(self): with self.assertRaises(ValueError): services.create_sales_order( merchant=self.merchant, customer=self.customer, order_date=timezone.now().date(), warehouse=self.warehouse_strict_out, operator=self.operator, items=[{'product_id': self.product.id, 'quantity': 10, 'price': '20', 'unit': '米'}], created_by=self.user, ) def test_sales_order_stores_consume_ids(self): sales_order = services.create_sales_order( merchant=self.merchant, customer=self.customer, order_date=timezone.now().date(), warehouse=self.warehouse_strict_out, operator=self.operator, items=self.strict_out_items, created_by=self.user, ) item = sales_order.items.first() self.assertEqual(item.consume_detail_ids, '1,2') def test_sales_order_outgoing_missing_consume_ids_raises(self): payload = [{ 'product_id': self.product.id, 'quantity': 30, 'price': '25', 'unit': '米', }] with self.assertRaises(ValueError): services.create_sales_order( merchant=self.merchant, customer=self.customer, order_date=timezone.now().date(), warehouse=self.warehouse_strict_out, operator=self.operator, items=payload, created_by=self.user, ) class PaymentReceiptServiceTestCase(TestCase): def setUp(self): ( self.merchant, self.supplier, self.warehouse_strict, self.warehouse_relaxed, self.product, self.operator, ) = create_basic_fixtures() self.customer = basic_models.Customer.objects.create( merchant=self.merchant, name='客户B', mobile='13888888888', created_by=None, ) def test_create_payment_order_and_approve(self): order = services.create_payment_order( merchant=self.merchant, supplier=self.supplier, payment_date=timezone.now().date(), amount='120.50', operator=self.operator, remarks='付款测试', ) self.assertEqual(order.status, business_models.PaymentOrderStatusEnum.PENDING) reviewed = services.review_payment_order( payment_order=order, target_status=business_models.PaymentOrderStatusEnum.APPROVED, reviewed_by=self.operator, ) self.assertEqual(reviewed.status, business_models.PaymentOrderStatusEnum.APPROVED) def test_create_receipt_order_and_cancel(self): order = services.create_receipt_order( merchant=self.merchant, customer=self.customer, receipt_date=timezone.now().date(), amount='80.00', operator=self.operator, remarks='收款测试', ) cancelled = services.review_receipt_order( receipt_order=order, target_status=business_models.ReceiptOrderStatusEnum.CANCELLED, reviewed_by=self.operator, ) self.assertEqual(cancelled.status, business_models.ReceiptOrderStatusEnum.CANCELLED) def test_payment_amount_must_be_positive(self): with self.assertRaises(ValueError): services.create_payment_order( merchant=self.merchant, supplier=self.supplier, payment_date=timezone.now().date(), amount='-1', operator=self.operator, ) def test_receipt_approval_is_idempotent(self): order = services.create_receipt_order( merchant=self.merchant, customer=self.customer, receipt_date=timezone.now().date(), amount='10', operator=self.operator, ) services.review_receipt_order( receipt_order=order, target_status=business_models.ReceiptOrderStatusEnum.APPROVED, reviewed_by=self.operator, ) reviewed_again = services.review_receipt_order( receipt_order=order, target_status=business_models.ReceiptOrderStatusEnum.APPROVED, reviewed_by=self.operator, ) self.assertEqual(reviewed_again.status, business_models.ReceiptOrderStatusEnum.APPROVED) class PurchaseOrderStockServiceTestCase(TestCase): def setUp(self): ( self.merchant, self.supplier, self.warehouse_strict, self.warehouse_relaxed, self.product, self.operator, ) = create_basic_fixtures() User = get_user_model() self.user = User.objects.create_user(username='svc-user', password='pass123') self.purchase_order = business_models.PurchaseOrder.objects.create( merchant=self.merchant, supplier=self.supplier, purchase_date=timezone.now().date(), operator=self.operator, warehouse=self.warehouse_strict, ) self.items = [{'product_id': self.product.id, 'quantities': ['8']}] def test_service_calls_stock_flow_service(self): with patch('business.services.StockFlowService') as mock_flow_cls: mock_instance = mock_flow_cls.return_value mock_instance.stock_in.return_value = (MagicMock(id=321), [], 2) payload = services.create_purchase_order_stock_entries_sync( purchase_order_id=self.purchase_order.id, warehouse_id=self.warehouse_strict.id, items=self.items, created_by_id=self.user.id, ) mock_flow_cls.assert_called_once_with(merchant=self.merchant, created_by=self.user) mock_instance.stock_in.assert_called_once_with( warehouse_id=self.warehouse_strict.id, source_type=stock_models.StockChangeSourceEnum.PURCHASE, source_id=self.purchase_order.id, items=[{'product_id': self.product.id, 'quantities': ['8']}], ) self.assertEqual(payload['purchase_order_id'], self.purchase_order.id) self.assertEqual(payload['stock_change_record_id'], 321) class SalesOrderStockServiceTestCase(TestCase): def setUp(self): ( self.merchant, self.customer, self.warehouse_strict, self.warehouse_relaxed, self.warehouse_strict_out, self.product, self.operator, ) = create_sales_fixtures() User = get_user_model() self.user = User.objects.create_user(username='sales-svc', password='pass123') self.sales_order = business_models.SalesOrder.objects.create( merchant=self.merchant, customer=self.customer, sales_date=timezone.now().date(), operator=self.operator, warehouse=self.warehouse_strict, ) self.items = [{'product_id': self.product.id, 'quantities': ['5']}] def test_service_calls_stock_out(self): with patch('business.services.StockFlowService') as mock_flow_cls: mock_instance = mock_flow_cls.return_value mock_instance.stock_out.return_value = (MagicMock(id=654), [], 1) payload = services.create_sales_order_stock_entries_sync( sales_order_id=self.sales_order.id, warehouse_id=self.warehouse_strict.id, items=self.items, created_by_id=self.user.id, ) mock_flow_cls.assert_called_once_with(merchant=self.merchant, created_by=self.user) mock_instance.stock_out.assert_called_once_with( warehouse_id=self.warehouse_strict.id, source_type=stock_models.StockChangeSourceEnum.SALES, source_id=self.sales_order.id, items=[{'product_id': self.product.id, 'quantities': ['5']}], ) self.assertEqual(payload['sales_order_id'], self.sales_order.id) self.assertEqual(payload['stock_change_record_id'], 654) def test_service_calls_stock_out_strict_mode(self): strict_order = business_models.SalesOrder.objects.create( merchant=self.merchant, customer=self.customer, sales_date=timezone.now().date(), operator=self.operator, warehouse=self.warehouse_strict_out, ) consume_items = [{'product_id': self.product.id, 'consume_detail_ids': [11, 12], 'quantity': 50}] with patch('business.services.StockFlowService') as mock_flow_cls: mock_instance = mock_flow_cls.return_value mock_instance.stock_out.return_value = (MagicMock(id=987), [], 2) payload = services.create_sales_order_stock_entries_sync( sales_order_id=strict_order.id, warehouse_id=self.warehouse_strict_out.id, items=consume_items, created_by_id=self.user.id, ) mock_flow_cls.assert_called_once_with(merchant=self.merchant, created_by=self.user) mock_instance.stock_out.assert_called_once_with( warehouse_id=self.warehouse_strict_out.id, source_type=stock_models.StockChangeSourceEnum.SALES, source_id=strict_order.id, items=[{'product_id': self.product.id, 'consume_detail_ids': [11, 12], 'quantity': 50}], ) self.assertEqual(payload['sales_order_id'], strict_order.id) self.assertEqual(payload['stock_change_record_id'], 987) @override_settings( CELERY_TASK_ALWAYS_EAGER=True, CELERY_TASK_EAGER_PROPAGATES=True, ) class PurchaseOrderStockTaskTestCase(TestCase): def setUp(self): self.purchase_order_id = 123 self.warehouse_id = 456 self.items = [{'product_id': 1, 'quantities': ['5']}] def test_task_delegates_to_service(self): with patch('business.tasks.business_services.create_purchase_order_stock_entries_sync') as mock_sync: mock_sync.return_value = {'purchase_order_id': self.purchase_order_id} async_result = tasks.create_purchase_order_stock_entries.delay( purchase_order_id=self.purchase_order_id, warehouse_id=self.warehouse_id, items=self.items, created_by_id=999, ) payload = async_result.get(timeout=5) mock_sync.assert_called_once_with( purchase_order_id=self.purchase_order_id, warehouse_id=self.warehouse_id, items=self.items, created_by_id=999, ) self.assertEqual(payload['purchase_order_id'], self.purchase_order_id) self.assertIn('task_id', payload) @override_settings( CELERY_TASK_ALWAYS_EAGER=True, CELERY_TASK_EAGER_PROPAGATES=True, ) class SalesOrderStockTaskTestCase(TestCase): def setUp(self): self.sales_order_id = 321 self.warehouse_id = 654 self.items = [{'product_id': 2, 'quantities': ['4']}] def test_task_delegates_to_service(self): with patch('business.tasks.business_services.create_sales_order_stock_entries_sync') as mock_sync: mock_sync.return_value = {'sales_order_id': self.sales_order_id} async_result = tasks.create_sales_order_stock_entries.delay( sales_order_id=self.sales_order_id, warehouse_id=self.warehouse_id, items=self.items, created_by_id=777, ) payload = async_result.get(timeout=5) mock_sync.assert_called_once_with( sales_order_id=self.sales_order_id, warehouse_id=self.warehouse_id, items=self.items, created_by_id=777, ) self.assertEqual(payload['sales_order_id'], self.sales_order_id) self.assertIn('task_id', payload)