erpnew/business/tests/test_allocation_services.py

from decimal import Decimal

from django.contrib.auth import get_user_model
from django.test import TestCase, override_settings

from basic_info.models import (
    Merchant,
    MerchantTypeEnum,
    Customer,
    WareHouse,
    WareHouseModeEnum,
    ProductCategory,
    Product,
    ProductUnitEnum,
    Employee,
    EmployeeStatusEnum,
)
from business import models as business_models
from business import allocation_services
from stock import models as stock_models


@override_settings(
    CELERY_TASK_ALWAYS_EAGER=True,
    CELERY_TASK_EAGER_PROPAGATES=True,
)
class AllocationServicesTestCase(TestCase):
    def setUp(self):
        self.merchant = Merchant.objects.create(name='配货商户', type=MerchantTypeEnum.FACTORY)
        self.customer = Customer.objects.create(
            merchant=self.merchant,
            name='配货客户',
            created_by=None,
        )
        self.warehouse = WareHouse.objects.create(
            merchant=self.merchant,
            name='配货仓库',
            mode=WareHouseModeEnum.UNRESTRICTED,
        )
        category = ProductCategory.objects.create(
            merchant=self.merchant,
            name='配货品类',
            product_prefix='ALC',
        )
        self.product = Product.objects.create(
            merchant=self.merchant,
            category=category,
            name='配货产品',
            human_id='ALC-001',
            unit=ProductUnitEnum.METER,
        )
        self.other_product = Product.objects.create(
            merchant=self.merchant,
            category=category,
            name='其他产品',
            human_id='ALC-002',
            unit=ProductUnitEnum.METER,
        )

        User = get_user_model()
        self.user = User.objects.create_user(username='allocation_user', password='pass123')
        self.employee = Employee.objects.create(
            merchant=self.merchant,
            sys_user=self.user,
            name='配货员',
            status=EmployeeStatusEnum.ACTIVE,
        )

        self.pre_sales_order = business_models.PreSalesOrder.objects.create(
            merchant=self.merchant,
            customer=self.customer,
            warehouse=self.warehouse,
            created_by=self.user,
            operator=self.employee,
            kind=business_models.SalesOrderKindEnum.WHOLESALE,
            remarks='预销售单备注',
        )
        self.item = business_models.PreSalesOrderItem.objects.create(
            pre_sales_order=self.pre_sales_order,
            product_id=self.product.id,
            product_name=self.product.name,
            quantity=Decimal('12.5'),
            unit='米',
            remarks='明细备注',
        )

        self.stock_record = stock_models.StockChangeRecord.objects.create(
            merchant=self.merchant,
            type=stock_models.StockChangeTypeEnum.ADD,
            warehouse=self.warehouse,
            source_type=stock_models.StockChangeSourceEnum.PURCHASE,
            created_by=self.user,
        )
        self.stock_detail = stock_models.StockChangeDetail.objects.create(
            merchant=self.merchant,
            product=self.product,
            unit=ProductUnitEnum.METER,
            stock_change_record=self.stock_record,
            quantity='12.5',
        )

    def test_validate_stock_details_missing(self):
        with self.assertRaises(ValueError) as ctx:
            allocation_services.validate_stock_details(
                merchant=self.merchant,
                stock_ids=[999999],
            )
        self.assertIn('库存明细不存在', str(ctx.exception))

    def test_validate_stock_details_consumed(self):
        self.stock_detail.is_consumed = True
        self.stock_detail.save(update_fields=['is_consumed'])

        with self.assertRaises(ValueError) as ctx:
            allocation_services.validate_stock_details(
                merchant=self.merchant,
                stock_ids=[self.stock_detail.id],
            )
        self.assertIn('库存明细已被消费', str(ctx.exception))

    def test_validate_stock_details_frozen(self):
        stock_models.StockFreeze.objects.create(
            merchant=self.merchant,
            product=self.product,
            warehouse=self.warehouse,
            stock_detail=self.stock_detail,
            quantity=Decimal('1'),
            unit=ProductUnitEnum.METER,
            status=stock_models.StockFreezeStatusEnum.FROZEN,
            frozen_by=self.user,
        )

        with self.assertRaises(ValueError) as ctx:
            allocation_services.validate_stock_details(
                merchant=self.merchant,
                stock_ids=[self.stock_detail.id],
            )
        self.assertIn('库存明细已被冻结', str(ctx.exception))

    def test_create_allocation_record_product_mismatch(self):
        other_detail = stock_models.StockChangeDetail.objects.create(
            merchant=self.merchant,
            product=self.other_product,
            unit=ProductUnitEnum.METER,
            stock_change_record=self.stock_record,
            quantity='3.0',
        )

        with self.assertRaises(ValueError) as ctx:
            allocation_services.create_allocation_record(
                item=self.item,
                stock_ids=[other_detail.id],
                quantity='3.0',
                unit='米',
                scanned_by=self.employee,
            )
        self.assertIn('库存明细产品不匹配', str(ctx.exception))

    def test_create_allocation_record_success(self):
        record = allocation_services.create_allocation_record(
            item=self.item,
            stock_ids=[self.stock_detail.id],
            quantity='5.5',
            unit='米',
            scanned_by=self.employee,
            remarks='扫码录入',
        )
        self.assertEqual(record.pre_sales_order_item_id, self.item.id)
        self.assertEqual(record.stock_ids, str(self.stock_detail.id))
        self.assertEqual(record.stock_id_list, [self.stock_detail.id])
        self.assertEqual(record.status, business_models.AllocationRecordStatusEnum.ACTIVE)
        freeze_qs = stock_models.StockFreeze.objects.filter(stock_detail=self.stock_detail)
        self.assertEqual(freeze_qs.count(), 1)
        self.assertEqual(freeze_qs.first().status, stock_models.StockFreezeStatusEnum.FROZEN)

    def test_cancel_allocation_record(self):
        record = allocation_services.create_allocation_record(
            item=self.item,
            stock_ids=[self.stock_detail.id],
            quantity='2.0',
            unit='米',
            scanned_by=self.employee,
        )
        updated = allocation_services.cancel_allocation_record(record=record)
        self.assertEqual(updated.status, business_models.AllocationRecordStatusEnum.CANCELLED)