forked from erp-dev/erp
feat: allocation record for pre sales order
This commit is contained in:
185
business/tests/test_allocation_services.py
Normal file
185
business/tests/test_allocation_services.py
Normal file
@@ -0,0 +1,185 @@
|
||||
from decimal import Decimal
|
||||
|
||||
from django.contrib.auth import get_user_model
|
||||
from django.test import TestCase, override_settings
|
||||
|
||||
from basic_info.models import (
|
||||
Merchant,
|
||||
MerchantTypeEnum,
|
||||
Customer,
|
||||
WareHouse,
|
||||
WareHouseModeEnum,
|
||||
ProductCategory,
|
||||
Product,
|
||||
ProductUnitEnum,
|
||||
Employee,
|
||||
EmployeeStatusEnum,
|
||||
)
|
||||
from business import models as business_models
|
||||
from business import allocation_services
|
||||
from stock import models as stock_models
|
||||
|
||||
|
||||
@override_settings(
|
||||
CELERY_TASK_ALWAYS_EAGER=True,
|
||||
CELERY_TASK_EAGER_PROPAGATES=True,
|
||||
)
|
||||
class AllocationServicesTestCase(TestCase):
|
||||
def setUp(self):
|
||||
self.merchant = Merchant.objects.create(name='配货商户', type=MerchantTypeEnum.FACTORY)
|
||||
self.customer = Customer.objects.create(
|
||||
merchant=self.merchant,
|
||||
name='配货客户',
|
||||
created_by=None,
|
||||
)
|
||||
self.warehouse = WareHouse.objects.create(
|
||||
merchant=self.merchant,
|
||||
name='配货仓库',
|
||||
mode=WareHouseModeEnum.UNRESTRICTED,
|
||||
)
|
||||
category = ProductCategory.objects.create(
|
||||
merchant=self.merchant,
|
||||
name='配货品类',
|
||||
product_prefix='ALC',
|
||||
)
|
||||
self.product = Product.objects.create(
|
||||
merchant=self.merchant,
|
||||
category=category,
|
||||
name='配货产品',
|
||||
human_id='ALC-001',
|
||||
unit=ProductUnitEnum.METER,
|
||||
)
|
||||
self.other_product = Product.objects.create(
|
||||
merchant=self.merchant,
|
||||
category=category,
|
||||
name='其他产品',
|
||||
human_id='ALC-002',
|
||||
unit=ProductUnitEnum.METER,
|
||||
)
|
||||
|
||||
User = get_user_model()
|
||||
self.user = User.objects.create_user(username='allocation_user', password='pass123')
|
||||
self.employee = Employee.objects.create(
|
||||
merchant=self.merchant,
|
||||
sys_user=self.user,
|
||||
name='配货员',
|
||||
status=EmployeeStatusEnum.ACTIVE,
|
||||
)
|
||||
|
||||
self.pre_sales_order = business_models.PreSalesOrder.objects.create(
|
||||
merchant=self.merchant,
|
||||
customer=self.customer,
|
||||
warehouse=self.warehouse,
|
||||
created_by=self.user,
|
||||
operator=self.employee,
|
||||
kind=business_models.SalesOrderKindEnum.WHOLESALE,
|
||||
remarks='预销售单备注',
|
||||
)
|
||||
self.item = business_models.PreSalesOrderItem.objects.create(
|
||||
pre_sales_order=self.pre_sales_order,
|
||||
product_id=self.product.id,
|
||||
product_name=self.product.name,
|
||||
quantity=Decimal('12.5'),
|
||||
unit='米',
|
||||
remarks='明细备注',
|
||||
)
|
||||
|
||||
self.stock_record = stock_models.StockChangeRecord.objects.create(
|
||||
merchant=self.merchant,
|
||||
type=stock_models.StockChangeTypeEnum.ADD,
|
||||
warehouse=self.warehouse,
|
||||
source_type=stock_models.StockChangeSourceEnum.PURCHASE,
|
||||
created_by=self.user,
|
||||
)
|
||||
self.stock_detail = stock_models.StockChangeDetail.objects.create(
|
||||
merchant=self.merchant,
|
||||
product=self.product,
|
||||
unit=ProductUnitEnum.METER,
|
||||
stock_change_record=self.stock_record,
|
||||
quantity='12.5',
|
||||
)
|
||||
|
||||
def test_validate_stock_details_missing(self):
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
allocation_services.validate_stock_details(
|
||||
merchant=self.merchant,
|
||||
stock_ids=[999999],
|
||||
)
|
||||
self.assertIn('库存明细不存在', str(ctx.exception))
|
||||
|
||||
def test_validate_stock_details_consumed(self):
|
||||
self.stock_detail.is_consumed = True
|
||||
self.stock_detail.save(update_fields=['is_consumed'])
|
||||
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
allocation_services.validate_stock_details(
|
||||
merchant=self.merchant,
|
||||
stock_ids=[self.stock_detail.id],
|
||||
)
|
||||
self.assertIn('库存明细已被消费', str(ctx.exception))
|
||||
|
||||
def test_validate_stock_details_frozen(self):
|
||||
stock_models.StockFreeze.objects.create(
|
||||
merchant=self.merchant,
|
||||
product=self.product,
|
||||
warehouse=self.warehouse,
|
||||
stock_detail=self.stock_detail,
|
||||
quantity=Decimal('1'),
|
||||
unit=ProductUnitEnum.METER,
|
||||
status=stock_models.StockFreezeStatusEnum.FROZEN,
|
||||
frozen_by=self.user,
|
||||
)
|
||||
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
allocation_services.validate_stock_details(
|
||||
merchant=self.merchant,
|
||||
stock_ids=[self.stock_detail.id],
|
||||
)
|
||||
self.assertIn('库存明细已被冻结', str(ctx.exception))
|
||||
|
||||
def test_create_allocation_record_product_mismatch(self):
|
||||
other_detail = stock_models.StockChangeDetail.objects.create(
|
||||
merchant=self.merchant,
|
||||
product=self.other_product,
|
||||
unit=ProductUnitEnum.METER,
|
||||
stock_change_record=self.stock_record,
|
||||
quantity='3.0',
|
||||
)
|
||||
|
||||
with self.assertRaises(ValueError) as ctx:
|
||||
allocation_services.create_allocation_record(
|
||||
item=self.item,
|
||||
stock_ids=[other_detail.id],
|
||||
quantity='3.0',
|
||||
unit='米',
|
||||
scanned_by=self.employee,
|
||||
)
|
||||
self.assertIn('库存明细产品不匹配', str(ctx.exception))
|
||||
|
||||
def test_create_allocation_record_success(self):
|
||||
record = allocation_services.create_allocation_record(
|
||||
item=self.item,
|
||||
stock_ids=[self.stock_detail.id],
|
||||
quantity='5.5',
|
||||
unit='米',
|
||||
scanned_by=self.employee,
|
||||
remarks='扫码录入',
|
||||
)
|
||||
self.assertEqual(record.pre_sales_order_item_id, self.item.id)
|
||||
self.assertEqual(record.stock_ids, str(self.stock_detail.id))
|
||||
self.assertEqual(record.stock_id_list, [self.stock_detail.id])
|
||||
self.assertEqual(record.status, business_models.AllocationRecordStatusEnum.ACTIVE)
|
||||
freeze_qs = stock_models.StockFreeze.objects.filter(stock_detail=self.stock_detail)
|
||||
self.assertEqual(freeze_qs.count(), 1)
|
||||
self.assertEqual(freeze_qs.first().status, stock_models.StockFreezeStatusEnum.FROZEN)
|
||||
|
||||
def test_cancel_allocation_record(self):
|
||||
record = allocation_services.create_allocation_record(
|
||||
item=self.item,
|
||||
stock_ids=[self.stock_detail.id],
|
||||
quantity='2.0',
|
||||
unit='米',
|
||||
scanned_by=self.employee,
|
||||
)
|
||||
updated = allocation_services.cancel_allocation_record(record=record)
|
||||
self.assertEqual(updated.status, business_models.AllocationRecordStatusEnum.CANCELLED)
|
||||
Reference in New Issue
Block a user