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