erpnew/mes/tests.py

from django.contrib.contenttypes.models import ContentType
from django.contrib.auth import get_user_model
from django.db.models import ProtectedError
from django.test import TestCase

from basic_info.models import Customer, Employee, Merchant, MerchantTypeEnum
from mes.models import CapacityUnitEnum, Device, DeviceCategory, ProductionAssignment, ProductionAssignmentStatusEnum
from mes import services
from printing.models import PrintingOrder


class MesServiceTestCase(TestCase):
    def setUp(self):
        self.user = get_user_model().objects.create_user(username='mes-user', password='pass12345')
        self.other_user = get_user_model().objects.create_user(username='mes-other-user', password='pass12345')
        self.merchant = Merchant.objects.create(name='MES商户', type=MerchantTypeEnum.STORE)
        self.other_merchant = Merchant.objects.create(name='其他商户', type=MerchantTypeEnum.STORE)
        self.operator = Employee.objects.create(merchant=self.merchant, sys_user=self.user, name='操作员')
        self.other_operator = Employee.objects.create(merchant=self.other_merchant, sys_user=self.other_user, name='其他操作员')

    def test_create_device_category(self):
        category = services.create_device_category(
            merchant=self.merchant,
            name='打印机',
            created_by=self.user,
            operator=self.operator,
        )

        self.assertEqual(category.merchant, self.merchant)
        self.assertEqual(category.created_by, self.user)
        self.assertEqual(category.operator, self.operator)

    def test_create_device_category_rejects_cross_merchant_operator(self):
        with self.assertRaisesMessage(ValueError, '操作人不属于当前商户'):
            services.create_device_category(
                merchant=self.merchant,
                name='打印机',
                created_by=self.user,
                operator=self.other_operator,
            )

    def test_create_device(self):
        category = services.create_device_category(
            merchant=self.merchant,
            name='打印机',
            created_by=self.user,
            operator=self.operator,
        )

        device = services.create_device(
            merchant=self.merchant,
            category=category,
            name='设备A',
            created_by=self.user,
            operator=self.operator,
            peak_capacity=120,
            capacity_unit=CapacityUnitEnum.METER,
            extra={'capacity': {'per_hour': 120}},
        )

        self.assertEqual(device.category, category)
        self.assertEqual(device.peak_capacity, 120)
        self.assertEqual(device.capacity_unit, CapacityUnitEnum.METER)
        self.assertEqual(device.extra, {'capacity': {'per_hour': 120}})

    def test_create_device_rejects_cross_merchant_category(self):
        other_category = services.create_device_category(
            merchant=self.other_merchant,
            name='其他分类',
            created_by=self.other_user,
            operator=self.other_operator,
        )

        with self.assertRaisesMessage(ValueError, '设备分类不属于当前商户'):
            services.create_device(
                merchant=self.merchant,
                category=other_category,
                name='设备A',
                created_by=self.user,
                operator=self.operator,
                peak_capacity=120,
            )

    def test_update_device(self):
        category = services.create_device_category(
            merchant=self.merchant,
            name='打印机',
            created_by=self.user,
            operator=self.operator,
        )
        another_category = services.create_device_category(
            merchant=self.merchant,
            name='滚筒机',
            created_by=self.user,
            operator=self.operator,
        )
        device = services.create_device(
            merchant=self.merchant,
            category=category,
            name='设备A',
            created_by=self.user,
            operator=self.operator,
            peak_capacity=120,
        )

        updated = services.update_device(
            device=device,
            operator=self.operator,
            name='设备B',
            category=another_category,
            peak_capacity=180,
            capacity_unit=CapacityUnitEnum.YARD,
            extra={'power': 220},
        )

        self.assertEqual(updated.name, '设备B')
        self.assertEqual(updated.category, another_category)
        self.assertEqual(updated.peak_capacity, 180)
        self.assertEqual(updated.capacity_unit, CapacityUnitEnum.YARD)
        self.assertEqual(updated.extra, {'power': 220})

    def test_update_device_allows_clearing_extra(self):
        category = services.create_device_category(
            merchant=self.merchant,
            name='打印机',
            created_by=self.user,
            operator=self.operator,
        )
        device = services.create_device(
            merchant=self.merchant,
            category=category,
            name='设备A',
            created_by=self.user,
            operator=self.operator,
            peak_capacity=120,
            extra={'power': 220},
        )

        updated = services.update_device(device=device, operator=self.operator, extra=None)

        self.assertIsNone(updated.extra)

    def test_delete_device_category_is_protected_when_devices_exist(self):
        category = services.create_device_category(
            merchant=self.merchant,
            name='打印机',
            created_by=self.user,
            operator=self.operator,
        )
        services.create_device(
            merchant=self.merchant,
            category=category,
            name='设备A',
            created_by=self.user,
            operator=self.operator,
            peak_capacity=120,
        )

        with self.assertRaises(ProtectedError):
            services.delete_device_category(category=category, operator=self.operator)

    def test_list_devices_for_merchant_filters_by_category(self):
        category = services.create_device_category(
            merchant=self.merchant,
            name='打印机',
            created_by=self.user,
            operator=self.operator,
        )
        another_category = services.create_device_category(
            merchant=self.merchant,
            name='滚筒机',
            created_by=self.user,
            operator=self.operator,
        )
        target = services.create_device(
            merchant=self.merchant,
            category=category,
            name='设备A',
            created_by=self.user,
            operator=self.operator,
            peak_capacity=120,
        )
        services.create_device(
            merchant=self.merchant,
            category=another_category,
            name='设备B',
            created_by=self.user,
            operator=self.operator,
            peak_capacity=80,
        )

        devices = list(services.list_devices_for_merchant(merchant=self.merchant, category_id=category.id))

        self.assertEqual([device.id for device in devices], [target.id])

    def test_list_devices_for_merchant_filters_by_extra_json_path(self):
        category = services.create_device_category(
            merchant=self.merchant,
            name='打印机',
            created_by=self.user,
            operator=self.operator,
        )
        target = services.create_device(
            merchant=self.merchant,
            category=category,
            name='设备A',
            created_by=self.user,
            operator=self.operator,
            peak_capacity=120,
            extra={'capacity': {'per_hour': 120}},
        )
        services.create_device(
            merchant=self.merchant,
            category=category,
            name='设备B',
            created_by=self.user,
            operator=self.operator,
            peak_capacity=80,
            extra={'capacity': {'per_hour': 80}},
        )

        devices = list(
            services.list_devices_for_merchant(
                merchant=self.merchant,
                extra_json_path=['capacity', 'per_hour'],
                extra_json_value=120,
            )
        )

        self.assertEqual([device.id for device in devices], [target.id])

    def test_create_production_assignment(self):
        customer = Customer.objects.create(
            merchant=self.merchant,
            name='测试客户',
            mobile='13800138000',
        )
        printing_order = PrintingOrder.objects.create(
            merchant=self.merchant,
            customer=customer,
            fabric='测试面料',
            width='150cm',
            created_by=self.user,
        )
        category = services.create_device_category(
            merchant=self.merchant,
            name='打印机',
            created_by=self.user,
            operator=self.operator,
        )
        device = services.create_device(
            merchant=self.merchant,
            category=category,
            name='设备A',
            created_by=self.user,
            operator=self.operator,
            peak_capacity=120,
        )
        content_type = ContentType.objects.get_for_model(PrintingOrder)

        assignment = services.create_production_assignment(
            merchant=self.merchant,
            device=device,
            content_type=content_type,
            object_id=printing_order.id,
            assigner=self.operator,
            production_quantity=300,
            assignee=None,
            created_by=self.user,
            operator=self.operator,
            extra={'batch': 'A1'},
        )

        self.assertEqual(assignment.device, device)
        self.assertEqual(assignment.assigner, self.operator)
        self.assertIsNone(assignment.assignee)
        self.assertEqual(assignment.production_quantity, 300)
        self.assertEqual(assignment.status, ProductionAssignmentStatusEnum.DRAFT)
        self.assertEqual(assignment.extra, {'batch': 'A1'})

    def test_create_production_assignment_rejects_cross_merchant_assignee(self):
        customer = Customer.objects.create(
            merchant=self.merchant,
            name='测试客户',
            mobile='13800138000',
        )
        printing_order = PrintingOrder.objects.create(
            merchant=self.merchant,
            customer=customer,
            fabric='测试面料',
            width='150cm',
            created_by=self.user,
        )
        category = services.create_device_category(
            merchant=self.merchant,
            name='打印机',
            created_by=self.user,
            operator=self.operator,
        )
        device = services.create_device(
            merchant=self.merchant,
            category=category,
            name='设备A',
            created_by=self.user,
            operator=self.operator,
            peak_capacity=120,
        )
        content_type = ContentType.objects.get_for_model(PrintingOrder)

        with self.assertRaisesMessage(ValueError, '被指派人不属于当前商户'):
            services.create_production_assignment(
                merchant=self.merchant,
                device=device,
                content_type=content_type,
                object_id=printing_order.id,
                assigner=self.operator,
                production_quantity=300,
                assignee=self.other_operator,
                created_by=self.user,
                operator=self.operator,
            )

    def test_update_production_assignment_rejects_invalid_status_transition(self):
        customer = Customer.objects.create(
            merchant=self.merchant,
            name='测试客户',
            mobile='13800138001',
        )
        printing_order = PrintingOrder.objects.create(
            merchant=self.merchant,
            customer=customer,
            fabric='测试面料',
            width='150cm',
            created_by=self.user,
        )
        category = services.create_device_category(
            merchant=self.merchant,
            name='打印机',
            created_by=self.user,
            operator=self.operator,
        )
        device = services.create_device(
            merchant=self.merchant,
            category=category,
            name='设备A',
            created_by=self.user,
            operator=self.operator,
            peak_capacity=120,
        )
        content_type = ContentType.objects.get_for_model(PrintingOrder)
        assignment = services.create_production_assignment(
            merchant=self.merchant,
            device=device,
            content_type=content_type,
            object_id=printing_order.id,
            assigner=self.operator,
            production_quantity=300,
            created_by=self.user,
            operator=self.operator,
            status=ProductionAssignmentStatusEnum.ACCEPTED,
        )

        with self.assertRaisesMessage(ValueError, '当前状态不允许变更为目标状态'):
            services.update_production_assignment(
                assignment=assignment,
                operator=self.operator,
                status=ProductionAssignmentStatusEnum.CANCELLED,
            )