forked from erp-dev/erp
324 lines
13 KiB
Python
324 lines
13 KiB
Python
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 basic_info import models as basic_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_review_purchase_order_approval_no_task_when_auto_disabled(self):
|
|
basic_models.MerchantSetting.objects.filter(
|
|
merchant=self.merchant,
|
|
key=basic_models.MerchantSettingKeyEnum.AUTO_CREATE_STOCK_CHANGE_TASKS,
|
|
).update(val_bool=False)
|
|
|
|
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_not_called()
|
|
|
|
def test_bind_purchase_order_stock_change_record(self):
|
|
basic_models.MerchantSetting.objects.filter(
|
|
merchant=self.merchant,
|
|
key=basic_models.MerchantSettingKeyEnum.AUTO_CREATE_STOCK_CHANGE_TASKS,
|
|
).update(val_bool=False)
|
|
|
|
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,
|
|
)
|
|
|
|
record = stock_models.StockChangeRecord.objects.create(
|
|
merchant=self.merchant,
|
|
type=stock_models.StockChangeTypeEnum.ADD,
|
|
warehouse=self.warehouse_strict,
|
|
source_type=stock_models.StockChangeSourceEnum.PURCHASE,
|
|
created_by=self.user,
|
|
)
|
|
|
|
bound = services.bind_purchase_order_stock_change_record(
|
|
purchase_order_id=purchase_order.id,
|
|
stock_change_record_id=record.id,
|
|
operator=self.user,
|
|
)
|
|
self.assertEqual(bound.source_id, purchase_order.id)
|
|
|
|
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')])
|