erpnew/api_v2/test_mes_api.py

from django.contrib.auth import get_user_model
from django.contrib.contenttypes.models import ContentType
from django.test import TestCase
from rest_framework.test import APIClient

from basic_info import models as basic_models
from mes import models as mes_models
from printing.models import PrintingOrder


class MesV2APITest(TestCase):
    def setUp(self):
        self.client = APIClient()
        self.merchant = basic_models.Merchant.objects.create(name="MES商户", type=basic_models.MerchantTypeEnum.STORE)
        self.other_merchant = basic_models.Merchant.objects.create(name="其他商户", type=basic_models.MerchantTypeEnum.STORE)
        self.user = get_user_model().objects.create_user(username="mes-api-user", password="pass12345")
        self.other_user = get_user_model().objects.create_user(username="mes-api-other-user", password="pass12345")
        self.employee = basic_models.Employee.objects.create(merchant=self.merchant, sys_user=self.user, name="MES员工")
        self.other_employee = basic_models.Employee.objects.create(merchant=self.other_merchant, sys_user=self.other_user, name="其他员工")
        self.client.force_authenticate(user=self.user)

    def test_create_device_category(self):
        resp = self.client.post("/api/v2/mes/device-categories/", {"name": "打印机"}, format="json")

        self.assertEqual(resp.status_code, 201)
        self.assertEqual(resp.data["name"], "打印机")
        self.assertEqual(resp.data["merchant"], self.merchant.id)
        self.assertEqual(resp.data["created_by"]["id"], self.user.id)
        self.assertEqual(resp.data["operator"]["id"], self.employee.id)

    def test_list_device_categories_supports_name_filter(self):
        mes_models.DeviceCategory.objects.create(
            merchant=self.merchant,
            name="打印机",
            created_by=self.user,
            operator=self.employee,
        )
        mes_models.DeviceCategory.objects.create(
            merchant=self.merchant,
            name="滚筒机",
            created_by=self.user,
            operator=self.employee,
        )

        resp = self.client.get("/api/v2/mes/device-categories/?name=打印")

        self.assertEqual(resp.status_code, 200)
        self.assertEqual([item["name"] for item in resp.data], ["打印机"])

    def test_update_device_category(self):
        category = mes_models.DeviceCategory.objects.create(
            merchant=self.merchant,
            name="打印机",
            created_by=self.user,
            operator=self.employee,
        )

        resp = self.client.patch(f"/api/v2/mes/device-categories/{category.id}/", {"name": "热转印机"}, format="json")

        self.assertEqual(resp.status_code, 200)
        self.assertEqual(resp.data["name"], "热转印机")

    def test_delete_device_category_rejects_when_devices_exist(self):
        category = mes_models.DeviceCategory.objects.create(
            merchant=self.merchant,
            name="打印机",
            created_by=self.user,
            operator=self.employee,
        )
        mes_models.Device.objects.create(
            merchant=self.merchant,
            category=category,
            name="设备A",
            created_by=self.user,
            operator=self.employee,
        )

        resp = self.client.delete(f"/api/v2/mes/device-categories/{category.id}/")

        self.assertEqual(resp.status_code, 400)
        self.assertIn("不能删除", resp.data["detail"])

    def test_create_device(self):
        category = mes_models.DeviceCategory.objects.create(
            merchant=self.merchant,
            name="打印机",
            created_by=self.user,
            operator=self.employee,
        )

        resp = self.client.post(
            "/api/v2/mes/devices/",
            {
                "name": "设备A",
                "category": category.id,
                "peak_capacity": 120,
                "capacity_unit": 1,
                "extra": {"capacity": {"per_hour": 120}},
            },
            format="json",
        )

        self.assertEqual(resp.status_code, 201)
        self.assertEqual(resp.data["name"], "设备A")
        self.assertEqual(resp.data["category"], category.id)
        self.assertEqual(resp.data["category_name"], "打印机")
        self.assertEqual(resp.data['peak_capacity'], 120)
        self.assertEqual(resp.data['capacity_unit'], 1)
        self.assertEqual(resp.data['capacity_unit_label'], '米')
        self.assertEqual(resp.data["extra"], {"capacity": {"per_hour": 120}})

    def test_create_device_rejects_cross_merchant_category(self):
        category = mes_models.DeviceCategory.objects.create(
            merchant=self.other_merchant,
            name="外部分类",
            created_by=self.other_user,
            operator=self.other_employee,
        )

        resp = self.client.post(
            "/api/v2/mes/devices/",
            {"name": "设备A", "category": category.id, "peak_capacity": 120},
            format="json",
        )

        self.assertEqual(resp.status_code, 400)
        self.assertIn("category", resp.data)

    def test_list_devices_supports_category_filter(self):
        category = mes_models.DeviceCategory.objects.create(
            merchant=self.merchant,
            name="打印机",
            created_by=self.user,
            operator=self.employee,
        )
        other_category = mes_models.DeviceCategory.objects.create(
            merchant=self.merchant,
            name="滚筒机",
            created_by=self.user,
            operator=self.employee,
        )
        target = mes_models.Device.objects.create(
            merchant=self.merchant,
            category=category,
            name="设备A",
            peak_capacity=120,
            created_by=self.user,
            operator=self.employee,
        )
        mes_models.Device.objects.create(
            merchant=self.merchant,
            category=other_category,
            name="设备B",
            created_by=self.user,
            operator=self.employee,
        )

        resp = self.client.get(f"/api/v2/mes/devices/?category={category.id}")

        self.assertEqual(resp.status_code, 200)
        self.assertEqual([item["id"] for item in resp.data], [target.id])

    def test_update_device_supports_clearing_extra(self):
        category = mes_models.DeviceCategory.objects.create(
            merchant=self.merchant,
            name="打印机",
            created_by=self.user,
            operator=self.employee,
        )
        device = mes_models.Device.objects.create(
            merchant=self.merchant,
            category=category,
            name="设备A",
            peak_capacity=120,
            extra={"power": 220},
            created_by=self.user,
            operator=self.employee,
        )

        resp = self.client.patch(
            f"/api/v2/mes/devices/{device.id}/",
            {"extra": None, "capacity_unit": 2, "peak_capacity": 150},
            format="json",
        )

        self.assertEqual(resp.status_code, 200)
        self.assertEqual(resp.data['peak_capacity'], 150)
        self.assertEqual(resp.data['capacity_unit'], 2)
        self.assertEqual(resp.data['capacity_unit_label'], '码')
        self.assertIsNone(resp.data["extra"])

    def test_list_devices_supports_extra_json_filter(self):
        category = mes_models.DeviceCategory.objects.create(
            merchant=self.merchant,
            name="打印机",
            created_by=self.user,
            operator=self.employee,
        )
        target = mes_models.Device.objects.create(
            merchant=self.merchant,
            category=category,
            name="设备A",
            peak_capacity=120,
            extra={"capacity": {"per_hour": 120}},
            created_by=self.user,
            operator=self.employee,
        )
        mes_models.Device.objects.create(
            merchant=self.merchant,
            category=category,
            name="设备B",
            peak_capacity=80,
            extra={"capacity": {"per_hour": 80}},
            created_by=self.user,
            operator=self.employee,
        )

        resp = self.client.get('/api/v2/mes/devices/?extra_path=capacity.per_hour&extra_value=120')

        self.assertEqual(resp.status_code, 200)
        self.assertEqual([item['id'] for item in resp.data], [target.id])

    def test_list_devices_rejects_incomplete_extra_json_filter(self):
        resp = self.client.get('/api/v2/mes/devices/?extra_path=capacity.per_hour')

        self.assertEqual(resp.status_code, 400)
        self.assertIn('detail', resp.data)

    def test_cross_merchant_device_detail_returns_404(self):
        category = mes_models.DeviceCategory.objects.create(
            merchant=self.other_merchant,
            name="外部分类",
            created_by=self.other_user,
            operator=self.other_employee,
        )
        device = mes_models.Device.objects.create(
            merchant=self.other_merchant,
            category=category,
            name="外部设备",
            peak_capacity=120,
            created_by=self.other_user,
            operator=self.other_employee,
        )

        resp = self.client.get(f"/api/v2/mes/devices/{device.id}/")

        self.assertEqual(resp.status_code, 404)

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

        resp = self.client.post(
            '/api/v2/mes/production-assignments/',
            {
                'device': device.id,
                'content_type': content_type.id,
                'object_id': printing_order.id,
                'assigner': self.employee.id,
                'production_quantity': 300,
                'extra': {'batch': 'A1'},
            },
            format='json',
        )

        self.assertEqual(resp.status_code, 201)
        self.assertEqual(resp.data['device'], device.id)
        self.assertEqual(resp.data['production_quantity'], 300)
        self.assertEqual(resp.data['status'], mes_models.ProductionAssignmentStatusEnum.DRAFT)
        self.assertEqual(resp.data['status_label'], 'Draft')

    def test_update_production_assignment_rejects_cancel_after_accepted(self):
        customer = basic_models.Customer.objects.create(
            merchant=self.merchant,
            name='测试客户2',
            mobile='13800138003',
        )
        printing_order = PrintingOrder.objects.create(
            merchant=self.merchant,
            customer=customer,
            fabric='测试面料',
            width='150cm',
            created_by=self.user,
        )
        category = mes_models.DeviceCategory.objects.create(
            merchant=self.merchant,
            name='打印机',
            created_by=self.user,
            operator=self.employee,
        )
        device = mes_models.Device.objects.create(
            merchant=self.merchant,
            category=category,
            name='设备A',
            peak_capacity=120,
            created_by=self.user,
            operator=self.employee,
        )
        content_type = ContentType.objects.get_for_model(PrintingOrder)
        assignment = mes_models.ProductionAssignment.objects.create(
            merchant=self.merchant,
            device=device,
            content_type=content_type,
            object_id=printing_order.id,
            assigner=self.employee,
            assignee=self.employee,
            production_quantity=300,
            status=mes_models.ProductionAssignmentStatusEnum.ACCEPTED,
            created_by=self.user,
            operator=self.employee,
        )

        resp = self.client.patch(
            f'/api/v2/mes/production-assignments/{assignment.id}/',
            {'status': mes_models.ProductionAssignmentStatusEnum.CANCELLED},
            format='json',
        )

        self.assertEqual(resp.status_code, 400)
        self.assertEqual(resp.data['detail'], '当前状态不允许变更为目标状态')

    def test_list_production_assignments_supports_status_filter(self):
        customer = basic_models.Customer.objects.create(
            merchant=self.merchant,
            name='测试客户3',
            mobile='13800138004',
        )
        printing_order = PrintingOrder.objects.create(
            merchant=self.merchant,
            customer=customer,
            fabric='测试面料',
            width='150cm',
            created_by=self.user,
        )
        category = mes_models.DeviceCategory.objects.create(
            merchant=self.merchant,
            name='打印机',
            created_by=self.user,
            operator=self.employee,
        )
        device = mes_models.Device.objects.create(
            merchant=self.merchant,
            category=category,
            name='设备A',
            peak_capacity=120,
            created_by=self.user,
            operator=self.employee,
        )
        content_type = ContentType.objects.get_for_model(PrintingOrder)
        target = mes_models.ProductionAssignment.objects.create(
            merchant=self.merchant,
            device=device,
            content_type=content_type,
            object_id=printing_order.id,
            assigner=self.employee,
            production_quantity=300,
            status=mes_models.ProductionAssignmentStatusEnum.PUBLISHED,
            created_by=self.user,
            operator=self.employee,
        )
        mes_models.ProductionAssignment.objects.create(
            merchant=self.merchant,
            device=device,
            content_type=content_type,
            object_id=printing_order.id,
            assigner=self.employee,
            production_quantity=100,
            status=mes_models.ProductionAssignmentStatusEnum.DRAFT,
            created_by=self.user,
            operator=self.employee,
        )

        resp = self.client.get(
            f'/api/v2/mes/production-assignments/?status={mes_models.ProductionAssignmentStatusEnum.PUBLISHED}'
        )

        self.assertEqual(resp.status_code, 200)
        self.assertEqual([item['id'] for item in resp.data], [target.id])