erpnew/business/tests/test_sales_order.py

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_sales_fixtures


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,
            )