diff --git a/api_v1/urls.py b/api_v1/urls.py index da28afb..06d68bc 100644 --- a/api_v1/urls.py +++ b/api_v1/urls.py @@ -2,7 +2,7 @@ from django.urls import path, include from rest_framework.routers import DefaultRouter from .views import stock_change_views, user_info, inventory, product_image, stateflow from .views.stock_change_views.snapshot import StockSnapshotListView -from printing.views import PrintingOrderViewSet +from printing.views import PrintingOrderViewSet, PrintingJobViewSet # 创建 DRF Router for Stateflow stateflow_router = DefaultRouter() @@ -12,6 +12,7 @@ stateflow_router.register(r'processes', stateflow.ProcessViewSet, basename='proc # 创建主 Router main_router = DefaultRouter() main_router.register(r'printing-orders', PrintingOrderViewSet, basename='printing-order') +main_router.register(r'printing-jobs', PrintingJobViewSet, basename='printing-job') urlpatterns = [ # 库存变动相关API diff --git a/api_v1/views/printing/test_printing_job_api.py b/api_v1/views/printing/test_printing_job_api.py new file mode 100644 index 0000000..b2cae99 --- /dev/null +++ b/api_v1/views/printing/test_printing_job_api.py @@ -0,0 +1,420 @@ +""" +PrintingJob API 测试 +""" +from django.test import TestCase +from rest_framework.test import APIClient +from rest_framework import status +from django.contrib.auth import get_user_model +from django.contrib.auth.models import Permission +from basic_info import models as basic_models +from printing import models as printing_models + +User = get_user_model() + + +class PrintingJobAPITestCase(TestCase): + """测试 PrintingJob API""" + + def setUp(self): + self.client = APIClient() + + # 创建商户 + self.merchant = basic_models.Merchant.objects.create( + name='测试印花厂', + type=basic_models.MerchantTypeEnum.FACTORY + ) + + # 创建用户 + self.user = User.objects.create_user( + username='testuser', + password='testpass123', + email='test@example.com' + ) + + # 创建员工并关联商户 + self.employee = basic_models.Employee.objects.create( + sys_user=self.user, + merchant=self.merchant, + name='测试员工', + mobile='13800138000', + job_type=basic_models.EmployeeTypeEnum.PRINTER, + status=basic_models.EmployeeStatusEnum.ACTIVE + ) + + # 创建客户 + self.customer = basic_models.Customer.objects.create( + merchant=self.merchant, + name='测试客户', + mobile='13900139000', + area='测试地区' + ) + + # 创建印染订单 + self.printing_order = printing_models.PrintingOrder.objects.create( + customer=self.customer, + fabric='测试布料', + width='150cm' + ) + + # 创建产品类别 + self.category = basic_models.ProductCategory.objects.create( + merchant=self.merchant, + name='测试类别' + ) + + # 创建产品 + self.product = basic_models.Product.objects.create( + merchant=self.merchant, + category=self.category, + name='测试产品', + human_id='TEST001', + unit=basic_models.ProductUnitEnum.METER + ) + + # 认证用户 + self.client.force_authenticate(user=self.user) + + # 给用户添加基础权限 + view_perm = Permission.objects.get(codename='view_printingjob') + add_perm = Permission.objects.get(codename='add_printingjob') + change_perm = Permission.objects.get(codename='change_printingjob') + self.user.user_permissions.add(view_perm, add_perm, change_perm) + + def test_create_printing_job(self): + """测试创建印染款式明细""" + data = { + 'printing_order': self.printing_order.id, + 'product': self.product.id, + 'quantity': 100, + 'unit': '米', + 'size': '50*60', + 'pieces': 10, + 'description': '测试备注' + } + + response = self.client.post('/api/v1/printing-jobs/', data, format='json') + self.assertEqual(response.status_code, status.HTTP_201_CREATED) + + # 验证创建成功 + job = printing_models.PrintingJob.objects.filter( + printing_order=self.printing_order, + product=self.product + ).first() + self.assertIsNotNone(job) + self.assertEqual(job.quantity, 100) + self.assertEqual(job.unit, '米') + self.assertEqual(job.pieces, 10) + + def test_list_printing_jobs(self): + """测试获取款式明细列表""" + # 创建测试数据 + printing_models.PrintingJob.objects.create( + printing_order=self.printing_order, + product=self.product, + quantity=100, + unit='米', + size='50*60', + pieces=10 + ) + printing_models.PrintingJob.objects.create( + printing_order=self.printing_order, + product=self.product, + quantity=200, + unit='米', + size='60*70', + pieces=20 + ) + + response = self.client.get('/api/v1/printing-jobs/') + self.assertEqual(response.status_code, status.HTTP_200_OK) + self.assertEqual(len(response.data), 2) + + def test_retrieve_printing_job(self): + """测试获取款式明细详情""" + job = printing_models.PrintingJob.objects.create( + printing_order=self.printing_order, + product=self.product, + quantity=100, + unit='米', + size='50*60', + pieces=10, + description='详情测试' + ) + + response = self.client.get(f'/api/v1/printing-jobs/{job.id}/') + self.assertEqual(response.status_code, status.HTTP_200_OK) + self.assertEqual(response.data['quantity'], 100) + self.assertEqual(response.data['unit'], '米') + self.assertIn('product_name', response.data) + self.assertEqual(response.data['product_name'], self.product.name) + + def test_update_printing_job(self): + """测试更新款式明细""" + job = printing_models.PrintingJob.objects.create( + printing_order=self.printing_order, + product=self.product, + quantity=100, + unit='米', + size='50*60', + pieces=10 + ) + + update_data = { + 'printing_order': self.printing_order.id, + 'product': self.product.id, + 'quantity': 200, + 'unit': '码', + 'size': '60*70', + 'pieces': 20, + 'description': '更新后的备注' + } + + response = self.client.put( + f'/api/v1/printing-jobs/{job.id}/', + update_data, + format='json' + ) + self.assertEqual(response.status_code, status.HTTP_200_OK) + + job.refresh_from_db() + self.assertEqual(job.quantity, 200) + self.assertEqual(job.unit, '码') + self.assertEqual(job.pieces, 20) + + def test_partial_update_printing_job(self): + """测试部分更新款式明细""" + job = printing_models.PrintingJob.objects.create( + printing_order=self.printing_order, + product=self.product, + quantity=100, + unit='米', + size='50*60', + pieces=10 + ) + + patch_data = { + 'quantity': 150, + 'pieces': 15 + } + + response = self.client.patch( + f'/api/v1/printing-jobs/{job.id}/', + patch_data, + format='json' + ) + self.assertEqual(response.status_code, status.HTTP_200_OK) + + job.refresh_from_db() + self.assertEqual(job.quantity, 150) + self.assertEqual(job.pieces, 15) + self.assertEqual(job.unit, '米') # 未修改字段保持不变 + + def test_delete_printing_job_forbidden(self): + """测试删除款式明细被禁用""" + # 添加删除权限以便测试destroy方法的自定义逻辑 + delete_perm = Permission.objects.get(codename='delete_printingjob') + self.user.user_permissions.add(delete_perm) + + job = printing_models.PrintingJob.objects.create( + printing_order=self.printing_order, + product=self.product, + quantity=100, + unit='米', + size='50*60', + pieces=10 + ) + + response = self.client.delete(f'/api/v1/printing-jobs/{job.id}/') + self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED) + self.assertIn('不支持删除', response.data['detail']) + + # 验证明细仍然存在 + self.assertTrue( + printing_models.PrintingJob.objects.filter(id=job.id).exists() + ) + + def test_filter_by_printing_order(self): + """测试按印染订单过滤""" + order2 = printing_models.PrintingOrder.objects.create( + customer=self.customer, + fabric='其他布料', + width='160cm' + ) + + printing_models.PrintingJob.objects.create( + printing_order=self.printing_order, + product=self.product, + quantity=100, + unit='米', + size='50*60', + pieces=10 + ) + printing_models.PrintingJob.objects.create( + printing_order=order2, + product=self.product, + quantity=200, + unit='米', + size='60*70', + pieces=20 + ) + + response = self.client.get(f'/api/v1/printing-jobs/?printing_order={self.printing_order.id}') + self.assertEqual(response.status_code, status.HTTP_200_OK) + self.assertEqual(len(response.data), 1) + self.assertEqual(response.data[0]['printing_order'], self.printing_order.id) + + def test_filter_by_product(self): + """测试按产品过滤""" + product2 = basic_models.Product.objects.create( + merchant=self.merchant, + category=self.category, + name='产品2', + human_id='TEST002', + unit=basic_models.ProductUnitEnum.METER + ) + + printing_models.PrintingJob.objects.create( + printing_order=self.printing_order, + product=self.product, + quantity=100, + unit='米', + size='50*60', + pieces=10 + ) + printing_models.PrintingJob.objects.create( + printing_order=self.printing_order, + product=product2, + quantity=200, + unit='米', + size='60*70', + pieces=20 + ) + + response = self.client.get(f'/api/v1/printing-jobs/?product={self.product.id}') + self.assertEqual(response.status_code, status.HTTP_200_OK) + self.assertEqual(len(response.data), 1) + self.assertEqual(response.data[0]['product'], self.product.id) + + def test_filter_by_quantity_range(self): + """测试按数量范围过滤""" + printing_models.PrintingJob.objects.create( + printing_order=self.printing_order, + product=self.product, + quantity=50, + unit='米', + size='50*60', + pieces=10 + ) + printing_models.PrintingJob.objects.create( + printing_order=self.printing_order, + product=self.product, + quantity=150, + unit='米', + size='60*70', + pieces=20 + ) + printing_models.PrintingJob.objects.create( + printing_order=self.printing_order, + product=self.product, + quantity=250, + unit='米', + size='70*80', + pieces=30 + ) + + response = self.client.get('/api/v1/printing-jobs/?quantity_min=100&quantity_max=200') + self.assertEqual(response.status_code, status.HTTP_200_OK) + self.assertEqual(len(response.data), 1) + self.assertEqual(response.data[0]['quantity'], 150) + + def test_search_by_product_name(self): + """测试按产品名称搜索""" + product2 = basic_models.Product.objects.create( + merchant=self.merchant, + category=self.category, + name='特殊产品', + human_id='TEST003', + unit=basic_models.ProductUnitEnum.METER + ) + + printing_models.PrintingJob.objects.create( + printing_order=self.printing_order, + product=self.product, + quantity=100, + unit='米', + size='50*60', + pieces=10 + ) + printing_models.PrintingJob.objects.create( + printing_order=self.printing_order, + product=product2, + quantity=200, + unit='米', + size='60*70', + pieces=20 + ) + + response = self.client.get('/api/v1/printing-jobs/?search=特殊') + self.assertEqual(response.status_code, status.HTTP_200_OK) + self.assertEqual(len(response.data), 1) + self.assertIn('特殊', response.data[0]['product_name']) + + def test_ordering(self): + """测试排序""" + job1 = printing_models.PrintingJob.objects.create( + printing_order=self.printing_order, + product=self.product, + quantity=100, + unit='米', + size='50*60', + pieces=10 + ) + job2 = printing_models.PrintingJob.objects.create( + printing_order=self.printing_order, + product=self.product, + quantity=200, + unit='米', + size='60*70', + pieces=20 + ) + + # 按数量升序 + response = self.client.get('/api/v1/printing-jobs/?ordering=quantity') + self.assertEqual(response.status_code, status.HTTP_200_OK) + self.assertEqual(response.data[0]['quantity'], 100) + self.assertEqual(response.data[1]['quantity'], 200) + + # 按数量降序 + response = self.client.get('/api/v1/printing-jobs/?ordering=-quantity') + self.assertEqual(response.data[0]['quantity'], 200) + self.assertEqual(response.data[1]['quantity'], 100) + + def test_validate_quantity_positive(self): + """测试数量必须大于0""" + data = { + 'printing_order': self.printing_order.id, + 'product': self.product.id, + 'quantity': 0, # 无效数量 + 'unit': '米', + 'size': '50*60', + 'pieces': 10 + } + + response = self.client.post('/api/v1/printing-jobs/', data, format='json') + self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) + self.assertIn('quantity', response.data) + + def test_validate_pieces_positive(self): + """测试件数必须大于0""" + data = { + 'printing_order': self.printing_order.id, + 'product': self.product.id, + 'quantity': 100, + 'unit': '米', + 'size': '50*60', + 'pieces': 0 # 无效件数 + } + + response = self.client.post('/api/v1/printing-jobs/', data, format='json') + self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) + self.assertIn('pieces', response.data) diff --git a/basic_info/admin.py b/basic_info/admin.py index 9e9b3f4..05d5cd6 100644 --- a/basic_info/admin.py +++ b/basic_info/admin.py @@ -25,9 +25,10 @@ class AdminBase(admin.ModelAdmin): return result def save_model(self, request, obj, form, change): - if not change: + if not change and isinstance(obj, models.Employee) is False: if not obj.merchant_id: - obj.merchant_id = request.user.employee.merchant.id + if request.user.is_superuser is False: + obj.merchant_id = request.user.employee.merchant.id return super().save_model(request, obj, form, change) diff --git a/printing/serializers.py b/printing/serializers.py index 546707f..084c18c 100644 --- a/printing/serializers.py +++ b/printing/serializers.py @@ -57,3 +57,68 @@ class PrintingOrderCreateUpdateSerializer(serializers.ModelSerializer): if not value: raise serializers.ValidationError("客户不能为空") return value + + +class PrintingJobListSerializer(serializers.ModelSerializer): + """印染款式明细列表序列化器""" + printing_order_id = serializers.CharField(source='printing_order.human_id', read_only=True) + product_name = serializers.CharField(source='product.name', read_only=True) + + class Meta: + model = models.PrintingJob + fields = [ + 'id', 'printing_order', 'printing_order_id', 'product', 'product_name', + 'quantity', 'unit', 'size', 'pieces', 'description', + 'created_at', 'updated_at' + ] + read_only_fields = ['id', 'created_at', 'updated_at'] + + +class PrintingJobDetailSerializer(serializers.ModelSerializer): + """印染款式明细详情序列化器""" + printing_order_id = serializers.CharField(source='printing_order.human_id', read_only=True) + product_name = serializers.CharField(source='product.name', read_only=True) + product_code = serializers.CharField(source='product.human_id', read_only=True) + + class Meta: + model = models.PrintingJob + fields = [ + 'id', 'printing_order', 'printing_order_id', 'product', 'product_name', 'product_code', + 'quantity', 'unit', 'size', 'pieces', 'description', + 'created_at', 'updated_at' + ] + read_only_fields = ['id', 'created_at', 'updated_at'] + + +class PrintingJobCreateUpdateSerializer(serializers.ModelSerializer): + """印染款式明细创建/更新序列化器""" + + class Meta: + model = models.PrintingJob + fields = [ + 'printing_order', 'product', 'quantity', 'unit', 'size', 'pieces', 'description' + ] + + def validate_printing_order(self, value): + """验证印染订单是否存在""" + if not value: + raise serializers.ValidationError("印染订单不能为空") + return value + + def validate_product(self, value): + """验证产品是否存在""" + if not value: + raise serializers.ValidationError("产品不能为空") + return value + + def validate_quantity(self, value): + """验证数量""" + if value <= 0: + raise serializers.ValidationError("数量必须大于0") + return value + + def validate_pieces(self, value): + """验证件数""" + if value <= 0: + raise serializers.ValidationError("件数必须大于0") + return value diff --git a/printing/views.py b/printing/views.py index bd7f923..d0f031f 100644 --- a/printing/views.py +++ b/printing/views.py @@ -16,6 +16,9 @@ from printing.serializers import ( PrintingOrderListSerializer, PrintingOrderDetailSerializer, PrintingOrderCreateUpdateSerializer, + PrintingJobListSerializer, + PrintingJobDetailSerializer, + PrintingJobCreateUpdateSerializer, ) @@ -225,3 +228,76 @@ class PrintingOrderViewSet(viewsets.ModelViewSet): 'detail': '已标记布料已收', 'data': serializer.data }) + + +class PrintingJobFilterSet(django_filters.FilterSet): + """印染款式明细过滤器""" + printing_order = django_filters.NumberFilter() + product = django_filters.NumberFilter() + product_name = django_filters.CharFilter(field_name='product__name', lookup_expr='icontains') + unit = django_filters.CharFilter(lookup_expr='icontains') + quantity_min = django_filters.NumberFilter(field_name='quantity', lookup_expr='gte') + quantity_max = django_filters.NumberFilter(field_name='quantity', lookup_expr='lte') + pieces_min = django_filters.NumberFilter(field_name='pieces', lookup_expr='gte') + pieces_max = django_filters.NumberFilter(field_name='pieces', lookup_expr='lte') + + class Meta: + model = models.PrintingJob + fields = ['printing_order', 'product'] + + +class PrintingJobViewSet(viewsets.ModelViewSet): + """ + 印染款式明细 ViewSet + + 提供印染款式明细的增改查功能(不支持删除) + + list: 获取款式明细列表 + retrieve: 获取款式明细详情 + create: 创建款式明细 + update: 更新款式明细 + partial_update: 部分更新款式明细 + + 查询参数: + - printing_order: 印染订单ID + - product: 产品ID + - product_name: 产品名称(模糊查询) + - unit: 单位(模糊查询) + - quantity_min: 最小数量 + - quantity_max: 最大数量 + - pieces_min: 最小件数 + - pieces_max: 最大件数 + - search: 全文搜索(产品名称、单位、尺寸、备注) + - ordering: 排序字段 + """ + queryset = models.PrintingJob.objects.all() + permission_classes = [DjangoModelPermissions, IsPrintingFactory] + pagination_class = LimitOffsetPagination + filter_backends = [DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter] + filterset_class = PrintingJobFilterSet + search_fields = ['product__name', 'unit', 'size', 'description'] + ordering_fields = ['id', 'created_at', 'updated_at', 'quantity', 'pieces'] + ordering = ['-created_at'] + + def get_serializer_class(self): + """根据动作选择序列化器""" + if self.action == 'list': + return PrintingJobListSerializer + elif self.action in ['create', 'update', 'partial_update']: + return PrintingJobCreateUpdateSerializer + else: # retrieve + return PrintingJobDetailSerializer + + def get_queryset(self): + """优化查询""" + queryset = super().get_queryset() + if self.action in ['list', 'retrieve']: + queryset = queryset.select_related('printing_order', 'product') + return queryset + + def destroy(self, request, *args, **kwargs): + """禁用删除操作""" + return Response( + {'detail': '印染款式明细不支持删除操作'}, + status=status.HTTP_405_METHOD_NOT_ALLOWED + )