forked from erp-dev/erp
270 lines
10 KiB
Python
270 lines
10 KiB
Python
from django.test import TestCase
|
|
from django.contrib.auth import get_user_model
|
|
from django.test import TransactionTestCase
|
|
from django.db import transaction
|
|
|
|
from basic_info import models as basic_models
|
|
from business import models as business_models
|
|
from business import pre_order_services
|
|
from business import allocation_services
|
|
from stock import models as stock_models
|
|
|
|
from .fixtures import create_sales_fixtures
|
|
|
|
|
|
class PreSalesOrderServiceTestCase(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()
|
|
|
|
self.user = get_user_model().objects.create_user(username='sales-user', password='pass')
|
|
self.operator.sys_user = self.user
|
|
self.operator.save(update_fields=['sys_user'])
|
|
|
|
def test_create_pre_sales_order_success(self):
|
|
pre_sales_order = pre_order_services.create_pre_sales_order(
|
|
merchant=self.merchant,
|
|
customer_id=self.customer.id,
|
|
warehouse_id=self.warehouse_relaxed.id,
|
|
created_by=self.user,
|
|
operator=self.operator,
|
|
kind=business_models.SalesOrderKindEnum.WHOLESALE,
|
|
items=[
|
|
{
|
|
'product_id': self.product.id,
|
|
'quantity': '12.5',
|
|
'unit': '米',
|
|
'order_quantity': 8,
|
|
'remarks': 'item-remarks',
|
|
}
|
|
],
|
|
remarks='order-remarks',
|
|
)
|
|
|
|
self.assertEqual(pre_sales_order.merchant_id, self.merchant.id)
|
|
self.assertEqual(pre_sales_order.customer_id, self.customer.id)
|
|
self.assertEqual(pre_sales_order.warehouse_id, self.warehouse_relaxed.id)
|
|
self.assertEqual(pre_sales_order.created_by_id, self.user.id)
|
|
self.assertEqual(pre_sales_order.operator_id, self.operator.id)
|
|
self.assertEqual(pre_sales_order.items.count(), 1)
|
|
|
|
item = pre_sales_order.items.first()
|
|
self.assertEqual(item.product_id, self.product.id)
|
|
self.assertEqual(item.unit, '米')
|
|
self.assertEqual(item.order_quantity, 8)
|
|
|
|
def test_create_pre_sales_order_rejects_cross_merchant_customer(self):
|
|
other_merchant = basic_models.Merchant.objects.create(
|
|
name='其他商户',
|
|
type=basic_models.MerchantTypeEnum.FACTORY,
|
|
)
|
|
other_customer = basic_models.Customer.objects.create(
|
|
merchant=other_merchant,
|
|
name='其他客户',
|
|
created_by=None,
|
|
)
|
|
|
|
with self.assertRaises(ValueError):
|
|
pre_order_services.create_pre_sales_order(
|
|
merchant=self.merchant,
|
|
customer_id=other_customer.id,
|
|
warehouse_id=self.warehouse_relaxed.id,
|
|
created_by=self.user,
|
|
operator=self.operator,
|
|
items=[{'product_id': self.product.id, 'quantity': '1', 'unit': '米'}],
|
|
)
|
|
|
|
def test_update_pre_sales_order_replaces_items(self):
|
|
pre_sales_order = pre_order_services.create_pre_sales_order(
|
|
merchant=self.merchant,
|
|
customer_id=self.customer.id,
|
|
warehouse_id=self.warehouse_relaxed.id,
|
|
created_by=self.user,
|
|
operator=self.operator,
|
|
items=[{'product_id': self.product.id, 'quantity': '10', 'unit': '米'}],
|
|
)
|
|
|
|
updated = pre_order_services.update_pre_sales_order(
|
|
pre_sales_order=pre_sales_order,
|
|
warehouse_id=self.warehouse_strict.id,
|
|
items=[
|
|
{
|
|
'product_id': self.product.id,
|
|
'quantity': '20',
|
|
'unit': '米',
|
|
'order_quantity': 3,
|
|
}
|
|
],
|
|
remarks='updated-remarks',
|
|
)
|
|
|
|
self.assertEqual(updated.warehouse_id, self.warehouse_strict.id)
|
|
self.assertEqual(updated.remarks, 'updated-remarks')
|
|
self.assertEqual(updated.items.count(), 1)
|
|
self.assertEqual(str(updated.items.first().quantity), '20.00')
|
|
self.assertEqual(updated.items.first().order_quantity, 3)
|
|
|
|
def test_delete_pre_sales_order(self):
|
|
pre_sales_order = pre_order_services.create_pre_sales_order(
|
|
merchant=self.merchant,
|
|
customer_id=self.customer.id,
|
|
warehouse_id=self.warehouse_relaxed.id,
|
|
created_by=self.user,
|
|
operator=self.operator,
|
|
items=[{'product_id': self.product.id, 'quantity': '10', 'unit': '米'}],
|
|
)
|
|
order_id = pre_sales_order.id
|
|
|
|
pre_order_services.delete_pre_sales_order(pre_sales_order=pre_sales_order)
|
|
self.assertFalse(business_models.PreSalesOrder.objects.filter(id=order_id).exists())
|
|
|
|
def test_convert_pre_sales_order_to_sales_order_rejects_non_strict_out(self):
|
|
pre_sales_order = pre_order_services.create_pre_sales_order(
|
|
merchant=self.merchant,
|
|
customer_id=self.customer.id,
|
|
warehouse_id=self.warehouse_relaxed.id,
|
|
created_by=self.user,
|
|
operator=self.operator,
|
|
items=[{'product_id': self.product.id, 'quantity': '10', 'unit': '米'}],
|
|
)
|
|
|
|
with self.assertRaises(PermissionError):
|
|
pre_order_services.convert_pre_sales_order_to_sales_order(
|
|
merchant=self.merchant,
|
|
pre_sales_order_id=pre_sales_order.id,
|
|
operator=self.operator,
|
|
created_by=self.user,
|
|
)
|
|
|
|
def test_convert_pre_sales_order_to_sales_order_success(self):
|
|
pre_sales_order = pre_order_services.create_pre_sales_order(
|
|
merchant=self.merchant,
|
|
customer_id=self.customer.id,
|
|
warehouse_id=self.warehouse_strict_out.id,
|
|
created_by=self.user,
|
|
operator=self.operator,
|
|
items=[{'product_id': self.product.id, 'quantity': '10', 'unit': '米'}],
|
|
)
|
|
item = pre_sales_order.items.first()
|
|
|
|
stock_record = stock_models.StockChangeRecord.objects.create(
|
|
merchant=self.merchant,
|
|
type=stock_models.StockChangeTypeEnum.ADD,
|
|
warehouse=self.warehouse_strict_out,
|
|
source_type=stock_models.StockChangeSourceEnum.PURCHASE,
|
|
created_by=self.user,
|
|
)
|
|
stock_detail = stock_models.StockChangeDetail.objects.create(
|
|
merchant=self.merchant,
|
|
product=self.product,
|
|
unit=basic_models.ProductUnitEnum.METER,
|
|
stock_change_record=stock_record,
|
|
quantity='10',
|
|
)
|
|
|
|
allocation_services.create_allocation_record(
|
|
item=item,
|
|
stock_ids=[stock_detail.id],
|
|
quantity='10',
|
|
unit='米',
|
|
scanned_by=self.operator,
|
|
)
|
|
|
|
sales_order = pre_order_services.convert_pre_sales_order_to_sales_order(
|
|
merchant=self.merchant,
|
|
pre_sales_order_id=pre_sales_order.id,
|
|
operator=self.operator,
|
|
created_by=self.user,
|
|
)
|
|
|
|
self.assertEqual(sales_order.customer_id, self.customer.id)
|
|
self.assertEqual(sales_order.warehouse_id, self.warehouse_strict_out.id)
|
|
self.assertEqual(sales_order.from_pre_sales_order_id, pre_sales_order.id)
|
|
self.assertEqual(sales_order.status, business_models.SalesOrderStatusEnum.PENDING)
|
|
self.assertEqual(sales_order.items.count(), 1)
|
|
sales_item = sales_order.items.first()
|
|
self.assertEqual(sales_item.product_id, self.product.id)
|
|
self.assertEqual(sales_item.consume_detail_ids, str(stock_detail.id))
|
|
|
|
|
|
class PreSalesOrderSignalTestCase(TransactionTestCase):
|
|
def setUp(self):
|
|
(
|
|
self.merchant,
|
|
self.customer,
|
|
self.warehouse_strict,
|
|
self.warehouse_relaxed,
|
|
self.warehouse_strict_out,
|
|
self.product,
|
|
self.operator,
|
|
) = create_sales_fixtures()
|
|
|
|
self.user = get_user_model().objects.create_user(username='signal-user', password='pass')
|
|
self.operator.sys_user = self.user
|
|
self.operator.save(update_fields=['sys_user'])
|
|
|
|
def test_pre_sales_order_created_signal_fires_on_commit(self):
|
|
from business.signals import pre_sales_order_created
|
|
|
|
received = []
|
|
uid = 'tests.pre_sales_order_created.on_commit'
|
|
|
|
def _receiver(sender, **kwargs):
|
|
received.append(kwargs)
|
|
|
|
pre_sales_order_created.connect(_receiver, dispatch_uid=uid, weak=False)
|
|
try:
|
|
pre_order_services.create_pre_sales_order(
|
|
merchant=self.merchant,
|
|
customer_id=self.customer.id,
|
|
warehouse_id=self.warehouse_relaxed.id,
|
|
operator=self.operator,
|
|
created_by=self.user,
|
|
items=[{'product_id': self.product.id, 'quantity': '1', 'unit': '米'}],
|
|
)
|
|
|
|
self.assertEqual(len(received), 1)
|
|
payload = received[0]
|
|
self.assertIsNotNone(payload.get('instance'))
|
|
self.assertEqual(payload.get('created_by').id, self.user.id)
|
|
self.assertEqual(payload.get('operator').id, self.operator.id)
|
|
self.assertEqual(payload.get('items_count'), 1)
|
|
finally:
|
|
pre_sales_order_created.disconnect(dispatch_uid=uid)
|
|
|
|
def test_pre_sales_order_created_signal_not_fired_on_rollback(self):
|
|
from business.signals import pre_sales_order_created
|
|
|
|
received = []
|
|
uid = 'tests.pre_sales_order_created.on_rollback'
|
|
|
|
def _receiver(sender, **kwargs):
|
|
received.append(kwargs)
|
|
|
|
pre_sales_order_created.connect(_receiver, dispatch_uid=uid, weak=False)
|
|
try:
|
|
try:
|
|
with transaction.atomic():
|
|
pre_order_services.create_pre_sales_order(
|
|
merchant=self.merchant,
|
|
customer_id=self.customer.id,
|
|
warehouse_id=self.warehouse_relaxed.id,
|
|
operator=self.operator,
|
|
created_by=self.user,
|
|
items=[{'product_id': self.product.id, 'quantity': '1', 'unit': '米'}],
|
|
)
|
|
raise RuntimeError('force rollback')
|
|
except RuntimeError:
|
|
pass
|
|
|
|
self.assertEqual(received, [])
|
|
finally:
|
|
pre_sales_order_created.disconnect(dispatch_uid=uid)
|