forked from erp-dev/erp
568 lines
21 KiB
Python
568 lines
21 KiB
Python
from django.core.exceptions import ValidationError
|
|
from django.test import TestCase
|
|
from django.utils import timezone
|
|
from decimal import Decimal
|
|
from unittest.mock import patch, MagicMock
|
|
import logging
|
|
|
|
from basic_info.models import Product, WareHouse, Supplier, ProductUnitEnum, ProductCategory, Merchant, MerchantTypeEnum
|
|
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.supplier = Supplier.objects.create(
|
|
merchant=cls.merchant,
|
|
name="测试供应商",
|
|
contact="张三",
|
|
mobile="13800138000"
|
|
)
|
|
|
|
# 创建产品
|
|
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_order = models.PurchaseOrder.objects.create(
|
|
merchant=self.merchant,
|
|
supplier=self.supplier,
|
|
order_date=timezone.now().date(),
|
|
total_amount=Decimal('1000.00'),
|
|
remarks="测试采购单"
|
|
)
|
|
|
|
# 创建库存变动记录（入库）
|
|
self.stock_change_in = models.StockChangeRecord.objects.create(
|
|
merchant=self.merchant,
|
|
type=models.StockChangeTypeEnum.ADD,
|
|
source_type=models.StockChangeSourceEnum.PURCHASE,
|
|
source_id=self.purchase_order.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) # 50 + 100
|
|
self.assertEqual(existing_inventory.num_of_rolls, 6) # 5 + 1
|
|
|
|
# 验证记录状态
|
|
self.stock_change_in.refresh_from_db()
|
|
self.assertTrue(self.stock_change_in.is_finished)
|
|
|
|
|
|
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_order.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_order.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_order.id,
|
|
products=products,
|
|
)
|
|
|
|
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)
|
|
|
|
|
|
# 禁用测试期间的日志输出
|
|
logging.disable(logging.CRITICAL)
|