erpnew/business/tests.py

from typing import List, Dict, Any

from django.contrib.auth import get_user_model
from django.test import TestCase, override_settings
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


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')
        mock_delay.assert_not_called()

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

    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,
        )
        self.assertIsNone(purchase_order.items.first().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)

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


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)


@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)
from django.test import TestCase

# Create your tests here.