forked from erp-dev/erp
feat: added pre sales order in business module, and explode tests of business module
This commit is contained in:
265
business/tests/test_purchase_order.py
Normal file
265
business/tests/test_purchase_order.py
Normal file
@@ -0,0 +1,265 @@
|
||||
from typing import Any, Dict, List
|
||||
from decimal import Decimal
|
||||
|
||||
from django.contrib.auth import get_user_model
|
||||
from django.test import TestCase
|
||||
from django.utils import timezone
|
||||
from unittest.mock import patch
|
||||
|
||||
from stock import models as stock_models
|
||||
from business import models as business_models, services
|
||||
|
||||
from .fixtures import create_basic_fixtures
|
||||
|
||||
|
||||
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')])
|
||||
Reference in New Issue
Block a user