1
0
forked from erp-dev/erp

fix: binding merchant and employee at admin site

This commit is contained in:
2025-11-12 17:48:57 +08:00
parent 808e5699f9
commit 22f7b6cea4
5 changed files with 566 additions and 3 deletions

View File

@@ -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

View File

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

View File

@@ -25,8 +25,9 @@ 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:
if request.user.is_superuser is False:
obj.merchant_id = request.user.employee.merchant.id
return super().save_model(request, obj, form, change)

View File

@@ -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

View File

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