forked from erp-dev/erp
1405 lines
56 KiB
Python
1405 lines
56 KiB
Python
from typing import List, Dict, Any
|
||
from decimal import Decimal
|
||
from concurrent.futures import ThreadPoolExecutor
|
||
|
||
from django.contrib.auth import get_user_model
|
||
from django.test import TestCase, TransactionTestCase, override_settings
|
||
from django.db import connections, close_old_connections
|
||
from django.utils import timezone
|
||
from unittest.mock import patch, MagicMock
|
||
|
||
from basic_info import models as basic_models
|
||
from stock import models as stock_models
|
||
from business import models as business_models, services, tasks
|
||
|
||
|
||
def create_basic_fixtures():
|
||
merchant = basic_models.Merchant.objects.create(
|
||
name='测试商户',
|
||
type=basic_models.MerchantTypeEnum.FACTORY,
|
||
)
|
||
basic_models.MerchantSetting.objects.create(
|
||
merchant=merchant,
|
||
key=basic_models.MerchantSettingKeyEnum.AUTO_CREATE_STOCK_CHANGE_TASKS,
|
||
type=basic_models.MerchantSettingTypeEnum.BOOL,
|
||
val_bool=True,
|
||
)
|
||
supplier = basic_models.Supplier.objects.create(
|
||
merchant=merchant,
|
||
name='测试供应商',
|
||
)
|
||
warehouse_strict = basic_models.WareHouse.objects.create(
|
||
merchant=merchant,
|
||
name='严进仓',
|
||
mode=basic_models.WareHouseModeEnum.RESTRICT_IN,
|
||
)
|
||
warehouse_relaxed = basic_models.WareHouse.objects.create(
|
||
merchant=merchant,
|
||
name='宽进仓',
|
||
mode=basic_models.WareHouseModeEnum.UNRESTRICTED,
|
||
)
|
||
category = basic_models.ProductCategory.objects.create(
|
||
merchant=merchant,
|
||
name='面料',
|
||
product_prefix='FAB',
|
||
)
|
||
product = basic_models.Product.objects.create(
|
||
merchant=merchant,
|
||
category=category,
|
||
name='测试面料',
|
||
human_id='FAB-001',
|
||
unit=basic_models.ProductUnitEnum.METER,
|
||
)
|
||
operator = basic_models.Employee.objects.create(
|
||
merchant=merchant,
|
||
name='经办人',
|
||
status=basic_models.EmployeeStatusEnum.ACTIVE,
|
||
)
|
||
return merchant, supplier, warehouse_strict, warehouse_relaxed, product, operator
|
||
|
||
|
||
def create_sales_fixtures():
|
||
merchant = basic_models.Merchant.objects.create(
|
||
name='销售商户',
|
||
type=basic_models.MerchantTypeEnum.FACTORY,
|
||
)
|
||
basic_models.MerchantSetting.objects.create(
|
||
merchant=merchant,
|
||
key=basic_models.MerchantSettingKeyEnum.AUTO_CREATE_STOCK_CHANGE_TASKS,
|
||
type=basic_models.MerchantSettingTypeEnum.BOOL,
|
||
val_bool=True,
|
||
)
|
||
customer = basic_models.Customer.objects.create(
|
||
merchant=merchant,
|
||
name='客户A',
|
||
mobile='13800000000',
|
||
created_by=None,
|
||
)
|
||
warehouse_strict = basic_models.WareHouse.objects.create(
|
||
merchant=merchant,
|
||
name='销售严进仓',
|
||
mode=basic_models.WareHouseModeEnum.RESTRICT_IN,
|
||
)
|
||
warehouse_relaxed = basic_models.WareHouse.objects.create(
|
||
merchant=merchant,
|
||
name='销售宽进仓',
|
||
mode=basic_models.WareHouseModeEnum.UNRESTRICTED,
|
||
)
|
||
warehouse_strict_out = basic_models.WareHouse.objects.create(
|
||
merchant=merchant,
|
||
name='销售严出仓',
|
||
mode=basic_models.WareHouseModeEnum.RESTRICT_IN_OUT,
|
||
)
|
||
category = basic_models.ProductCategory.objects.create(
|
||
merchant=merchant,
|
||
name='销售品类',
|
||
product_prefix='SAL',
|
||
)
|
||
product = basic_models.Product.objects.create(
|
||
merchant=merchant,
|
||
category=category,
|
||
name='销售面料',
|
||
human_id='SAL-001',
|
||
unit=basic_models.ProductUnitEnum.METER,
|
||
)
|
||
operator = basic_models.Employee.objects.create(
|
||
merchant=merchant,
|
||
name='销售员',
|
||
status=basic_models.EmployeeStatusEnum.ACTIVE,
|
||
)
|
||
return merchant, customer, warehouse_strict, warehouse_relaxed, warehouse_strict_out, product, operator
|
||
|
||
|
||
class PurchaseOrderServiceTestCase(TestCase):
|
||
def setUp(self):
|
||
(
|
||
self.merchant,
|
||
self.supplier,
|
||
self.warehouse_strict,
|
||
self.warehouse_relaxed,
|
||
self.product,
|
||
self.operator,
|
||
) = create_basic_fixtures()
|
||
User = get_user_model()
|
||
self.user = User.objects.create_user(username='creator', password='pass123')
|
||
self.strict_items: List[Dict[str, Any]] = [
|
||
{'product_id': self.product.id, 'numbers': [10, 5], 'price': '12.5', 'unit': '米'}
|
||
]
|
||
self.relaxed_items: List[Dict[str, Any]] = [
|
||
{'product_id': self.product.id, 'quantity': 120, 'num_of_rolls': 3, 'price': '10.0', 'unit': '米'}
|
||
]
|
||
|
||
def test_create_purchase_order_does_not_trigger_task_immediately(self):
|
||
with patch('business.services.create_purchase_order_stock_entries.delay') as mock_delay:
|
||
purchase_order = services.create_purchase_order(
|
||
merchant=self.merchant,
|
||
supplier=self.supplier,
|
||
order_date=timezone.now().date(),
|
||
warehouse=self.warehouse_strict,
|
||
operator=self.operator,
|
||
items=self.strict_items,
|
||
remarks='自动化测试',
|
||
created_by=self.user,
|
||
)
|
||
|
||
self.assertIsInstance(purchase_order, business_models.PurchaseOrder)
|
||
self.assertEqual(purchase_order.status, business_models.PurchaseOrderStatusEnum.PENDING)
|
||
self.assertEqual(purchase_order.items.count(), 1)
|
||
item = purchase_order.items.first()
|
||
self.assertEqual(item.quantity_of_rolls, '10,5')
|
||
self.assertEqual(item.split_quantity_of_rolls(), [10, 5])
|
||
mock_delay.assert_not_called()
|
||
|
||
def test_purchase_order_human_id_generation(self):
|
||
purchase_order = services.create_purchase_order(
|
||
merchant=self.merchant,
|
||
supplier=self.supplier,
|
||
order_date=timezone.now().date(),
|
||
warehouse=self.warehouse_strict,
|
||
operator=self.operator,
|
||
items=self.strict_items,
|
||
created_by=self.user,
|
||
)
|
||
self.assertIsNotNone(purchase_order.human_id)
|
||
self.assertEqual(
|
||
purchase_order.human_id,
|
||
f"CG{purchase_order.created_at.strftime('%Y%m%d')}{purchase_order.id:06d}",
|
||
)
|
||
|
||
def test_review_purchase_order_approval_triggers_task(self):
|
||
purchase_order = services.create_purchase_order(
|
||
merchant=self.merchant,
|
||
supplier=self.supplier,
|
||
order_date=timezone.now().date(),
|
||
warehouse=self.warehouse_strict,
|
||
operator=self.operator,
|
||
items=self.strict_items,
|
||
created_by=self.user,
|
||
)
|
||
|
||
with patch('business.services.create_purchase_order_stock_entries.delay') as mock_delay:
|
||
reviewed = services.review_purchase_order(
|
||
purchase_order=purchase_order,
|
||
target_status=business_models.PurchaseOrderStatusEnum.APPROVED,
|
||
reviewed_by=self.user,
|
||
)
|
||
|
||
self.assertEqual(reviewed.status, business_models.PurchaseOrderStatusEnum.APPROVED)
|
||
mock_delay.assert_called_once_with(
|
||
purchase_order_id=purchase_order.id,
|
||
warehouse_id=self.warehouse_strict.id,
|
||
items=[{'product_id': self.product.id, 'quantities': ['10', '5']}],
|
||
created_by_id=self.user.id,
|
||
)
|
||
balance = business_models.SupplierBalance.objects.get(
|
||
merchant=self.merchant,
|
||
supplier=self.supplier,
|
||
)
|
||
self.assertEqual(balance.balance, purchase_order.get_total_amount())
|
||
record = business_models.BalanceChangeRecord.objects.get(
|
||
merchant=self.merchant,
|
||
source_type=business_models.BalanceChangeSourceEnum.PURCHASE_ORDER,
|
||
source_id=purchase_order.id,
|
||
)
|
||
self.assertEqual(record.target_type, business_models.BalanceChangeTargetEnum.SUPPLIER)
|
||
self.assertEqual(record.delta, purchase_order.get_total_amount())
|
||
self.assertEqual(record.balance_after, balance.balance)
|
||
self.assertEqual(record.direction, business_models.BalanceChangeDirectionEnum.INCREASE)
|
||
|
||
def test_create_purchase_order_without_items_raises(self):
|
||
with self.assertRaises(ValueError):
|
||
services.create_purchase_order(
|
||
merchant=self.merchant,
|
||
supplier=self.supplier,
|
||
order_date=timezone.now().date(),
|
||
warehouse=self.warehouse_strict,
|
||
operator=self.operator,
|
||
items=[],
|
||
)
|
||
|
||
def test_review_purchase_order_cancel_without_stock(self):
|
||
purchase_order = services.create_purchase_order(
|
||
merchant=self.merchant,
|
||
supplier=self.supplier,
|
||
order_date=timezone.now().date(),
|
||
warehouse=self.warehouse_relaxed,
|
||
operator=self.operator,
|
||
items=self.relaxed_items,
|
||
created_by=self.user,
|
||
)
|
||
item = purchase_order.items.first()
|
||
self.assertIsNone(item.quantity_of_rolls)
|
||
self.assertEqual(item.split_quantity_of_rolls(), [])
|
||
|
||
cancelled = services.review_purchase_order(
|
||
purchase_order=purchase_order,
|
||
target_status=business_models.PurchaseOrderStatusEnum.CANCELLED,
|
||
reviewed_by=self.user,
|
||
)
|
||
self.assertEqual(cancelled.status, business_models.PurchaseOrderStatusEnum.CANCELLED)
|
||
self.assertFalse(
|
||
business_models.BalanceChangeRecord.objects.filter(
|
||
merchant=self.merchant,
|
||
source_type=business_models.BalanceChangeSourceEnum.PURCHASE_ORDER,
|
||
source_id=purchase_order.id,
|
||
).exists()
|
||
)
|
||
|
||
def test_review_purchase_order_cancel_blocked_after_stock_created(self):
|
||
purchase_order = services.create_purchase_order(
|
||
merchant=self.merchant,
|
||
supplier=self.supplier,
|
||
order_date=timezone.now().date(),
|
||
warehouse=self.warehouse_relaxed,
|
||
operator=self.operator,
|
||
items=self.relaxed_items,
|
||
created_by=self.user,
|
||
)
|
||
stock_models.StockChangeRecord.objects.create(
|
||
merchant=self.merchant,
|
||
type=stock_models.StockChangeTypeEnum.ADD,
|
||
warehouse=self.warehouse_relaxed,
|
||
source_type=stock_models.StockChangeSourceEnum.PURCHASE,
|
||
source_id=purchase_order.id,
|
||
)
|
||
|
||
with self.assertRaises(ValueError):
|
||
services.review_purchase_order(
|
||
purchase_order=purchase_order,
|
||
target_status=business_models.PurchaseOrderStatusEnum.CANCELLED,
|
||
reviewed_by=self.user,
|
||
)
|
||
|
||
def test_review_purchase_order_relaxed_mode_approval_payload(self):
|
||
purchase_order = services.create_purchase_order(
|
||
merchant=self.merchant,
|
||
supplier=self.supplier,
|
||
order_date=timezone.now().date(),
|
||
warehouse=self.warehouse_relaxed,
|
||
operator=self.operator,
|
||
items=self.relaxed_items,
|
||
created_by=self.user,
|
||
)
|
||
|
||
with patch('business.services.create_purchase_order_stock_entries.delay') as mock_delay:
|
||
services.review_purchase_order(
|
||
purchase_order=purchase_order,
|
||
target_status=business_models.PurchaseOrderStatusEnum.APPROVED,
|
||
reviewed_by=self.user,
|
||
)
|
||
|
||
mock_delay.assert_called_once_with(
|
||
purchase_order_id=purchase_order.id,
|
||
warehouse_id=self.warehouse_relaxed.id,
|
||
items=[{'product_id': self.product.id, 'value': '120.00', 'num_of_rolls': 3}],
|
||
created_by_id=self.user.id,
|
||
)
|
||
|
||
def test_update_purchase_order_success(self):
|
||
purchase_order = services.create_purchase_order(
|
||
merchant=self.merchant,
|
||
supplier=self.supplier,
|
||
order_date=timezone.now().date(),
|
||
warehouse=self.warehouse_relaxed,
|
||
operator=self.operator,
|
||
items=self.relaxed_items,
|
||
created_by=self.user,
|
||
)
|
||
updated = services.update_purchase_order(
|
||
purchase_order=purchase_order,
|
||
warehouse=self.warehouse_strict,
|
||
items=self.strict_items,
|
||
remarks='updated',
|
||
)
|
||
self.assertEqual(updated.remarks, 'updated')
|
||
self.assertEqual(updated.items.count(), 1)
|
||
self.assertEqual(updated.items.first().split_quantity_of_rolls(), [10, 5])
|
||
|
||
def test_update_purchase_order_rejects_non_pending(self):
|
||
purchase_order = services.create_purchase_order(
|
||
merchant=self.merchant,
|
||
supplier=self.supplier,
|
||
order_date=timezone.now().date(),
|
||
warehouse=self.warehouse_relaxed,
|
||
operator=self.operator,
|
||
items=self.relaxed_items,
|
||
created_by=self.user,
|
||
)
|
||
purchase_order.status = business_models.PurchaseOrderStatusEnum.APPROVED
|
||
purchase_order.save(update_fields=['status'])
|
||
with self.assertRaises(ValueError):
|
||
services.update_purchase_order(
|
||
purchase_order=purchase_order,
|
||
items=self.relaxed_items,
|
||
)
|
||
|
||
def test_purchase_order_relaxed_mode_even_split(self):
|
||
purchase_order = services.create_purchase_order(
|
||
merchant=self.merchant,
|
||
supplier=self.supplier,
|
||
order_date=timezone.now().date(),
|
||
warehouse=self.warehouse_relaxed,
|
||
operator=self.operator,
|
||
items=[{'product_id': self.product.id, 'quantity': 99, 'num_of_rolls': 2, 'price': '10.0', 'unit': '米'}],
|
||
created_by=self.user,
|
||
)
|
||
stock_flow_items = services._build_stock_flow_items_from_order(purchase_order)
|
||
payload = services.create_purchase_order_stock_entries_sync(
|
||
purchase_order_id=purchase_order.id,
|
||
warehouse_id=self.warehouse_relaxed.id,
|
||
items=stock_flow_items,
|
||
created_by_id=self.user.id,
|
||
)
|
||
record = stock_models.StockChangeRecord.objects.get(id=payload['stock_change_record_id'])
|
||
quantities = sorted([detail.quantity for detail in record.details.order_by('id')], reverse=True)
|
||
self.assertEqual(quantities, [Decimal('49.50'), Decimal('49.50')])
|
||
|
||
|
||
class SalesOrderServiceTestCase(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()
|
||
User = get_user_model()
|
||
self.user = User.objects.create_user(username='sales-creator', password='pass123')
|
||
self.strict_items = [
|
||
{'product_id': self.product.id, 'numbers': [6, 4], 'price': '30.5', 'unit': '米'}
|
||
]
|
||
self.relaxed_items = [
|
||
{'product_id': self.product.id, 'quantity': 80, 'num_of_rolls': 2, 'price': '28.0', 'unit': '米'}
|
||
]
|
||
self.strict_out_items = [
|
||
{
|
||
'product_id': self.product.id,
|
||
'consume_detail_ids': [1, 2],
|
||
'quantity': 40,
|
||
'price': '32.0',
|
||
'unit': '米',
|
||
}
|
||
]
|
||
|
||
def test_create_sales_order_success(self):
|
||
with patch('business.services.create_sales_order_stock_entries.delay') as mock_delay:
|
||
sales_order = services.create_sales_order(
|
||
merchant=self.merchant,
|
||
customer=self.customer,
|
||
order_date=timezone.now().date(),
|
||
warehouse=self.warehouse_strict,
|
||
operator=self.operator,
|
||
items=self.strict_items,
|
||
remarks='销售测试',
|
||
created_by=self.user,
|
||
)
|
||
|
||
self.assertIsInstance(sales_order, business_models.SalesOrder)
|
||
self.assertEqual(sales_order.status, business_models.SalesOrderStatusEnum.PENDING)
|
||
self.assertEqual(sales_order.items.count(), 1)
|
||
item = sales_order.items.first()
|
||
self.assertEqual(item.quantity_of_rolls, '6,4')
|
||
self.assertEqual(item.split_quantity_of_rolls(), [6, 4])
|
||
mock_delay.assert_not_called()
|
||
|
||
def test_sales_order_human_id_generation(self):
|
||
sales_order = services.create_sales_order(
|
||
merchant=self.merchant,
|
||
customer=self.customer,
|
||
order_date=timezone.now().date(),
|
||
warehouse=self.warehouse_strict,
|
||
operator=self.operator,
|
||
items=self.strict_items,
|
||
created_by=self.user,
|
||
)
|
||
|
||
self.assertIsNotNone(sales_order.human_id)
|
||
self.assertTrue(len(sales_order.human_id) >= 14)
|
||
self.assertEqual(
|
||
sales_order.human_id,
|
||
f"XS{sales_order.created_at.strftime('%Y%m%d')}{sales_order.id:06d}",
|
||
)
|
||
|
||
def test_review_sales_order_triggers_task(self):
|
||
sales_order = services.create_sales_order(
|
||
merchant=self.merchant,
|
||
customer=self.customer,
|
||
order_date=timezone.now().date(),
|
||
warehouse=self.warehouse_strict,
|
||
operator=self.operator,
|
||
items=self.strict_items,
|
||
created_by=self.user,
|
||
)
|
||
with patch('business.services.create_sales_order_stock_entries.delay') as mock_delay:
|
||
reviewed = services.review_sales_order(
|
||
sales_order=sales_order,
|
||
target_status=business_models.SalesOrderStatusEnum.APPROVED,
|
||
reviewed_by=self.user,
|
||
)
|
||
self.assertEqual(reviewed.status, business_models.SalesOrderStatusEnum.APPROVED)
|
||
mock_delay.assert_called_once_with(
|
||
sales_order_id=sales_order.id,
|
||
warehouse_id=self.warehouse_strict.id,
|
||
items=[{'product_id': self.product.id, 'quantities': ['6', '4']}],
|
||
created_by_id=self.user.id,
|
||
)
|
||
balance = business_models.CustomerBalance.objects.get(
|
||
merchant=self.merchant,
|
||
customer=self.customer,
|
||
)
|
||
self.assertEqual(balance.balance, sales_order.get_total_amount())
|
||
record = business_models.BalanceChangeRecord.objects.get(
|
||
merchant=self.merchant,
|
||
source_type=business_models.BalanceChangeSourceEnum.SALES_ORDER,
|
||
source_id=sales_order.id,
|
||
)
|
||
self.assertEqual(record.target_type, business_models.BalanceChangeTargetEnum.CUSTOMER)
|
||
self.assertEqual(record.delta, sales_order.get_total_amount())
|
||
self.assertEqual(record.balance_after, balance.balance)
|
||
self.assertEqual(record.direction, business_models.BalanceChangeDirectionEnum.INCREASE)
|
||
|
||
def test_sales_order_cancel_blocked_after_stock_created(self):
|
||
sales_order = services.create_sales_order(
|
||
merchant=self.merchant,
|
||
customer=self.customer,
|
||
order_date=timezone.now().date(),
|
||
warehouse=self.warehouse_relaxed,
|
||
operator=self.operator,
|
||
items=self.relaxed_items,
|
||
created_by=self.user,
|
||
)
|
||
stock_models.StockChangeRecord.objects.create(
|
||
merchant=self.merchant,
|
||
type=stock_models.StockChangeTypeEnum.REMOVE,
|
||
warehouse=self.warehouse_relaxed,
|
||
source_type=stock_models.StockChangeSourceEnum.SALES,
|
||
source_id=sales_order.id,
|
||
)
|
||
with self.assertRaises(ValueError):
|
||
services.review_sales_order(
|
||
sales_order=sales_order,
|
||
target_status=business_models.SalesOrderStatusEnum.CANCELLED,
|
||
reviewed_by=self.user,
|
||
)
|
||
self.assertFalse(
|
||
business_models.BalanceChangeRecord.objects.filter(
|
||
merchant=self.merchant,
|
||
source_type=business_models.BalanceChangeSourceEnum.SALES_ORDER,
|
||
source_id=sales_order.id,
|
||
).exists()
|
||
)
|
||
|
||
def test_sales_order_requires_consume_ids_for_strict_out(self):
|
||
with self.assertRaises(ValueError):
|
||
services.create_sales_order(
|
||
merchant=self.merchant,
|
||
customer=self.customer,
|
||
order_date=timezone.now().date(),
|
||
warehouse=self.warehouse_strict_out,
|
||
operator=self.operator,
|
||
items=[{'product_id': self.product.id, 'quantity': 10, 'price': '20', 'unit': '米'}],
|
||
created_by=self.user,
|
||
)
|
||
|
||
def test_sales_order_stores_consume_ids(self):
|
||
sales_order = services.create_sales_order(
|
||
merchant=self.merchant,
|
||
customer=self.customer,
|
||
order_date=timezone.now().date(),
|
||
warehouse=self.warehouse_strict_out,
|
||
operator=self.operator,
|
||
items=self.strict_out_items,
|
||
created_by=self.user,
|
||
)
|
||
item = sales_order.items.first()
|
||
self.assertEqual(item.consume_detail_ids, '1,2')
|
||
|
||
def test_sales_order_outgoing_missing_consume_ids_raises(self):
|
||
payload = [{
|
||
'product_id': self.product.id,
|
||
'quantity': 30,
|
||
'price': '25',
|
||
'unit': '米',
|
||
}]
|
||
with self.assertRaises(ValueError):
|
||
services.create_sales_order(
|
||
merchant=self.merchant,
|
||
customer=self.customer,
|
||
order_date=timezone.now().date(),
|
||
warehouse=self.warehouse_strict_out,
|
||
operator=self.operator,
|
||
items=payload,
|
||
created_by=self.user,
|
||
)
|
||
|
||
def test_update_sales_order_success(self):
|
||
sales_order = services.create_sales_order(
|
||
merchant=self.merchant,
|
||
customer=self.customer,
|
||
order_date=timezone.now().date(),
|
||
warehouse=self.warehouse_strict,
|
||
operator=self.operator,
|
||
items=self.strict_items,
|
||
created_by=self.user,
|
||
)
|
||
updated = services.update_sales_order(
|
||
sales_order=sales_order,
|
||
warehouse=self.warehouse_relaxed,
|
||
items=self.relaxed_items,
|
||
remarks='updated',
|
||
)
|
||
self.assertEqual(updated.remarks, 'updated')
|
||
self.assertEqual(updated.items.count(), 1)
|
||
item = updated.items.first()
|
||
self.assertEqual(item.split_quantity_of_rolls(), [])
|
||
|
||
def test_update_sales_order_rejects_non_pending(self):
|
||
sales_order = services.create_sales_order(
|
||
merchant=self.merchant,
|
||
customer=self.customer,
|
||
order_date=timezone.now().date(),
|
||
warehouse=self.warehouse_strict,
|
||
operator=self.operator,
|
||
items=self.strict_items,
|
||
created_by=self.user,
|
||
)
|
||
sales_order.status = business_models.SalesOrderStatusEnum.APPROVED
|
||
sales_order.save(update_fields=['status'])
|
||
with self.assertRaises(ValueError):
|
||
services.update_sales_order(
|
||
sales_order=sales_order,
|
||
items=self.strict_items,
|
||
)
|
||
|
||
|
||
class PurchaseReturnServiceTestCase(TestCase):
|
||
def setUp(self):
|
||
(
|
||
self.merchant,
|
||
self.supplier,
|
||
self.warehouse_strict,
|
||
self.warehouse_relaxed,
|
||
self.product,
|
||
self.operator,
|
||
) = create_basic_fixtures()
|
||
self.warehouse_strict_out = basic_models.WareHouse.objects.create(
|
||
merchant=self.merchant,
|
||
name='采退严出仓',
|
||
mode=basic_models.WareHouseModeEnum.RESTRICT_IN_OUT,
|
||
)
|
||
User = get_user_model()
|
||
self.user = User.objects.create_user(username='purchase-return', password='pass123')
|
||
self.strict_items = [
|
||
{'product_id': self.product.id, 'numbers': [4, 2], 'price': '11.5', 'unit': '米'}
|
||
]
|
||
self.strict_out_items = [
|
||
{
|
||
'product_id': self.product.id,
|
||
'consume_detail_ids': [10, 11],
|
||
'quantity': 30,
|
||
'price': '12.0',
|
||
'unit': '米',
|
||
}
|
||
]
|
||
|
||
def test_create_purchase_return_order_success(self):
|
||
order = services.create_purchase_return_order(
|
||
merchant=self.merchant,
|
||
supplier=self.supplier,
|
||
return_date=timezone.now().date(),
|
||
warehouse=self.warehouse_strict,
|
||
operator=self.operator,
|
||
items=self.strict_items,
|
||
created_by=self.user,
|
||
)
|
||
self.assertEqual(order.status, business_models.PurchaseReturnStatusEnum.PENDING)
|
||
self.assertEqual(order.items.count(), 1)
|
||
item = order.items.first()
|
||
self.assertEqual(item.quantity_of_rolls, '4,2')
|
||
self.assertEqual(item.split_quantity_of_rolls(), [4, 2])
|
||
|
||
def test_purchase_return_order_human_id_generation(self):
|
||
order = services.create_purchase_return_order(
|
||
merchant=self.merchant,
|
||
supplier=self.supplier,
|
||
return_date=timezone.now().date(),
|
||
warehouse=self.warehouse_strict,
|
||
operator=self.operator,
|
||
items=self.strict_items,
|
||
created_by=self.user,
|
||
)
|
||
self.assertIsNotNone(order.human_id)
|
||
self.assertEqual(
|
||
order.human_id,
|
||
f"CT{order.created_at.strftime('%Y%m%d')}{order.id:06d}",
|
||
)
|
||
|
||
def test_review_purchase_return_order_triggers_task_and_balance(self):
|
||
order = services.create_purchase_return_order(
|
||
merchant=self.merchant,
|
||
supplier=self.supplier,
|
||
return_date=timezone.now().date(),
|
||
warehouse=self.warehouse_relaxed,
|
||
operator=self.operator,
|
||
items=[{'product_id': self.product.id, 'quantity': 40, 'num_of_rolls': 2, 'price': '10.0'}],
|
||
created_by=self.user,
|
||
)
|
||
with patch('business.services.create_purchase_return_order_stock_entries.delay') as mock_delay:
|
||
reviewed = services.review_purchase_return_order(
|
||
purchase_return_order=order,
|
||
target_status=business_models.PurchaseReturnStatusEnum.APPROVED,
|
||
reviewed_by=self.user,
|
||
)
|
||
self.assertEqual(reviewed.status, business_models.PurchaseReturnStatusEnum.APPROVED)
|
||
mock_delay.assert_called_once()
|
||
balance = business_models.SupplierBalance.objects.get(
|
||
merchant=self.merchant,
|
||
supplier=self.supplier,
|
||
)
|
||
self.assertEqual(balance.balance, Decimal('-400.00'))
|
||
record = business_models.BalanceChangeRecord.objects.get(
|
||
merchant=self.merchant,
|
||
source_type=business_models.BalanceChangeSourceEnum.PURCHASE_RETURN_ORDER,
|
||
source_id=order.id,
|
||
)
|
||
self.assertEqual(record.direction, business_models.BalanceChangeDirectionEnum.DECREASE)
|
||
|
||
def test_purchase_return_cancel_blocked_after_stock_created(self):
|
||
order = services.create_purchase_return_order(
|
||
merchant=self.merchant,
|
||
supplier=self.supplier,
|
||
return_date=timezone.now().date(),
|
||
warehouse=self.warehouse_relaxed,
|
||
operator=self.operator,
|
||
items=[{'product_id': self.product.id, 'quantity': 20, 'num_of_rolls': 2, 'price': '9.5'}],
|
||
)
|
||
stock_models.StockChangeRecord.objects.create(
|
||
merchant=self.merchant,
|
||
type=stock_models.StockChangeTypeEnum.REMOVE,
|
||
warehouse=self.warehouse_relaxed,
|
||
source_type=stock_models.StockChangeSourceEnum.PURCHASE_RETURN,
|
||
source_id=order.id,
|
||
)
|
||
with self.assertRaises(ValueError):
|
||
services.review_purchase_return_order(
|
||
purchase_return_order=order,
|
||
target_status=business_models.PurchaseReturnStatusEnum.CANCELLED,
|
||
)
|
||
|
||
def test_purchase_return_strict_out_requires_consume_ids(self):
|
||
payload = [
|
||
{
|
||
'product_id': self.product.id,
|
||
'quantity': 10,
|
||
'price': '8.5',
|
||
'unit': '米',
|
||
}
|
||
]
|
||
with self.assertRaises(ValueError):
|
||
services.create_purchase_return_order(
|
||
merchant=self.merchant,
|
||
supplier=self.supplier,
|
||
return_date=timezone.now().date(),
|
||
warehouse=self.warehouse_strict_out,
|
||
operator=self.operator,
|
||
items=payload,
|
||
)
|
||
|
||
def test_update_purchase_return_order_success(self):
|
||
order = services.create_purchase_return_order(
|
||
merchant=self.merchant,
|
||
supplier=self.supplier,
|
||
return_date=timezone.now().date(),
|
||
warehouse=self.warehouse_strict,
|
||
operator=self.operator,
|
||
items=self.strict_items,
|
||
created_by=self.user,
|
||
)
|
||
updated = services.update_purchase_return_order(
|
||
purchase_return_order=order,
|
||
items=self.strict_items,
|
||
remarks='updated',
|
||
)
|
||
self.assertEqual(updated.remarks, 'updated')
|
||
self.assertEqual(updated.items.count(), 1)
|
||
self.assertEqual(updated.items.first().split_quantity_of_rolls(), [4, 2])
|
||
|
||
def test_update_purchase_return_order_rejects_non_pending(self):
|
||
order = services.create_purchase_return_order(
|
||
merchant=self.merchant,
|
||
supplier=self.supplier,
|
||
return_date=timezone.now().date(),
|
||
warehouse=self.warehouse_strict,
|
||
operator=self.operator,
|
||
items=self.strict_items,
|
||
created_by=self.user,
|
||
)
|
||
order.status = business_models.PurchaseReturnStatusEnum.APPROVED
|
||
order.save(update_fields=['status'])
|
||
with self.assertRaises(ValueError):
|
||
services.update_purchase_return_order(
|
||
purchase_return_order=order,
|
||
items=self.strict_items,
|
||
)
|
||
|
||
|
||
class SalesReturnServiceTestCase(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()
|
||
User = get_user_model()
|
||
self.user = User.objects.create_user(username='sales-return', password='pass123')
|
||
self.strict_items = [
|
||
{'product_id': self.product.id, 'numbers': [6, 2], 'price': '20.0', 'unit': '米'}
|
||
]
|
||
|
||
def test_create_sales_return_order_success(self):
|
||
order = services.create_sales_return_order(
|
||
merchant=self.merchant,
|
||
customer=self.customer,
|
||
return_date=timezone.now().date(),
|
||
warehouse=self.warehouse_relaxed,
|
||
operator=self.operator,
|
||
items=[{'product_id': self.product.id, 'quantity': 50, 'num_of_rolls': 2, 'price': '19.5'}],
|
||
)
|
||
self.assertEqual(order.status, business_models.SalesReturnStatusEnum.PENDING)
|
||
item = order.items.first()
|
||
self.assertEqual(item.split_quantity_of_rolls(), [])
|
||
|
||
def test_sales_return_order_human_id_generation(self):
|
||
order = services.create_sales_return_order(
|
||
merchant=self.merchant,
|
||
customer=self.customer,
|
||
return_date=timezone.now().date(),
|
||
warehouse=self.warehouse_relaxed,
|
||
operator=self.operator,
|
||
items=[{'product_id': self.product.id, 'quantity': 50, 'num_of_rolls': 2, 'price': '19.5'}],
|
||
created_by=self.user,
|
||
)
|
||
self.assertIsNotNone(order.human_id)
|
||
self.assertEqual(
|
||
order.human_id,
|
||
f"XT{order.created_at.strftime('%Y%m%d')}{order.id:06d}",
|
||
)
|
||
|
||
def test_review_sales_return_order_updates_balance(self):
|
||
order = services.create_sales_return_order(
|
||
merchant=self.merchant,
|
||
customer=self.customer,
|
||
return_date=timezone.now().date(),
|
||
warehouse=self.warehouse_strict,
|
||
operator=self.operator,
|
||
items=self.strict_items,
|
||
created_by=self.user,
|
||
)
|
||
self.assertEqual(order.items.first().split_quantity_of_rolls(), [6, 2])
|
||
with patch('business.services.create_sales_return_order_stock_entries.delay') as mock_delay:
|
||
reviewed = services.review_sales_return_order(
|
||
sales_return_order=order,
|
||
target_status=business_models.SalesReturnStatusEnum.APPROVED,
|
||
reviewed_by=self.user,
|
||
)
|
||
self.assertEqual(reviewed.status, business_models.SalesReturnStatusEnum.APPROVED)
|
||
mock_delay.assert_called_once()
|
||
balance = business_models.CustomerBalance.objects.get(
|
||
merchant=self.merchant,
|
||
customer=self.customer,
|
||
)
|
||
self.assertLess(balance.balance, Decimal('0'))
|
||
record = business_models.BalanceChangeRecord.objects.get(
|
||
merchant=self.merchant,
|
||
source_type=business_models.BalanceChangeSourceEnum.SALES_RETURN_ORDER,
|
||
source_id=order.id,
|
||
)
|
||
self.assertEqual(record.direction, business_models.BalanceChangeDirectionEnum.DECREASE)
|
||
|
||
def test_sales_return_cancel_blocked_after_stock_created(self):
|
||
order = services.create_sales_return_order(
|
||
merchant=self.merchant,
|
||
customer=self.customer,
|
||
return_date=timezone.now().date(),
|
||
warehouse=self.warehouse_relaxed,
|
||
operator=self.operator,
|
||
items=[{'product_id': self.product.id, 'quantity': 15, 'num_of_rolls': 1, 'price': '18.0'}],
|
||
)
|
||
stock_models.StockChangeRecord.objects.create(
|
||
merchant=self.merchant,
|
||
type=stock_models.StockChangeTypeEnum.ADD,
|
||
warehouse=self.warehouse_relaxed,
|
||
source_type=stock_models.StockChangeSourceEnum.SALES_RETURN,
|
||
source_id=order.id,
|
||
)
|
||
with self.assertRaises(ValueError):
|
||
services.review_sales_return_order(
|
||
sales_return_order=order,
|
||
target_status=business_models.SalesReturnStatusEnum.CANCELLED,
|
||
)
|
||
|
||
def test_sales_return_relaxed_mode_stock_payload_preserves_rolls(self):
|
||
order = services.create_sales_return_order(
|
||
merchant=self.merchant,
|
||
customer=self.customer,
|
||
return_date=timezone.now().date(),
|
||
warehouse=self.warehouse_relaxed,
|
||
operator=self.operator,
|
||
items=[{'product_id': self.product.id, 'quantity': 100, 'num_of_rolls': 1, 'price': '22.0'}],
|
||
)
|
||
stock_flow_items = services._build_stock_flow_items_from_order(order)
|
||
self.assertEqual(len(stock_flow_items), 1)
|
||
self.assertEqual(stock_flow_items[0]['num_of_rolls'], 1)
|
||
|
||
def test_sales_return_relaxed_mode_even_split(self):
|
||
order = services.create_sales_return_order(
|
||
merchant=self.merchant,
|
||
customer=self.customer,
|
||
return_date=timezone.now().date(),
|
||
warehouse=self.warehouse_relaxed,
|
||
operator=self.operator,
|
||
items=[{'product_id': self.product.id, 'quantity': 100, 'num_of_rolls': 3, 'price': '22.0'}],
|
||
created_by=self.user,
|
||
)
|
||
stock_flow_items = services._build_stock_flow_items_from_order(order)
|
||
payload = services.create_sales_return_order_stock_entries_sync(
|
||
sales_return_order_id=order.id,
|
||
warehouse_id=self.warehouse_relaxed.id,
|
||
items=stock_flow_items,
|
||
created_by_id=self.user.id,
|
||
)
|
||
record = stock_models.StockChangeRecord.objects.get(id=payload['stock_change_record_id'])
|
||
quantities = sorted([detail.quantity for detail in record.details.order_by('id')], reverse=True)
|
||
self.assertEqual(quantities, [Decimal('33.34'), Decimal('33.33'), Decimal('33.33')])
|
||
|
||
def test_update_sales_return_order_success(self):
|
||
order = services.create_sales_return_order(
|
||
merchant=self.merchant,
|
||
customer=self.customer,
|
||
return_date=timezone.now().date(),
|
||
warehouse=self.warehouse_strict,
|
||
operator=self.operator,
|
||
items=self.strict_items,
|
||
created_by=self.user,
|
||
)
|
||
updated = services.update_sales_return_order(
|
||
sales_return_order=order,
|
||
items=self.strict_items,
|
||
remarks='updated',
|
||
)
|
||
self.assertEqual(updated.remarks, 'updated')
|
||
self.assertEqual(updated.items.count(), 1)
|
||
self.assertEqual(updated.items.first().split_quantity_of_rolls(), [6, 2])
|
||
|
||
def test_update_sales_return_order_rejects_non_pending(self):
|
||
order = services.create_sales_return_order(
|
||
merchant=self.merchant,
|
||
customer=self.customer,
|
||
return_date=timezone.now().date(),
|
||
warehouse=self.warehouse_strict,
|
||
operator=self.operator,
|
||
items=self.strict_items,
|
||
created_by=self.user,
|
||
)
|
||
order.status = business_models.SalesReturnStatusEnum.APPROVED
|
||
order.save(update_fields=['status'])
|
||
with self.assertRaises(ValueError):
|
||
services.update_sales_return_order(
|
||
sales_return_order=order,
|
||
items=self.strict_items,
|
||
)
|
||
|
||
class PaymentReceiptServiceTestCase(TestCase):
|
||
def setUp(self):
|
||
(
|
||
self.merchant,
|
||
self.supplier,
|
||
self.warehouse_strict,
|
||
self.warehouse_relaxed,
|
||
self.product,
|
||
self.operator,
|
||
) = create_basic_fixtures()
|
||
self.customer = basic_models.Customer.objects.create(
|
||
merchant=self.merchant,
|
||
name='客户B',
|
||
mobile='13888888888',
|
||
created_by=None,
|
||
)
|
||
self.bank_account = basic_models.BankAccount.objects.create(
|
||
merchant=self.merchant,
|
||
name='测试账户',
|
||
auto_number='BANK-SVC',
|
||
)
|
||
|
||
def test_create_payment_order_and_approve(self):
|
||
order = services.create_payment_order(
|
||
merchant=self.merchant,
|
||
supplier=self.supplier,
|
||
payment_date=timezone.now().date(),
|
||
amount='120.50',
|
||
discount_amount='20.50',
|
||
operator=self.operator,
|
||
remarks='付款测试',
|
||
bank_account=self.bank_account,
|
||
markup='服务层附言',
|
||
)
|
||
self.assertEqual(order.status, business_models.PaymentOrderStatusEnum.PENDING)
|
||
self.assertEqual(order.bank_account, self.bank_account)
|
||
self.assertEqual(order.markup, '服务层附言')
|
||
self.assertEqual(order.discount_amount, Decimal('20.50'))
|
||
self.assertEqual(order.settlement_amount, Decimal('141.00'))
|
||
reviewed = services.review_payment_order(
|
||
payment_order=order,
|
||
target_status=business_models.PaymentOrderStatusEnum.APPROVED,
|
||
reviewed_by=self.operator,
|
||
)
|
||
self.assertEqual(reviewed.status, business_models.PaymentOrderStatusEnum.APPROVED)
|
||
balance = business_models.SupplierBalance.objects.get(
|
||
merchant=self.merchant,
|
||
supplier=self.supplier,
|
||
)
|
||
self.assertEqual(balance.balance, Decimal('-141.00'))
|
||
record = business_models.BalanceChangeRecord.objects.get(
|
||
merchant=self.merchant,
|
||
source_type=business_models.BalanceChangeSourceEnum.PAYMENT_ORDER,
|
||
source_id=order.id,
|
||
)
|
||
self.assertEqual(record.direction, business_models.BalanceChangeDirectionEnum.DECREASE)
|
||
self.assertEqual(record.delta, Decimal('-141.00'))
|
||
self.assertEqual(record.balance_after, balance.balance)
|
||
with self.assertRaises(ValueError):
|
||
services.review_payment_order(
|
||
payment_order=order,
|
||
target_status=business_models.PaymentOrderStatusEnum.CANCELLED,
|
||
reviewed_by=self.operator,
|
||
)
|
||
|
||
def test_create_receipt_order_and_cancel(self):
|
||
order = services.create_receipt_order(
|
||
merchant=self.merchant,
|
||
customer=self.customer,
|
||
receipt_date=timezone.now().date(),
|
||
amount='80.00',
|
||
discount_amount='5',
|
||
operator=self.operator,
|
||
remarks='收款测试',
|
||
bank_account=self.bank_account,
|
||
markup='收款附言',
|
||
)
|
||
self.assertEqual(order.bank_account, self.bank_account)
|
||
self.assertEqual(order.discount_amount, Decimal('5'))
|
||
self.assertEqual(order.settlement_amount, Decimal('85'))
|
||
cancelled = services.review_receipt_order(
|
||
receipt_order=order,
|
||
target_status=business_models.ReceiptOrderStatusEnum.CANCELLED,
|
||
reviewed_by=self.operator,
|
||
)
|
||
self.assertEqual(cancelled.status, business_models.ReceiptOrderStatusEnum.CANCELLED)
|
||
self.assertFalse(
|
||
business_models.BalanceChangeRecord.objects.filter(
|
||
merchant=self.merchant,
|
||
source_type=business_models.BalanceChangeSourceEnum.RECEIPT_ORDER,
|
||
source_id=order.id,
|
||
).exists()
|
||
)
|
||
|
||
def test_payment_amount_can_be_negative(self):
|
||
order = services.create_payment_order(
|
||
merchant=self.merchant,
|
||
supplier=self.supplier,
|
||
payment_date=timezone.now().date(),
|
||
amount='-25.50',
|
||
operator=self.operator,
|
||
)
|
||
self.assertEqual(order.amount, Decimal('-25.50'))
|
||
|
||
def test_payment_discount_can_exceed_amount(self):
|
||
order = services.create_payment_order(
|
||
merchant=self.merchant,
|
||
supplier=self.supplier,
|
||
payment_date=timezone.now().date(),
|
||
amount='50',
|
||
discount_amount='60',
|
||
operator=self.operator,
|
||
)
|
||
self.assertEqual(order.settlement_amount, Decimal('110'))
|
||
|
||
def test_receipt_discount_can_exceed_amount(self):
|
||
order = services.create_receipt_order(
|
||
merchant=self.merchant,
|
||
customer=self.customer,
|
||
receipt_date=timezone.now().date(),
|
||
amount='50',
|
||
discount_amount='60',
|
||
operator=self.operator,
|
||
)
|
||
self.assertEqual(order.settlement_amount, Decimal('110'))
|
||
|
||
def test_receipt_amount_can_be_negative(self):
|
||
order = services.create_receipt_order(
|
||
merchant=self.merchant,
|
||
customer=self.customer,
|
||
receipt_date=timezone.now().date(),
|
||
amount='-40',
|
||
operator=self.operator,
|
||
)
|
||
self.assertEqual(order.amount, Decimal('-40'))
|
||
|
||
def test_payment_order_rejects_foreign_bank_account(self):
|
||
other_merchant = basic_models.Merchant.objects.create(
|
||
name='无关商户',
|
||
type=basic_models.MerchantTypeEnum.FACTORY,
|
||
)
|
||
foreign_bank = basic_models.BankAccount.objects.create(
|
||
merchant=other_merchant,
|
||
name='其他账户',
|
||
auto_number='BANK-FOREIGN',
|
||
)
|
||
with self.assertRaises(ValueError):
|
||
services.create_payment_order(
|
||
merchant=self.merchant,
|
||
supplier=self.supplier,
|
||
payment_date=timezone.now().date(),
|
||
amount='10',
|
||
operator=self.operator,
|
||
bank_account=foreign_bank,
|
||
)
|
||
|
||
def test_receipt_order_rejects_foreign_bank_account(self):
|
||
other_merchant = basic_models.Merchant.objects.create(
|
||
name='第二商户',
|
||
type=basic_models.MerchantTypeEnum.FACTORY,
|
||
)
|
||
foreign_bank = basic_models.BankAccount.objects.create(
|
||
merchant=other_merchant,
|
||
name='第二账户',
|
||
auto_number='BANK-SECOND',
|
||
)
|
||
with self.assertRaises(ValueError):
|
||
services.create_receipt_order(
|
||
merchant=self.merchant,
|
||
customer=self.customer,
|
||
receipt_date=timezone.now().date(),
|
||
amount='10',
|
||
operator=self.operator,
|
||
bank_account=foreign_bank,
|
||
)
|
||
|
||
def test_receipt_approval_is_idempotent(self):
|
||
order = services.create_receipt_order(
|
||
merchant=self.merchant,
|
||
customer=self.customer,
|
||
receipt_date=timezone.now().date(),
|
||
amount='10',
|
||
operator=self.operator,
|
||
)
|
||
services.review_receipt_order(
|
||
receipt_order=order,
|
||
target_status=business_models.ReceiptOrderStatusEnum.APPROVED,
|
||
reviewed_by=self.operator,
|
||
)
|
||
reviewed_again = services.review_receipt_order(
|
||
receipt_order=order,
|
||
target_status=business_models.ReceiptOrderStatusEnum.APPROVED,
|
||
reviewed_by=self.operator,
|
||
)
|
||
self.assertEqual(reviewed_again.status, business_models.ReceiptOrderStatusEnum.APPROVED)
|
||
balance = business_models.CustomerBalance.objects.get(
|
||
merchant=self.merchant,
|
||
customer=self.customer,
|
||
)
|
||
self.assertEqual(balance.balance, Decimal('-10'))
|
||
records = business_models.BalanceChangeRecord.objects.filter(
|
||
merchant=self.merchant,
|
||
source_type=business_models.BalanceChangeSourceEnum.RECEIPT_ORDER,
|
||
source_id=order.id,
|
||
)
|
||
self.assertEqual(records.count(), 1)
|
||
self.assertEqual(records.first().balance_after, balance.balance)
|
||
|
||
def test_payment_approval_is_idempotent(self):
|
||
order = services.create_payment_order(
|
||
merchant=self.merchant,
|
||
supplier=self.supplier,
|
||
payment_date=timezone.now().date(),
|
||
amount='75.00',
|
||
operator=self.operator,
|
||
)
|
||
services.review_payment_order(
|
||
payment_order=order,
|
||
target_status=business_models.PaymentOrderStatusEnum.APPROVED,
|
||
reviewed_by=self.operator,
|
||
)
|
||
services.review_payment_order(
|
||
payment_order=order,
|
||
target_status=business_models.PaymentOrderStatusEnum.APPROVED,
|
||
reviewed_by=self.operator,
|
||
)
|
||
balance = business_models.SupplierBalance.objects.get(
|
||
merchant=self.merchant,
|
||
supplier=self.supplier,
|
||
)
|
||
self.assertEqual(balance.balance, Decimal('-75.00'))
|
||
records = business_models.BalanceChangeRecord.objects.filter(
|
||
merchant=self.merchant,
|
||
source_type=business_models.BalanceChangeSourceEnum.PAYMENT_ORDER,
|
||
source_id=order.id,
|
||
)
|
||
self.assertEqual(records.count(), 1)
|
||
self.assertEqual(records.first().balance_after, balance.balance)
|
||
|
||
|
||
class PurchaseOrderStockServiceTestCase(TestCase):
|
||
def setUp(self):
|
||
(
|
||
self.merchant,
|
||
self.supplier,
|
||
self.warehouse_strict,
|
||
self.warehouse_relaxed,
|
||
self.product,
|
||
self.operator,
|
||
) = create_basic_fixtures()
|
||
User = get_user_model()
|
||
self.user = User.objects.create_user(username='svc-user', password='pass123')
|
||
self.purchase_order = business_models.PurchaseOrder.objects.create(
|
||
merchant=self.merchant,
|
||
supplier=self.supplier,
|
||
purchase_date=timezone.now().date(),
|
||
operator=self.operator,
|
||
warehouse=self.warehouse_strict,
|
||
)
|
||
self.items = [{'product_id': self.product.id, 'quantities': ['8']}]
|
||
|
||
def test_service_calls_stock_flow_service(self):
|
||
with patch('business.services.StockFlowService') as mock_flow_cls:
|
||
mock_instance = mock_flow_cls.return_value
|
||
mock_instance.stock_in.return_value = (MagicMock(id=321), [], 2)
|
||
|
||
payload = services.create_purchase_order_stock_entries_sync(
|
||
purchase_order_id=self.purchase_order.id,
|
||
warehouse_id=self.warehouse_strict.id,
|
||
items=self.items,
|
||
created_by_id=self.user.id,
|
||
)
|
||
|
||
mock_flow_cls.assert_called_once_with(merchant=self.merchant, created_by=self.user)
|
||
mock_instance.stock_in.assert_called_once_with(
|
||
warehouse_id=self.warehouse_strict.id,
|
||
source_type=stock_models.StockChangeSourceEnum.PURCHASE,
|
||
source_id=self.purchase_order.id,
|
||
items=[{'product_id': self.product.id, 'quantities': ['8']}],
|
||
)
|
||
self.assertEqual(payload['purchase_order_id'], self.purchase_order.id)
|
||
self.assertEqual(payload['stock_change_record_id'], 321)
|
||
|
||
|
||
class SalesOrderStockServiceTestCase(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()
|
||
User = get_user_model()
|
||
self.user = User.objects.create_user(username='sales-svc', password='pass123')
|
||
self.sales_order = business_models.SalesOrder.objects.create(
|
||
merchant=self.merchant,
|
||
customer=self.customer,
|
||
sales_date=timezone.now().date(),
|
||
operator=self.operator,
|
||
warehouse=self.warehouse_strict,
|
||
)
|
||
self.items = [{'product_id': self.product.id, 'quantities': ['5']}]
|
||
|
||
def test_service_calls_stock_out(self):
|
||
with patch('business.services.StockFlowService') as mock_flow_cls:
|
||
mock_instance = mock_flow_cls.return_value
|
||
mock_instance.stock_out.return_value = (MagicMock(id=654), [], 1)
|
||
|
||
payload = services.create_sales_order_stock_entries_sync(
|
||
sales_order_id=self.sales_order.id,
|
||
warehouse_id=self.warehouse_strict.id,
|
||
items=self.items,
|
||
created_by_id=self.user.id,
|
||
)
|
||
|
||
mock_flow_cls.assert_called_once_with(merchant=self.merchant, created_by=self.user)
|
||
mock_instance.stock_out.assert_called_once_with(
|
||
warehouse_id=self.warehouse_strict.id,
|
||
source_type=stock_models.StockChangeSourceEnum.SALES,
|
||
source_id=self.sales_order.id,
|
||
items=[{'product_id': self.product.id, 'quantities': ['5']}],
|
||
)
|
||
self.assertEqual(payload['sales_order_id'], self.sales_order.id)
|
||
self.assertEqual(payload['stock_change_record_id'], 654)
|
||
|
||
def test_service_calls_stock_out_strict_mode(self):
|
||
strict_order = business_models.SalesOrder.objects.create(
|
||
merchant=self.merchant,
|
||
customer=self.customer,
|
||
sales_date=timezone.now().date(),
|
||
operator=self.operator,
|
||
warehouse=self.warehouse_strict_out,
|
||
)
|
||
consume_items = [{'product_id': self.product.id, 'consume_detail_ids': [11, 12], 'quantity': 50}]
|
||
with patch('business.services.StockFlowService') as mock_flow_cls:
|
||
mock_instance = mock_flow_cls.return_value
|
||
mock_instance.stock_out.return_value = (MagicMock(id=987), [], 2)
|
||
|
||
payload = services.create_sales_order_stock_entries_sync(
|
||
sales_order_id=strict_order.id,
|
||
warehouse_id=self.warehouse_strict_out.id,
|
||
items=consume_items,
|
||
created_by_id=self.user.id,
|
||
)
|
||
|
||
mock_flow_cls.assert_called_once_with(merchant=self.merchant, created_by=self.user)
|
||
mock_instance.stock_out.assert_called_once_with(
|
||
warehouse_id=self.warehouse_strict_out.id,
|
||
source_type=stock_models.StockChangeSourceEnum.SALES,
|
||
source_id=strict_order.id,
|
||
items=[{'product_id': self.product.id, 'consume_detail_ids': [11, 12], 'quantity': 50}],
|
||
)
|
||
self.assertEqual(payload['sales_order_id'], strict_order.id)
|
||
self.assertEqual(payload['stock_change_record_id'], 987)
|
||
|
||
|
||
class SalesOrderConcurrencyTestCase(TransactionTestCase):
|
||
reset_sequences = True
|
||
|
||
def setUp(self):
|
||
(
|
||
self.merchant,
|
||
self.customer,
|
||
self.warehouse_strict,
|
||
self.warehouse_relaxed,
|
||
self.warehouse_strict_out,
|
||
self.product,
|
||
self.operator,
|
||
) = create_sales_fixtures()
|
||
basic_models.MerchantSetting.objects.filter(
|
||
merchant=self.merchant,
|
||
key=basic_models.MerchantSettingKeyEnum.AUTO_CREATE_STOCK_CHANGE_TASKS,
|
||
).update(val_bool=False)
|
||
User = get_user_model()
|
||
self.user = User.objects.create_user(username='concurrent', password='pass123')
|
||
self.sales_order = services.create_sales_order(
|
||
merchant=self.merchant,
|
||
customer=self.customer,
|
||
order_date=timezone.now().date(),
|
||
warehouse=self.warehouse_strict,
|
||
operator=self.operator,
|
||
items=[{'product_id': self.product.id, 'numbers': [5], 'price': '12', 'unit': '米'}],
|
||
created_by=self.user,
|
||
)
|
||
|
||
def test_concurrent_sales_order_approval_updates_balance_once(self):
|
||
def approve():
|
||
# ThreadPoolExecutor 会复用线程；Django 的 DB connection 是线程局部的，
|
||
# 若不显式关闭，可能导致测试 DB 在 teardown 时仍被占用，无法 DROP。
|
||
close_old_connections()
|
||
try:
|
||
services.review_sales_order(
|
||
sales_order_id=self.sales_order.id,
|
||
target_status=business_models.SalesOrderStatusEnum.APPROVED,
|
||
reviewed_by=self.user,
|
||
)
|
||
finally:
|
||
connections.close_all()
|
||
|
||
with ThreadPoolExecutor(max_workers=2) as executor:
|
||
futures = [executor.submit(approve) for _ in range(2)]
|
||
for future in futures:
|
||
future.result()
|
||
|
||
balance = business_models.CustomerBalance.objects.get(
|
||
merchant=self.merchant,
|
||
customer=self.customer,
|
||
)
|
||
self.assertEqual(balance.balance, self.sales_order.get_total_amount())
|
||
records = business_models.BalanceChangeRecord.objects.filter(
|
||
merchant=self.merchant,
|
||
source_type=business_models.BalanceChangeSourceEnum.SALES_ORDER,
|
||
source_id=self.sales_order.id,
|
||
)
|
||
self.assertEqual(records.count(), 1)
|
||
self.assertEqual(records.first().balance_after, balance.balance)
|
||
|
||
def tearDown(self):
|
||
connections.close_all()
|
||
|
||
|
||
@override_settings(
|
||
CELERY_TASK_ALWAYS_EAGER=True,
|
||
CELERY_TASK_EAGER_PROPAGATES=True,
|
||
)
|
||
class PurchaseOrderStockTaskTestCase(TestCase):
|
||
def setUp(self):
|
||
self.purchase_order_id = 123
|
||
self.warehouse_id = 456
|
||
self.items = [{'product_id': 1, 'quantities': ['5']}]
|
||
|
||
def test_task_delegates_to_service(self):
|
||
with patch('business.tasks.business_services.create_purchase_order_stock_entries_sync') as mock_sync:
|
||
mock_sync.return_value = {'purchase_order_id': self.purchase_order_id}
|
||
|
||
async_result = tasks.create_purchase_order_stock_entries.delay(
|
||
purchase_order_id=self.purchase_order_id,
|
||
warehouse_id=self.warehouse_id,
|
||
items=self.items,
|
||
created_by_id=999,
|
||
)
|
||
payload = async_result.get(timeout=5)
|
||
|
||
mock_sync.assert_called_once_with(
|
||
purchase_order_id=self.purchase_order_id,
|
||
warehouse_id=self.warehouse_id,
|
||
items=self.items,
|
||
created_by_id=999,
|
||
)
|
||
self.assertEqual(payload['purchase_order_id'], self.purchase_order_id)
|
||
self.assertIn('task_id', payload)
|
||
|
||
|
||
@override_settings(
|
||
CELERY_TASK_ALWAYS_EAGER=True,
|
||
CELERY_TASK_EAGER_PROPAGATES=True,
|
||
)
|
||
class SalesOrderStockTaskTestCase(TestCase):
|
||
def setUp(self):
|
||
self.sales_order_id = 321
|
||
self.warehouse_id = 654
|
||
self.items = [{'product_id': 2, 'quantities': ['4']}]
|
||
|
||
def test_task_delegates_to_service(self):
|
||
with patch('business.tasks.business_services.create_sales_order_stock_entries_sync') as mock_sync:
|
||
mock_sync.return_value = {'sales_order_id': self.sales_order_id}
|
||
|
||
async_result = tasks.create_sales_order_stock_entries.delay(
|
||
sales_order_id=self.sales_order_id,
|
||
warehouse_id=self.warehouse_id,
|
||
items=self.items,
|
||
created_by_id=777,
|
||
)
|
||
payload = async_result.get(timeout=5)
|
||
|
||
mock_sync.assert_called_once_with(
|
||
sales_order_id=self.sales_order_id,
|
||
warehouse_id=self.warehouse_id,
|
||
items=self.items,
|
||
created_by_id=777,
|
||
)
|
||
self.assertEqual(payload['sales_order_id'], self.sales_order_id)
|
||
self.assertIn('task_id', payload)
|