from django.core.exceptions import ValidationError from django.test import TestCase from decimal import Decimal from unittest.mock import patch, MagicMock import logging from basic_info.models import Product, WareHouse, ProductUnitEnum, ProductCategory, Merchant, MerchantTypeEnum, WareHouseModeEnum from . import models, services class StockServicesTestCase(TestCase): """库存服务层测试""" @classmethod def setUpTestData(cls): """设置测试数据（类级别，所有测试方法共享）""" # 产品单位（使用枚举值） cls.unit_meter = ProductUnitEnum.METER cls.unit_kg = ProductUnitEnum.KG # 创建测试商户 cls.merchant = Merchant.objects.create( name="测试商户", type=MerchantTypeEnum.STORE, area="测试地区", mobile="13800138000", contact="测试联系人" ) # 创建产品类别 cls.fabric_category = ProductCategory.objects.create( merchant=cls.merchant, name="布料类", description="各种布料产品", product_prefix="FAB" ) # 创建仓库 cls.warehouse_main = WareHouse.objects.create( merchant=cls.merchant, name="主仓库", location="主仓库地址" ) cls.warehouse_backup = WareHouse.objects.create( merchant=cls.merchant, name="备用仓库", location="备用仓库地址" ) # 创建产品 cls.product_fabric_a = Product.objects.create( merchant=cls.merchant, category=cls.fabric_category, name="布料A", human_id="FABRIC-A-001", color="红色", spec="规格A", single_price_in=Decimal('100.00'), single_price_out=Decimal('120.00'), unit=cls.unit_meter ) cls.product_fabric_b = Product.objects.create( merchant=cls.merchant, category=cls.fabric_category, name="布料B", human_id="FABRIC-B-001", color="蓝色", spec="规格B", single_price_in=Decimal('150.00'), single_price_out=Decimal('180.00'), unit=cls.unit_kg ) def setUp(self): """每个测试方法执行前的设置""" self.purchase_source_id = 1001 # 创建库存变动记录（入库） self.stock_change_in = models.StockChangeRecord.objects.create( merchant=self.merchant, type=models.StockChangeTypeEnum.ADD, source_type=models.StockChangeSourceEnum.PURCHASE, source_id=self.purchase_source_id, warehouse=self.warehouse_main, is_finished=False ) # 创建库存变动记录（出库） self.stock_change_out = models.StockChangeRecord.objects.create( merchant=self.merchant, type=models.StockChangeTypeEnum.REMOVE, source_type=models.StockChangeSourceEnum.SALES, source_id=1, warehouse=self.warehouse_main, is_finished=False ) # 创建库存变动明细 self.stock_detail_in = models.StockChangeDetail.objects.create( merchant=self.merchant, stock_change_record=self.stock_change_in, product=self.product_fabric_a, quantity=Decimal('100'), unit=self.unit_meter ) self.stock_detail_out = models.StockChangeDetail.objects.create( merchant=self.merchant, stock_change_record=self.stock_change_out, product=self.product_fabric_a, quantity=Decimal('50'), unit=self.unit_meter ) class StockChangeRecordValidationTestCase(StockServicesTestCase): """验证 StockChangeRecord.clean 的来源类型校验""" def test_add_record_accepts_combine_source(self): record = models.StockChangeRecord( merchant=self.merchant, warehouse=self.warehouse_main, type=models.StockChangeTypeEnum.ADD, source_type=models.StockChangeSourceEnum.COMBINE, ) try: record.full_clean() except ValidationError as exc: self.fail(f'合并应视为入库来源，但触发校验错误: {exc}') def test_remove_record_accepts_explode_source(self): record = models.StockChangeRecord( merchant=self.merchant, warehouse=self.warehouse_main, type=models.StockChangeTypeEnum.REMOVE, source_type=models.StockChangeSourceEnum.EXPLODE, ) try: record.full_clean() except ValidationError as exc: self.fail(f'拆卷应视为出库来源，但触发校验错误: {exc}') def test_add_record_rejects_outgoing_source(self): record = models.StockChangeRecord( merchant=self.merchant, warehouse=self.warehouse_main, type=models.StockChangeTypeEnum.ADD, source_type=models.StockChangeSourceEnum.SALES, ) with self.assertRaises(ValidationError): record.full_clean() def test_remove_record_rejects_incoming_source(self): record = models.StockChangeRecord( merchant=self.merchant, warehouse=self.warehouse_main, type=models.StockChangeTypeEnum.REMOVE, source_type=models.StockChangeSourceEnum.PURCHASE, ) with self.assertRaises(ValidationError): record.full_clean() class FindInventoryTestCase(StockServicesTestCase): """测试 find_inventory 函数""" def test_find_existing_inventory(self): """测试查找已存在的库存记录""" # 先创建一个库存记录 inventory = models.Inventory.objects.create( merchant=self.merchant, product=self.product_fabric_a, warehouse=self.warehouse_main, quantity=50, num_of_rolls=5 ) # 测试查找 found_inventory = services.find_inventory( product_id=self.product_fabric_a.id, warehouse_id=self.warehouse_main.id ) self.assertIsNotNone(found_inventory) self.assertEqual(found_inventory.id, inventory.id) self.assertEqual(found_inventory.quantity, 50) self.assertEqual(found_inventory.num_of_rolls, 5) def test_find_nonexistent_inventory(self): """测试查找不存在的库存记录""" result = services.find_inventory( product_id=self.product_fabric_a.id, warehouse_id=self.warehouse_main.id ) self.assertIsNone(result) def test_find_inventory_wrong_warehouse(self): """测试在错误的仓库中查找库存""" # 在主仓库创建库存 models.Inventory.objects.create( merchant=self.merchant, product=self.product_fabric_a, warehouse=self.warehouse_main, quantity=50, num_of_rolls=5 ) # 在备用仓库查找 result = services.find_inventory( product_id=self.product_fabric_a.id, warehouse_id=self.warehouse_backup.id ) self.assertIsNone(result) class CreateStockSnapshotTestCase(StockServicesTestCase): """测试 create_stock_snapshot 函数""" def setUp(self): super().setUp() # 创建库存记录 self.inventory = models.Inventory.objects.create( merchant=self.merchant, product=self.product_fabric_a, warehouse=self.warehouse_main, quantity=150, num_of_rolls=15 ) def test_create_snapshot_for_incoming_stock(self): """测试为入库记录创建快照""" with patch.object(services.logger, 'info') as mock_logger: snapshot = services.create_stock_snapshot( self.stock_detail_in, self.inventory ) self.assertIsNotNone(snapshot) self.assertEqual(snapshot.product_id, self.product_fabric_a.id) self.assertEqual(snapshot.warehouse_id, self.warehouse_main.id) self.assertEqual(snapshot.delta, 100) # 入库数量为正 self.assertEqual(snapshot.quantity_before, 50) # 150 - 100 self.assertEqual(snapshot.quantity_after, 150) self.assertEqual(snapshot.stock_change_record_id, self.stock_change_in.id) # 验证日志调用 mock_logger.assert_called_once() def test_create_snapshot_for_outgoing_stock(self): """测试为出库记录创建快照""" snapshot = services.create_stock_snapshot( self.stock_detail_out, self.inventory ) self.assertIsNotNone(snapshot) self.assertEqual(snapshot.delta, -50) # 出库数量为负 self.assertEqual(snapshot.quantity_before, 200) # 150 - (-50) self.assertEqual(snapshot.quantity_after, 150) def test_create_snapshot_with_invalid_detail(self): """测试使用无效明细记录创建快照""" # 测试 None 参数 with self.assertRaises(ValueError) as cm: services.create_stock_snapshot(None, self.inventory) self.assertIn("无效的库存变动明细记录", str(cm.exception)) # 测试未保存的明细记录 unsaved_detail = models.StockChangeDetail( merchant=self.merchant, stock_change_record=self.stock_change_in, product=self.product_fabric_a, quantity=50 ) with self.assertRaises(ValueError): services.create_stock_snapshot(unsaved_detail, self.inventory) class MakeStockChangeCompletedTestCase(StockServicesTestCase): """测试 make_stock_change_completed 函数""" def test_complete_invalid_record(self): """测试完成无效的库存变动记录""" result = services.make_stock_change_completed(None) self.assertFalse(result) # 测试未保存的记录 unsaved_record = models.StockChangeRecord( merchant=self.merchant, type=models.StockChangeTypeEnum.ADD, warehouse=self.warehouse_main ) result = services.make_stock_change_completed(unsaved_record) self.assertFalse(result) def test_complete_already_finished_record(self): """测试完成已经完成的记录""" self.stock_change_in.is_finished = True self.stock_change_in.save() result = services.make_stock_change_completed(self.stock_change_in) self.assertTrue(result) @patch('stock.services.find_inventory') @patch('stock.services.create_stock_snapshot') def test_complete_inbound_record_new_inventory(self, mock_snapshot, mock_find): """测试完成入库记录（新建库存）""" # 模拟没有找到现有库存 mock_find.return_value = None mock_snapshot.return_value = MagicMock() result = services.make_stock_change_completed(self.stock_change_in) self.assertTrue(result) self.assertTrue(self.stock_change_in.is_finished) self.assertIsNotNone(self.stock_change_in.finished_at) # 验证创建了新的库存记录 new_inventory = models.Inventory.objects.get( product=self.product_fabric_a, warehouse=self.warehouse_main ) self.assertEqual(new_inventory.quantity, 100) self.assertEqual(new_inventory.num_of_rolls, 1) # 验证调用了相关函数 mock_find.assert_called_once_with( product_id=self.product_fabric_a.id, warehouse_id=self.warehouse_main.id ) def test_complete_inbound_record_existing_inventory(self): """测试完成入库记录（更新现有库存）""" existing_inventory = models.Inventory.objects.create( merchant=self.merchant, product=self.product_fabric_a, warehouse=self.warehouse_main, quantity=50, num_of_rolls=5, ) with patch('stock.services.create_stock_snapshot') as mock_snapshot: mock_snapshot.return_value = MagicMock() result = services.make_stock_change_completed(self.stock_change_in) self.assertTrue(result) existing_inventory.refresh_from_db() self.assertEqual(existing_inventory.quantity, 150) self.assertEqual(existing_inventory.num_of_rolls, 6) self.stock_change_in.refresh_from_db() self.assertTrue(self.stock_change_in.is_finished) class StockOffsetServiceTestCase(TestCase): """测试库存红冲逻辑""" def setUp(self): self.merchant = Merchant.objects.create( name='红冲商户', type=MerchantTypeEnum.STORE, ) self.warehouse = WareHouse.objects.create( merchant=self.merchant, name='红冲仓', mode=WareHouseModeEnum.RESTRICT_IN, ) category = ProductCategory.objects.create( merchant=self.merchant, name='红冲品类', product_prefix='RC', ) self.product = Product.objects.create( merchant=self.merchant, category=category, name='红冲产品', human_id='RC-001', unit=ProductUnitEnum.METER, ) self.stock_service = services.StockFlowService(merchant=self.merchant, created_by=None) self.source_record, _, _ = services.create_stock_change_record_with_details( merchant=self.merchant, created_by=None, type=models.StockChangeTypeEnum.ADD, warehouse_id=self.warehouse.id, source_type=models.StockChangeSourceEnum.PURCHASE, source_id=101, products=[{'product': self.product.id, 'quantity': [Decimal('10')]}], ) services.make_stock_change_completed(self.source_record) def test_offset_stock_change_creates_reverse_record(self): offset_record, details, detail_count = self.stock_service.offset_stock_change( source_record_id=self.source_record.id, reason='测试红冲', ) self.assertEqual(offset_record.type, models.StockChangeTypeEnum.REMOVE) self.assertEqual(offset_record.source_type, models.StockChangeSourceEnum.OFFSET) self.assertEqual(detail_count, 1) self.assertEqual(len(details), 1) inventory = models.Inventory.objects.get(product=self.product, warehouse=self.warehouse) self.assertEqual(inventory.quantity, 0) self.assertTrue( models.StockSnapshot.objects.filter( stock_change_record=self.source_record, cancelled=True, offset_id__isnull=False, ).exists() ) def test_offset_cannot_repeat(self): self.stock_service.offset_stock_change( source_record_id=self.source_record.id, reason='第一次红冲', ) with self.assertRaises(ValueError): self.stock_service.offset_stock_change( source_record_id=self.source_record.id, reason='再次红冲', ) def test_offset_requires_finished_record(self): record, _, _ = services.create_stock_change_record_with_details( merchant=self.merchant, created_by=None, type=models.StockChangeTypeEnum.ADD, warehouse_id=self.warehouse.id, source_type=models.StockChangeSourceEnum.PURCHASE, source_id=202, products=[{'product': self.product.id, 'quantity': [Decimal('8')]}], ) with self.assertRaises(ValueError): self.stock_service.offset_stock_change( source_record_id=record.id, reason='未完成记录红冲', ) def test_offset_checks_merchant(self): other_merchant = Merchant.objects.create( name='外部商户', type=MerchantTypeEnum.STORE, ) other_service = services.StockFlowService(merchant=other_merchant, created_by=None) with self.assertRaises(ValueError): other_service.offset_stock_change( source_record_id=self.source_record.id, reason='跨商户红冲', ) class CreateStockChangeRecordWithDetailsTestCase(StockServicesTestCase): """测试 create_stock_change_record_with_details 服务""" def setUp(self): super().setUp() self.products_payload = [ {'product': self.product_fabric_a.id, 'quantity': [Decimal('10.00'), Decimal('20.50')]}, {'product': self.product_fabric_b.id, 'quantity': [Decimal('5.00')]} ] def test_create_record_success(self): """创建库存变动记录并返回所有明细""" record, details, created_count = services.create_stock_change_record_with_details( merchant=self.merchant, created_by=None, type=models.StockChangeTypeEnum.ADD, warehouse_id=self.warehouse_main.id, source_type=models.StockChangeSourceEnum.PURCHASE, source_id=self.purchase_source_id, products=self.products_payload, ) self.assertEqual(record.details.count(), 3) self.assertEqual(len(details), 3) self.assertEqual(created_count, 3) self.assertEqual(record.merchant, self.merchant) self.assertEqual(record.warehouse, self.warehouse_main) self.assertCountEqual( [Decimal('10.00'), Decimal('20.50'), Decimal('5.00')], [detail.quantity for detail in details] ) def test_auto_complete_triggers_completion(self): """商户开启自动确认时会调用完成逻辑""" self.merchant.auto_complete_stock_change = True self.merchant.save() with patch('stock.services.make_stock_change_completed') as mock_complete: services.create_stock_change_record_with_details( merchant=self.merchant, created_by=None, type=models.StockChangeTypeEnum.ADD, warehouse_id=self.warehouse_main.id, source_type=models.StockChangeSourceEnum.PURCHASE, source_id=self.purchase_source_id, products=self.products_payload, ) mock_complete.assert_called_once() def test_invalid_product_raises_error(self): """无效的产品ID会触发 ValueError""" products = [{'product': 99999, 'quantity': [Decimal('1.00')]}] with self.assertRaises(ValueError): services.create_stock_change_record_with_details( merchant=self.merchant, created_by=None, type=models.StockChangeTypeEnum.ADD, warehouse_id=self.warehouse_main.id, source_type=models.StockChangeSourceEnum.PURCHASE, source_id=self.purchase_source_id, products=products, ) def test_strict_mode_requires_restrict_in(self): """严谨模式仅允许在严进宽出仓库中使用""" self.warehouse_main.mode = WareHouseModeEnum.UNRESTRICTED self.warehouse_main.save() with self.assertRaises(ValueError) as ctx: services.create_stock_change_record_with_details( merchant=self.merchant, created_by=None, type=models.StockChangeTypeEnum.ADD, warehouse_id=self.warehouse_main.id, source_type=models.StockChangeSourceEnum.PURCHASE, source_id=self.purchase_source_id, products=[{'product': self.product_fabric_a.id, 'quantity': [Decimal('10.00')]}], ) self.assertIn('宽进宽出', str(ctx.exception)) def test_restrict_in_out_incoming_behaves_like_strict(self): """严进严出模式的入库沿用严谨逻辑""" self.warehouse_main.mode = WareHouseModeEnum.RESTRICT_IN_OUT self.warehouse_main.save() record, details, created_count = services.create_stock_change_record_with_details( merchant=self.merchant, created_by=None, type=models.StockChangeTypeEnum.ADD, warehouse_id=self.warehouse_main.id, source_type=models.StockChangeSourceEnum.PURCHASE, source_id=self.purchase_source_id, products=[{'product': self.product_fabric_a.id, 'quantity': [Decimal('5.00'), Decimal('6.00')]}], ) self.assertEqual(created_count, 2) self.assertEqual(record.details.count(), 2) def test_restrict_in_out_outgoing_consumes_detail(self): """严进严出模式的出库需要消耗入库明细""" self.warehouse_main.mode = WareHouseModeEnum.RESTRICT_IN_OUT self.warehouse_main.save() products = [{ 'product': self.product_fabric_a.id, 'consume_with': [self.stock_detail_in.id], }] record, details, created_count = services.create_stock_change_record_with_details( merchant=self.merchant, created_by=None, type=models.StockChangeTypeEnum.REMOVE, warehouse_id=self.warehouse_main.id, source_type=models.StockChangeSourceEnum.SALES, source_id=123, products=products, ) self.assertEqual(created_count, 1) detail = details[0] self.stock_detail_in.refresh_from_db() self.assertTrue(self.stock_detail_in.is_consumed) self.assertEqual(detail.consume_with_id, self.stock_detail_in.id) self.assertEqual(detail.quantity, self.stock_detail_in.quantity) def test_restrict_in_out_outgoing_requires_consume_ids(self): self.warehouse_main.mode = WareHouseModeEnum.RESTRICT_IN_OUT self.warehouse_main.save() with self.assertRaises(ValueError): services.create_stock_change_record_with_details( merchant=self.merchant, created_by=None, type=models.StockChangeTypeEnum.REMOVE, warehouse_id=self.warehouse_main.id, source_type=models.StockChangeSourceEnum.SALES, source_id=123, products=[{'product': self.product_fabric_a.id}], ) def test_restrict_in_out_outgoing_rejects_consumed_detail(self): self.warehouse_main.mode = WareHouseModeEnum.RESTRICT_IN_OUT self.warehouse_main.save() products = [{ 'product': self.product_fabric_a.id, 'consume_with': [self.stock_detail_in.id], }] services.create_stock_change_record_with_details( merchant=self.merchant, created_by=None, type=models.StockChangeTypeEnum.REMOVE, warehouse_id=self.warehouse_main.id, source_type=models.StockChangeSourceEnum.SALES, source_id=123, products=products, ) with self.assertRaises(ValueError): services.create_stock_change_record_with_details( merchant=self.merchant, created_by=None, type=models.StockChangeTypeEnum.REMOVE, warehouse_id=self.warehouse_main.id, source_type=models.StockChangeSourceEnum.SALES, source_id=124, products=products, ) def test_restrict_in_out_outgoing_rejects_other_warehouse_detail(self): self.warehouse_main.mode = WareHouseModeEnum.RESTRICT_IN_OUT self.warehouse_main.save() other_record = models.StockChangeRecord.objects.create( merchant=self.merchant, type=models.StockChangeTypeEnum.ADD, source_type=models.StockChangeSourceEnum.PURCHASE, warehouse=self.warehouse_backup, ) other_detail = models.StockChangeDetail.objects.create( merchant=self.merchant, stock_change_record=other_record, product=self.product_fabric_a, quantity=Decimal('8.00'), unit=self.unit_meter, ) products = [{ 'product': self.product_fabric_a.id, 'consume_with': [other_detail.id], }] with self.assertRaises(ValueError): services.create_stock_change_record_with_details( merchant=self.merchant, created_by=None, type=models.StockChangeTypeEnum.REMOVE, warehouse_id=self.warehouse_main.id, source_type=models.StockChangeSourceEnum.SALES, source_id=125, products=products, ) class CreateStockChangeRecordRelaxedTestCase(StockServicesTestCase): """测试宽松模式的库存变动创建""" def setUp(self): super().setUp() self.warehouse_main.mode = WareHouseModeEnum.UNRESTRICTED self.warehouse_main.save() def test_relaxed_even_split(self): products = [ { 'product': self.product_fabric_a.id, 'quantity': {'value': Decimal('100.00'), 'unit_count': Decimal('25.00')}, } ] record, details, created_count = services.create_stock_change_record_relaxed( merchant=self.merchant, created_by=None, type=models.StockChangeTypeEnum.ADD, warehouse_id=self.warehouse_main.id, source_type=models.StockChangeSourceEnum.PURCHASE, source_id=self.purchase_source_id, products=products, ) self.assertEqual(created_count, 4) self.assertEqual(record.details.count(), 4) self.assertListEqual( [detail.quantity for detail in details], [Decimal('25.00')] * 4, ) def test_relaxed_with_remainder(self): products = [ { 'product': self.product_fabric_a.id, 'quantity': {'value': Decimal('105.00'), 'unit_count': Decimal('30.00')}, } ] _, details, created_count = services.create_stock_change_record_relaxed( merchant=self.merchant, created_by=None, type=models.StockChangeTypeEnum.ADD, warehouse_id=self.warehouse_main.id, source_type=models.StockChangeSourceEnum.PURCHASE, source_id=self.purchase_source_id, products=products, ) self.assertEqual(created_count, 4) self.assertListEqual( [detail.quantity for detail in details], [Decimal('30.00'), Decimal('30.00'), Decimal('30.00'), Decimal('15.00')], ) def test_relaxed_invalid_unit_count(self): products = [ { 'product': self.product_fabric_a.id, 'quantity': {'value': Decimal('50.00'), 'unit_count': Decimal('0')}, } ] with self.assertRaises(ValueError): services.create_stock_change_record_relaxed( merchant=self.merchant, created_by=None, type=models.StockChangeTypeEnum.ADD, warehouse_id=self.warehouse_main.id, source_type=models.StockChangeSourceEnum.PURCHASE, source_id=self.purchase_source_id, products=products, ) def test_relaxed_auto_complete_triggers_completion(self): self.merchant.auto_complete_stock_change = True self.merchant.save() products = [ { 'product': self.product_fabric_a.id, 'quantity': {'value': Decimal('10.00'), 'unit_count': Decimal('2.00')}, } ] with patch('stock.services.make_stock_change_completed') as mock_complete: services.create_stock_change_record_relaxed( merchant=self.merchant, created_by=None, type=models.StockChangeTypeEnum.ADD, warehouse_id=self.warehouse_main.id, source_type=models.StockChangeSourceEnum.PURCHASE, source_id=self.purchase_source_id, products=products, ) mock_complete.assert_called_once() def test_relaxed_mode_requires_unrestricted(self): self.warehouse_main.mode = WareHouseModeEnum.RESTRICT_IN self.warehouse_main.save() products = [ { 'product': self.product_fabric_a.id, 'quantity': {'value': Decimal('10.00'), 'unit_count': Decimal('2.00')}, } ] with self.assertRaises(ValueError) as ctx: services.create_stock_change_record_relaxed( merchant=self.merchant, created_by=None, type=models.StockChangeTypeEnum.ADD, warehouse_id=self.warehouse_main.id, source_type=models.StockChangeSourceEnum.PURCHASE, source_id=self.purchase_source_id, products=products, ) self.assertIn('严进宽出', str(ctx.exception)) def test_relaxed_mode_not_implemented_for_restrict_in_out(self): self.warehouse_main.mode = WareHouseModeEnum.RESTRICT_IN_OUT self.warehouse_main.save() products = [ { 'product': self.product_fabric_a.id, 'quantity': {'value': Decimal('10.00'), 'unit_count': Decimal('2.00')}, } ] with self.assertRaises(ValueError) as ctx: services.create_stock_change_record_relaxed( merchant=self.merchant, created_by=None, type=models.StockChangeTypeEnum.ADD, warehouse_id=self.warehouse_main.id, source_type=models.StockChangeSourceEnum.PURCHASE, source_id=self.purchase_source_id, products=products, ) self.assertIn('严进严出', str(ctx.exception)) def test_complete_outbound_record(self): """测试完成出库记录""" # 创建现有库存 existing_inventory = models.Inventory.objects.create( merchant=self.merchant, product=self.product_fabric_a, warehouse=self.warehouse_main, quantity=100, num_of_rolls=10 ) with patch('stock.services.create_stock_snapshot') as mock_snapshot: mock_snapshot.return_value = MagicMock() result = services.make_stock_change_completed(self.stock_change_out) self.assertTrue(result) # 检查库存减少 existing_inventory.refresh_from_db() self.assertEqual(existing_inventory.quantity, 50) # 100 - 50 self.assertEqual(existing_inventory.num_of_rolls, 9) # 10 - 1 def test_complete_record_with_multiple_details(self): """测试完成包含多个明细的记录""" # 添加另一个明细 models.StockChangeDetail.objects.create( merchant=self.merchant, stock_change_record=self.stock_change_in, product=self.product_fabric_b, quantity=Decimal('200'), unit=self.unit_kg ) with patch('stock.services.create_stock_snapshot') as mock_snapshot: mock_snapshot.return_value = MagicMock() result = services.make_stock_change_completed(self.stock_change_in) self.assertTrue(result) # 检查为两个产品都创建了库存记录 inventory_a = models.Inventory.objects.get( product=self.product_fabric_a, warehouse=self.warehouse_main ) inventory_b = models.Inventory.objects.get( product=self.product_fabric_b, warehouse=self.warehouse_main ) self.assertEqual(inventory_a.quantity, 100) self.assertEqual(inventory_b.quantity, 200) # 验证快照创建了两次 self.assertEqual(mock_snapshot.call_count, 2) @patch.object(services.logger, 'info') def test_logging_during_completion(self, mock_logger): """测试完成过程中的日志记录""" services.make_stock_change_completed(self.stock_change_in) # 验证日志调用 self.assertGreater(mock_logger.call_count, 0) # 检查最后一条日志消息 last_call_args = mock_logger.call_args_list[-1][0] self.assertIn('标记为已完成', last_call_args[0]) class StockServicesIntegrationTestCase(StockServicesTestCase): """集成测试：测试服务函数之间的协作""" def test_complete_workflow(self): """测试完整的库存变动工作流""" # 1. 确保开始时没有库存 self.assertEqual( models.Inventory.objects.filter( product=self.product_fabric_a, warehouse=self.warehouse_main ).count(), 0 ) # 2. 完成入库操作 result = services.make_stock_change_completed(self.stock_change_in) self.assertTrue(result) # 3. 验证库存被创建 inventory = services.find_inventory( self.product_fabric_a.id, self.warehouse_main.id ) self.assertIsNotNone(inventory) self.assertEqual(inventory.quantity, 100) # 4. 验证快照被创建 snapshots = models.StockSnapshot.objects.filter( stock_change_record=self.stock_change_in ) self.assertEqual(snapshots.count(), 1) snapshot = snapshots.first() self.assertEqual(snapshot.delta, 100) self.assertEqual(snapshot.quantity_before, 0) self.assertEqual(snapshot.quantity_after, 100) # 5. 完成出库操作 result = services.make_stock_change_completed(self.stock_change_out) self.assertTrue(result) # 6. 验证库存被更新 inventory.refresh_from_db() self.assertEqual(inventory.quantity, 50) # 100 - 50 # 7. 验证出库快照 out_snapshots = models.StockSnapshot.objects.filter( stock_change_record=self.stock_change_out ) self.assertEqual(out_snapshots.count(), 1) out_snapshot = out_snapshots.first() self.assertEqual(out_snapshot.delta, -50) self.assertEqual(out_snapshot.quantity_before, 100) self.assertEqual(out_snapshot.quantity_after, 50) class StockFlowServiceTestCase(TestCase): """StockFlowService 统一出入库服务测试""" def setUp(self): self.merchant = Merchant.objects.create( name='统一服务商户', type=MerchantTypeEnum.FACTORY, ) self.category = ProductCategory.objects.create( merchant=self.merchant, name='面料', product_prefix='FAB', ) self.product = Product.objects.create( merchant=self.merchant, category=self.category, name='测试面料', human_id='FAB-0001', unit=ProductUnitEnum.METER, ) self.strict_warehouse = WareHouse.objects.create( merchant=self.merchant, name='严谨仓', mode=WareHouseModeEnum.RESTRICT_IN, ) self.relaxed_warehouse = WareHouse.objects.create( merchant=self.merchant, name='宽松仓', mode=WareHouseModeEnum.UNRESTRICTED, ) self.restrict_in_out = WareHouse.objects.create( merchant=self.merchant, name='严进严出仓', mode=WareHouseModeEnum.RESTRICT_IN_OUT, ) self.service = services.StockFlowService(merchant=self.merchant, created_by=None) def test_stock_in_strict_mode_uses_quantities(self): record, details, count = self.service.stock_in( warehouse_id=self.strict_warehouse.id, source_type=models.StockChangeSourceEnum.PURCHASE, source_id=10, items=[{'product_id': self.product.id, 'quantities': ['5', '3']}], ) self.assertEqual(count, 2) self.assertEqual(record.warehouse_id, self.strict_warehouse.id) self.assertEqual(len(details), 2) def test_stock_in_relaxed_mode_splits_value(self): _, details, count = self.service.stock_in( warehouse_id=self.relaxed_warehouse.id, source_type=models.StockChangeSourceEnum.PURCHASE, source_id=11, items=[{'product_id': self.product.id, 'value': '9', 'num_of_rolls': 4}], ) self.assertEqual(count, 4) self.assertEqual(len(details), 4) def test_stock_out_restrict_mode_consumes_details(self): _, inbound_details, _ = self.service.stock_in( warehouse_id=self.restrict_in_out.id, source_type=models.StockChangeSourceEnum.PURCHASE, source_id=12, items=[{'product_id': self.product.id, 'quantities': ['6']}], ) detail_id = inbound_details[0].id _, outbound_details, _ = self.service.stock_out( warehouse_id=self.restrict_in_out.id, source_type=models.StockChangeSourceEnum.SALES, source_id=13, items=[{'product_id': self.product.id, 'consume_detail_ids': [detail_id]}], ) inbound_details[0].refresh_from_db() self.assertTrue(inbound_details[0].is_consumed) self.assertEqual(outbound_details[0].consume_with_id, detail_id) # 禁用测试期间的日志输出 logging.disable(logging.CRITICAL)