zuowei/erpnew
1
0
Fork 0
forked from erp-dev/erp
Code Pull Requests Activity

fix: rebuild stock change record

Browse Source
This commit is contained in:
colaftc
2025-11-05 16:57:54 +08:00
parent d88bf367f8
commit a759030cdc
16 changed files with 1116 additions and 66 deletions
Download Patch File Download Diff File Expand all files Collapse all files

36
stock/admin.py
View File

@@ -1,5 +1,6 @@
from django.contrib import admin
from django.contrib.admin import action
from django.utils.safestring import mark_safe
from . import models, services
@@ -58,29 +59,43 @@ class StockSnapshotAdmin(admin.ModelAdmin):
def has_delete_permission(self, request, obj = ...):
# 禁止通过admin界面删除库存快照
return False
return True
def get_readonly_fields(self, request, obj = ...):
# 所有字段均为只读
return [field.name for field in self.model._meta.fields]
@admin.register(models.StockChangeDetail)
class StockChangeDetailAdmin(admin.ModelAdmin):
list_display = (
'id',
'stock_change_record',
'product',
'quantity',
'unit',
)
search_fields = ('stock_change_record__id', 'product__name')
list_filter = ('stock_change_record', 'stock_change_record__type')
ordering = ('-created_at',)
@admin.register(models.StockChangeRecord)
class StockChangeRecordAdmin(admin.ModelAdmin):
list_display = (
'id',
'type',
'product',
'warehouse',
'quantity',
'details',
'source_type',
'source_id',
'num_of_rolls',
'is_finished',
'finished_at',
'created_by',
'created_at',
)
search_fields = ('product__name', 'warehouse__name', 'source_id')
search_fields = ('warehouse__name', 'source_id')
list_filter = ('type', 'source_type', 'type', 'is_finished', 'created_at')
ordering = ('-created_at', '-updated_at')
readonly_fields = ('created_by', 'is_finished', 'finished_at')
@@ -103,4 +118,17 @@ class StockChangeRecordAdmin(admin.ModelAdmin):
record = queryset.first()
services.make_stock_change_completed(record)
@admin.display(description='匹数')
def num_of_rolls(self, obj: models.StockChangeRecord):
return obj.num_of_rolls
@admin.display(description='明细概览')
def details(self, obj):
details = obj.details.all()
detail_list = [
f'{detail.product.name}, {detail.quantity}/{detail.get_unit_display()}'
for detail in details
]
return mark_safe('<br>'.join(detail_list))
actions = [make_selected_stock_changes_completed]

43
stock/migrations/0031_remove_stockchangerecord_num_of_rolls_and_more.py Normal file
View File

@@ -0,0 +1,43 @@
# Generated by Django 5.2.7 on 2025-11-05 05:34
import django.db.models.deletion
from django.db import migrations, models
class Migration(migrations.Migration):
dependencies = [
('basic_info', '0017_product_image'),
('stock', '0030_alter_stocksnapshot_delta'),
]
operations = [
migrations.RemoveField(
model_name='stockchangerecord',
name='num_of_rolls',
),
migrations.RemoveField(
model_name='stockchangerecord',
name='product',
),
migrations.RemoveField(
model_name='stockchangerecord',
name='quantity',
),
migrations.CreateModel(
name='StockChangeDetail',
fields=[
('created_at', models.DateTimeField(auto_now_add=True, verbose_name='创建时间')),
('updated_at', models.DateTimeField(auto_now=True, verbose_name='更新时间')),
('id', models.BigAutoField(primary_key=True, serialize=False)),
('quantity', models.DecimalField(decimal_places=2, max_digits=10, verbose_name='数量')),
('unit', models.IntegerField(choices=[(1, '米'), (2, '码'), (3, '公斤')], verbose_name='单位')),
('product', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, related_name='stock_change_details', to='basic_info.product', verbose_name='产品')),
('stock_change_record', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='details', to='stock.stockchangerecord', verbose_name='库存变动记录')),
],
options={
'verbose_name': '库存变动明细',
'verbose_name_plural': '库存变动明细',
},
),
]

17
stock/migrations/0032_remove_stockchangerecord_unit.py Normal file
View File

@@ -0,0 +1,17 @@
# Generated by Django 5.2.7 on 2025-11-05 06:50
from django.db import migrations
class Migration(migrations.Migration):
dependencies = [
('stock', '0031_remove_stockchangerecord_num_of_rolls_and_more'),
]
operations = [
migrations.RemoveField(
model_name='stockchangerecord',
name='unit',
),
]

68
stock/models.py
View File

@@ -2,6 +2,7 @@ from django.db import models
from django.core.exceptions import ValidationError
from flower.common import ModelBase
from basic_info import models as basic_info_models
from functools import reduce
import decimal
@@ -48,12 +49,6 @@ class StockChangeRecord(ModelBase):
id = models.BigAutoField(primary_key=True)
type = models.IntegerField(choices=StockChangeTypeEnum.choices, verbose_name='变动类型')
product = models.ForeignKey(
basic_info_models.Product,
on_delete=models.PROTECT,
related_name='stock_change_records',
verbose_name='产品',
)
warehouse = models.ForeignKey(
basic_info_models.WareHouse,
on_delete=models.PROTECT,
@@ -64,14 +59,6 @@ class StockChangeRecord(ModelBase):
choices=StockChangeSourceEnum.choices,
verbose_name='变动来源',
)
unit = models.IntegerField(
verbose_name='单位',
choices=basic_info_models.ProductUnitEnum.choices,
)
num_of_rolls = models.IntegerField(
default=1,
verbose_name='匹数',
)
created_by = models.ForeignKey(
'auth.User',
on_delete=models.PROTECT,
@@ -79,14 +66,13 @@ class StockChangeRecord(ModelBase):
blank=True,
verbose_name='操作人',
)
quantity = models.DecimalField(max_digits=10, decimal_places=2, verbose_name='数量')
source_id = models.BigIntegerField(null=True, blank=True, verbose_name='来源单据ID')
is_finished = models.BooleanField(default=False, verbose_name='完成出入库')
finished_at = models.DateTimeField(null=True, blank=True, verbose_name='完成时间')
remarks = models.TextField(blank=True, null=True, verbose_name='备注')
def __str__(self):
return f'出入库记录 {self.id} - 仓库：{self.warehouse.name} - 产品: {self.product.name}'
return f'出入库记录 {self.id} - 仓库：{self.warehouse.name}'
class Meta:
verbose_name = '出入库记录'
@@ -115,17 +101,57 @@ class StockChangeRecord(ModelBase):
return self.type == StockChangeTypeEnum.REMOVE
@property
def delta(self) -> decimal.Decimal:
"""计算库存变动的数量增减"""
def num_of_rolls(self) -> int:
"""获取匹数"""
self.details.count()
return self.details.count()
@property
def delta_num_of_rolls(self) -> int:
"""计算库存变动的匹数增减"""
# 入库为正，出库为负
if self.type == StockChangeTypeEnum.ADD:
return self.quantity
return self.num_of_rolls
elif self.type == StockChangeTypeEnum.REMOVE:
return -self.quantity
return self.num_of_rolls * -1
else:
raise ValueError("未知的库存变动类型")
class StockChangeDetail(ModelBase):
"""库存变动明细模型"""
id = models.BigAutoField(primary_key=True)
product = models.ForeignKey(
basic_info_models.Product,
on_delete=models.PROTECT,
related_name='stock_change_details',
verbose_name='产品',
)
unit = models.IntegerField(
verbose_name='单位',
choices=basic_info_models.ProductUnitEnum.choices,
)
stock_change_record = models.ForeignKey(
StockChangeRecord,
on_delete=models.CASCADE,
related_name='details',
verbose_name='库存变动记录',
)
quantity = models.DecimalField(max_digits=10, decimal_places=2, verbose_name='数量')
unit = models.IntegerField(
verbose_name='单位',
choices=basic_info_models.ProductUnitEnum.choices,
)
def __str__(self):
return f'库存变动明细 {self.id} - 记录ID: {self.stock_change_record.id}'
class Meta:
verbose_name = '库存变动明细'
verbose_name_plural = '库存变动明细'
class Inventory(ModelBase):
"""库存模型"""
@@ -217,7 +243,7 @@ class StockSnapshot(ModelBase):
)
def __str__(self):
return f'库存快照 {self.id} - 日期: {self.snapshot_date} - 产品: {self.product.name}'
return f'库存快照 {self.id} - 产品: {self.product.name}'
class Meta:
verbose_name = '库存快照'

90
stock/services.py
View File

@@ -19,13 +19,14 @@ def find_inventory(product_id: int, warehouse_id: int) -> models.Inventory | Non
return None
def create_stock_snapshot(stock_change_record: models.StockChangeRecord, inventory: models.Inventory) -> models.StockSnapshot:
def create_stock_snapshot(stock_change_detail: models.StockChangeDetail, inventory: models.Inventory) -> models.StockSnapshot:
"""创建库存快照记录"""
if not stock_change_record or stock_change_record.id is None:
raise ValueError("无效的库存变动记录")
if not stock_change_detail or stock_change_detail.id is None:
raise ValueError("无效的库存变动明细记录")
delta = stock_change_record.delta
delta = stock_change_detail.quantity if stock_change_detail.stock_change_record.is_incoming else -stock_change_detail.quantity
rolls_delta = 1 if stock_change_detail.stock_change_record.is_incoming else -1
# 必须在库存变动发生后创建快照
snapshot = models.StockSnapshot(
@@ -34,9 +35,9 @@ def create_stock_snapshot(stock_change_record: models.StockChangeRecord, invento
delta=delta,
quantity_before=inventory.quantity - delta,
quantity_after=inventory.quantity,
stock_change_record_id=stock_change_record.id,
unit=stock_change_record.unit,
num_of_rolls=stock_change_record.num_of_rolls or 1,
stock_change_record_id=stock_change_detail.stock_change_record.id,
unit=stock_change_detail.unit,
num_of_rolls=inventory.num_of_rolls + rolls_delta,
)
snapshot.save()
logger.info(
@@ -47,7 +48,7 @@ def create_stock_snapshot(stock_change_record: models.StockChangeRecord, invento
def make_stock_change_completed(stock_change_record: models.StockChangeRecord) -> bool:
"""处理库存变动完成后的逻辑"""
"""处理库存变动完成后的逻辑, 实际扣减库存也在这里发生"""
# TODO: 这里可以添加事务处理以确保数据一致性
if not stock_change_record or stock_change_record.id is None:
@@ -57,44 +58,55 @@ def make_stock_change_completed(stock_change_record: models.StockChangeRecord) -
if stock_change_record.is_finished:
return True
inventory_record = find_inventory(
product_id=stock_change_record.product_id,
warehouse_id=stock_change_record.warehouse_id,
)
if not inventory_record:
# TODO: 逻辑更改 = 出入库单现在不记录库存总量了
# 需要根据产品分别统计变动数量（要考虑不同单位）
for detail in stock_change_record.details.all():
detail: models.StockChangeDetail
logger.info(
f'库存记录未找到，产品ID: {stock_change_record.product_id}, '
f'仓库ID: {stock_change_record.warehouse_id}，正在创建新的库存记录。'
f'处理库存变动明细，产品ID: {detail.product_id}, '
f'数量: {detail.quantity}, 单位: {detail.get_unit_display()}'
)
# 如果没有找到库存记录且为入库，则创建一个新的库存记录
inventory_record = models.Inventory(
product_id=stock_change_record.product_id,
inventory_record = find_inventory(
product_id=detail.product_id,
warehouse_id=stock_change_record.warehouse_id,
quantity=stock_change_record.delta,
num_of_rolls=stock_change_record.num_of_rolls or 0,
minimum_quantity=0,
)
logger.info(
f'创建新的库存记录，产品ID: {inventory_record.product_id}, '
f'仓库ID: {inventory_record.warehouse_id}, 数量: {inventory_record.quantity},'
f'匹数: {inventory_record.num_of_rolls}'
)
else:
# 更新现有库存记录
inventory_record.quantity += stock_change_record.delta
if stock_change_record.num_of_rolls:
inventory_record.num_of_rolls += stock_change_record.num_of_rolls
logger.info(
f'更新库存记录 {inventory_record.id}，新数量: {inventory_record.quantity}, '
f'新匹数: {inventory_record.num_of_rolls}'
)
inventory_record.save()
create_stock_snapshot(stock_change_record, inventory_record)
# 确定方向
positive = 1 if stock_change_record.is_incoming else -1
if inventory_record:
# 更新现有库存记录
inventory_record.quantity += detail.quantity * positive
inventory_record.num_of_rolls += positive
logger.info(
f'更新库存记录 {inventory_record.id}，新数量: {inventory_record.quantity}, '
f'新匹数: {inventory_record.num_of_rolls}'
)
inventory_record.save()
else:
# 创建新库存记录
inventory_record = models.Inventory(
product_id=detail.product_id,
warehouse_id=stock_change_record.warehouse_id,
quantity=detail.quantity * positive,
num_of_rolls=1 if stock_change_record.is_incoming else -1,
)
inventory_record.save()
logger.info(
f'创建新库存记录 {inventory_record.id}，数量: {inventory_record.quantity}, '
f'匹数: {inventory_record.num_of_rolls}'
)
# 无论库存记录原本是否存在都创建库存快照
create_stock_snapshot(detail, inventory_record)
# 所有明细处理完毕，标记库存变动记录为已完成
stock_change_record.is_finished = True
stock_change_record.finished_at = timezone.now()
stock_change_record.save()

426
stock/tests.py
View File

@@ -1,3 +1,427 @@
from django.test import TestCase
from django.utils import timezone
from decimal import Decimal
from unittest.mock import patch, MagicMock
import logging
# Create your tests here.
from basic_info.models import Product, WareHouse, Supplier, ProductUnitEnum, ProductCategory
from . import models, services
class StockServicesTestCase(TestCase):
"""库存服务层测试"""
@classmethod
def setUpTestData(cls):
"""设置测试数据（类级别，所有测试方法共享）"""
# 产品单位（使用枚举值）
cls.unit_meter = ProductUnitEnum.METER
cls.unit_kg = ProductUnitEnum.KG
# 创建产品类别
cls.fabric_category = ProductCategory.objects.create(
name="布料类",
description="各种布料产品",
product_prefix="FAB"
)
# 创建仓库
cls.warehouse_main = WareHouse.objects.create(
name="主仓库",
location="主仓库地址"
)
cls.warehouse_backup = WareHouse.objects.create(
name="备用仓库",
location="备用仓库地址"
)
# 创建供应商
cls.supplier = Supplier.objects.create(
name="测试供应商",
contact="张三",
mobile="13800138000"
)
# 创建产品
cls.product_fabric_a = Product.objects.create(
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(
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(
supplier=self.supplier,
order_date=timezone.now().date(),
total_amount=Decimal('1000.00'),
remarks="测试采购单"
)
# 创建库存变动记录（入库）
self.stock_change_in = models.StockChangeRecord.objects.create(
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(
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(
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(
stock_change_record=self.stock_change_out,
product=self.product_fabric_a,
quantity=Decimal('50'),
unit=self.unit_meter
)
class FindInventoryTestCase(StockServicesTestCase):
"""测试 find_inventory 函数"""
def test_find_existing_inventory(self):
"""测试查找已存在的库存记录"""
# 先创建一个库存记录
inventory = models.Inventory.objects.create(
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(
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(
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(
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(
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
)
mock_snapshot.assert_called_once()
def test_complete_inbound_record_existing_inventory(self):
"""测试完成入库记录（更新现有库存）"""
# 创建现有库存
existing_inventory = models.Inventory.objects.create(
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)
def test_complete_outbound_record(self):
"""测试完成出库记录"""
# 创建现有库存
existing_inventory = models.Inventory.objects.create(
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(
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)