""" PrintingOrder API 测试 """ from datetime import datetime, timezone as dt_timezone from django.test import TestCase from django.db import connection from django.conf import settings from django.utils import timezone 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 from stateflow import models as stateflow_models User = get_user_model() class PrintingOrderAPITestCase(TestCase): """测试 PrintingOrder API""" def setUp(self): # 清除缓存，避免 cache_page 装饰器导致的测试干扰 from django.core.cache import cache cache.clear() 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', status=basic_models.EmployeeStatusEnum.ACTIVE ) # 创建客户（设置 created_by 以便客户可见性过滤） self.customer = basic_models.Customer.objects.create( merchant=self.merchant, name='测试客户', mobile='13900139000', area='测试地区', created_by=self.employee ) # 创建流程 self.state1 = stateflow_models.State.objects.create(name='待印染') self.state2 = stateflow_models.State.objects.create(name='印染中') self.state3 = stateflow_models.State.objects.create(name='已完成') self.process = stateflow_models.Process.objects.create(name='印染流程') self.process.replace_nodes([self.state1, self.state2, self.state3]) # 设置默认流程 settings.PRINTING_DEFAULT_PROCESS_ID = self.process.id # 认证用户 self.client.force_authenticate(user=self.user) # 给用户添加基础权限 view_perm = Permission.objects.get(codename='view_printingorder') add_perm = Permission.objects.get(codename='add_printingorder') change_perm = Permission.objects.get(codename='change_printingorder') # 添加 view_all 权限以便测试 API 功能（权限过滤逻辑在专门的测试类中验证） view_all_perm = Permission.objects.get(codename='view_all_printingorders') self.user.user_permissions.add(view_perm, add_perm, change_perm, view_all_perm) def test_create_printing_order(self): """测试创建印染订单""" data = { 'customer': self.customer.id, 'fabric': '纯棉布料', 'width': '150cm', 'is_urgent': True, 'area': '广州', 'address': '白云区xxx', 'fabric_source': '客户提供', 'is_fabric_received': False, 'craft': '活性印花', 'description': '测试订单描述', 'outgoing_date': '2025-11-20', 'kd_riqi': '2025-11-19 13:14:15', 'printing_warn': '注意颜色', 'rolling_warn': '注意温度', 'production_warn': '质量检查' } response = self.client.post('/api/v1/printing-orders/', data, format='json') self.assertEqual(response.status_code, status.HTTP_201_CREATED) # 检查响应数据 self.assertIn('fabric', response.data) self.assertEqual(response.data['fabric'], '纯棉布料') self.assertEqual(response.data['is_urgent'], True) # 验证订单已创建 - 通过fabric查找，因为响应可能不包含id order = printing_models.PrintingOrder.objects.filter(fabric='纯棉布料').first() self.assertIsNotNone(order) self.assertEqual(order.customer.id, self.customer.id) self.assertEqual(order.fabric, '纯棉布料') # 验证 merchant 自动绑定 self.assertEqual(order.merchant.id, self.merchant.id) # 验证 outgoing_date 接受日期字符串并归一为 00:00:00（按当前时区） self.assertIsNotNone(order.outgoing_date) local_dt = timezone.localtime(order.outgoing_date) self.assertEqual(str(local_dt.date()), '2025-11-20') self.assertEqual(local_dt.hour, 0) self.assertEqual(local_dt.minute, 0) self.assertIsNotNone(order.kd_riqi) local_kd_riqi = timezone.localtime(order.kd_riqi) self.assertEqual(local_kd_riqi.isoformat(), '2025-11-19T13:14:15+08:00') def test_filter_outgoing_date_to_includes_whole_day(self): """ outgoing_date_to 传 YYYY-MM-DD 时应包含整天： - outgoing_date < 次日 00:00:00 """ tz = timezone.get_current_timezone() d0 = timezone.make_aware(datetime(2025, 11, 20, 23, 0, 0), tz) d1 = timezone.make_aware(datetime(2025, 11, 21, 0, 0, 0), tz) printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric='布料A', width='150cm', outgoing_date=d0, ) printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric='布料B', width='150cm', outgoing_date=d1, ) resp = self.client.get('/api/v1/printing-orders/?limit=50&offset=0&outgoing_date_to=2025-11-20') self.assertEqual(resp.status_code, status.HTTP_200_OK) fabrics = [row['fabric'] for row in resp.data['results']] self.assertIn('布料A', fabrics) self.assertNotIn('布料B', fabrics) def test_list_printing_orders(self): """测试获取订单列表""" # 创建测试订单 printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric='布料1', width='150cm' ) printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric='布料2', width='160cm' ) response = self.client.get('/api/v1/printing-orders/') self.assertEqual(response.status_code, status.HTTP_200_OK) # 分页响应格式 self.assertEqual(response.data['count'], 2) def test_list_pagination_keeps_rows_with_dangling_customer_reference(self): """历史脏数据 customer_id 悬空时，列表分页不应被 INNER JOIN 推空。""" base = timezone.now() dangling = printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric='悬空客户订单', width='150cm', ) middle = printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric='中间订单', width='150cm', ) oldest = printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric='最旧订单', width='150cm', ) printing_models.PrintingOrder.objects.filter(id=dangling.id).update( created_at=base + timezone.timedelta(minutes=2), ) printing_models.PrintingOrder.objects.filter(id=middle.id).update( created_at=base + timezone.timedelta(minutes=1), ) printing_models.PrintingOrder.objects.filter(id=oldest.id).update(created_at=base) missing_customer_id = 99999999 with connection.cursor() as cursor: cursor.execute( 'UPDATE printing_printingorder SET customer_id = %s WHERE id = %s', [missing_customer_id, dangling.id], ) try: response = self.client.get( '/api/v1/printing-orders/?limit=1&offset=2&ordering=-created_at', ) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data['count'], 3) self.assertEqual(len(response.data['results']), 1) self.assertEqual(response.data['results'][0]['id'], oldest.id) finally: with connection.cursor() as cursor: cursor.execute( 'UPDATE printing_printingorder SET customer_id = %s WHERE id = %s', [self.customer.id, dangling.id], ) def test_list_printing_orders_default_excludes_invalid(self): """默认不返回作废订单；显式传 is_invalid=true 时可查询作废订单""" printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric='正常订单', width='150cm', is_invalid=False, ) printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric='作废订单', width='150cm', is_invalid=True, ) response = self.client.get('/api/v1/printing-orders/') self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data['count'], 1) self.assertEqual(response.data['results'][0]['is_invalid'], False) response = self.client.get('/api/v1/printing-orders/?is_invalid=true') self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data['count'], 1) self.assertEqual(response.data['results'][0]['is_invalid'], True) def test_filter_by_external_order_id_and_serializer_fields(self): """支持按 external_order_id 过滤，且列表/详情返回外部字段""" order1 = printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric='布料A', width='150cm', external_order_id='KD20410611', external_customer_id='KH00991', external_customer_name='李泽柔', external_employee_name='丽容', external_raw={ 'first_record': { 'BianHaoKD': '7.07', 'KdRiQi': '2026-07-06T21:48:58Z', } }, ) printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric='布料B', width='160cm', external_order_id='KD20419999', ) response = self.client.get('/api/v1/printing-orders/?external_order_id=KD20410611') self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data['count'], 1) item = response.data['results'][0] self.assertEqual(item['external_order_id'], 'KD20410611') self.assertEqual(item['external_customer_id'], 'KH00991') self.assertEqual(item['external_customer_name'], '李泽柔') self.assertEqual(item['external_employee_name'], '丽容') self.assertEqual(item['bianhao_kd'], '7.07') self.assertEqual(item['kd_riqi'], '2026-07-06T21:48:58+08:00') detail = self.client.get(f'/api/v1/printing-orders/{order1.id}/') self.assertEqual(detail.status_code, status.HTTP_200_OK) self.assertEqual(detail.data['external_order_id'], 'KD20410611') self.assertEqual(detail.data['external_customer_id'], 'KH00991') self.assertEqual(detail.data['bianhao_kd'], '7.07') self.assertEqual(detail.data['kd_riqi'], '2026-07-06T21:48:58+08:00') def test_bianhao_kd_returns_null_without_external_order_id(self): order = printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric='普通订单', width='150cm', external_raw={'first_record': {'BianHaoKD': '7.07'}}, ) response = self.client.get(f'/api/v1/printing-orders/{order.id}/') self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertIsNone(response.data['bianhao_kd']) self.assertIsNone(response.data['kd_riqi']) def test_filter_and_order_by_kd_riqi(self): older = printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric='旧开单时间', width='150cm', external_order_id='KD-OLDER', kd_riqi=datetime(2026, 7, 1, 10, 0, tzinfo=dt_timezone.utc), ) newer = printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric='新开单时间', width='150cm', external_order_id='KD-NEWER', kd_riqi=datetime(2026, 7, 2, 10, 0, tzinfo=dt_timezone.utc), ) printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric='范围外开单时间', width='150cm', external_order_id='KD-OUTSIDE', kd_riqi=datetime(2026, 7, 3, 10, 0, tzinfo=dt_timezone.utc), ) response = self.client.get( '/api/v1/printing-orders/?kd_riqi_from=2026-07-01&kd_riqi_to=2026-07-02&ordering=kd_riqi' ) self.assertEqual(response.status_code, status.HTTP_200_OK) result_ids = [item['id'] for item in response.data['results']] self.assertEqual(result_ids, [older.id, newer.id]) self.assertEqual(response.data['results'][0]['kd_riqi'], '2026-07-01T18:00:00+08:00') def test_order_by_kd_riqi_puts_nulls_last(self): null_order = printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric='空开单时间', width='150cm', external_order_id='KD-NULL', ) older = printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric='旧开单时间', width='150cm', external_order_id='KD-OLDER-NULLS-LAST', kd_riqi=datetime(2026, 7, 1, 10, 0, tzinfo=dt_timezone.utc), ) newer = printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric='新开单时间', width='150cm', external_order_id='KD-NEWER-NULLS-LAST', kd_riqi=datetime(2026, 7, 2, 10, 0, tzinfo=dt_timezone.utc), ) asc_response = self.client.get('/api/v1/printing-orders/?ordering=kd_riqi&limit=3') desc_response = self.client.get('/api/v1/printing-orders/?ordering=-kd_riqi&limit=3') self.assertEqual(asc_response.status_code, status.HTTP_200_OK) self.assertEqual(desc_response.status_code, status.HTTP_200_OK) self.assertEqual([item['id'] for item in asc_response.data['results']], [older.id, newer.id, null_order.id]) self.assertEqual([item['id'] for item in desc_response.data['results']], [newer.id, older.id, null_order.id]) def test_retrieve_printing_order(self): """测试获取订单详情""" order = printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric='测试布料', width='150cm', craft='活性印花', description='详情测试' ) response = self.client.get(f'/api/v1/printing-orders/{order.id}/') self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data['fabric'], '测试布料') self.assertEqual(response.data['craft'], '活性印花') self.assertIn('customer_name', response.data) self.assertEqual(response.data['customer_name'], self.customer.name) def test_get_printing_order_by_external_order_id(self): """测试通过 external_order_id 获取订单详情""" order = printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric='通过外部编号查询', width='150cm', external_order_id='KD20410611', ) response = self.client.get('/api/v1/printing-orders/by-external-order-id/KD20410611/') self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data['id'], order.id) self.assertEqual(response.data['external_order_id'], 'KD20410611') self.assertEqual(response.data['fabric'], '通过外部编号查询') def test_update_printing_order(self): """测试更新订单""" order = printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric='旧布料', width='150cm' ) update_data = { 'customer': self.customer.id, 'fabric': '新布料', 'width': '160cm', 'is_urgent': True, 'craft': '更新的工艺' } response = self.client.put( f'/api/v1/printing-orders/{order.id}/', update_data, format='json' ) self.assertEqual(response.status_code, status.HTTP_200_OK) order.refresh_from_db() self.assertEqual(order.fabric, '新布料') self.assertEqual(order.width, '160cm') self.assertEqual(order.is_urgent, True) def test_partial_update_printing_order(self): """测试部分更新订单""" order = printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric='原布料', width='150cm', is_urgent=False ) patch_data = { 'is_urgent': True, 'craft': '新工艺', 'kd_riqi': '2025-11-21 09:30:00', } response = self.client.patch( f'/api/v1/printing-orders/{order.id}/', patch_data, format='json' ) self.assertEqual(response.status_code, status.HTTP_200_OK) order.refresh_from_db() self.assertEqual(order.is_urgent, True) self.assertEqual(order.craft, '新工艺') self.assertEqual(timezone.localtime(order.kd_riqi).isoformat(), '2025-11-21T09:30:00+08:00') self.assertEqual(order.fabric, '原布料') # 未修改字段保持不变 def test_delete_printing_order_forbidden(self): """测试删除订单被禁用""" # 添加删除权限以便测试destroy方法的自定义逻辑 delete_perm = Permission.objects.get(codename='delete_printingorder') self.user.user_permissions.add(delete_perm) order = printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric='测试布料', width='150cm' ) response = self.client.delete(f'/api/v1/printing-orders/{order.id}/') self.assertEqual(response.status_code, status.HTTP_405_METHOD_NOT_ALLOWED) self.assertIn('不支持删除', response.data['detail']) # 验证订单仍然存在 self.assertTrue( printing_models.PrintingOrder.objects.filter(id=order.id).exists() ) def test_invalidate_printing_order(self): """测试作废订单""" # 添加作废权限 invalidate_perm = Permission.objects.get(codename='can_invalidate_printingorder') self.user.user_permissions.add(invalidate_perm) order = printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric='测试布料', width='150cm', is_invalid=False ) response = self.client.post(f'/api/v1/printing-orders/{order.id}/invalidate/') self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertIn('已作废', response.data['detail']) order.refresh_from_db() self.assertTrue(order.is_invalid) def test_invalidate_without_permission(self): """测试没有权限时作废订单失败""" order = printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric='测试布料', width='150cm' ) response = self.client.post(f'/api/v1/printing-orders/{order.id}/invalidate/') self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) self.assertIn('没有权限', response.data['detail']) def test_activate_printing_order(self): """测试恢复订单""" # 添加恢复权限 activate_perm = Permission.objects.get(codename='can_activate_printingorder') self.user.user_permissions.add(activate_perm) order = printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric='测试布料', width='150cm', is_invalid=True ) response = self.client.post(f'/api/v1/printing-orders/{order.id}/activate/') self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertIn('已恢复', response.data['detail']) order.refresh_from_db() self.assertFalse(order.is_invalid) def test_activate_without_permission(self): """测试没有权限时恢复订单失败""" order = printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric='测试布料', width='150cm', is_invalid=True ) response = self.client.post(f'/api/v1/printing-orders/{order.id}/activate/') self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN) self.assertIn('没有权限', response.data['detail']) def test_mark_fabric_received(self): """测试标记布料已收""" order = printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric='测试布料', width='150cm', is_fabric_received=False ) response = self.client.post(f'/api/v1/printing-orders/{order.id}/mark_fabric_received/') self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertIn('已标记布料已收', response.data['detail']) order.refresh_from_db() self.assertTrue(order.is_fabric_received) def test_mark_fabric_received_already_received(self): """测试重复标记布料已收""" order = printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric='测试布料', width='150cm', is_fabric_received=True ) response = self.client.post(f'/api/v1/printing-orders/{order.id}/mark_fabric_received/') self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn('已经标记', response.data['detail']) def _create_product_for_job(self, name='测试产品'): category = basic_models.ProductCategory.objects.create( merchant=self.merchant, name=f'{name}类别', product_prefix='JOB', ) return basic_models.Product.objects.create( merchant=self.merchant, category=category, name=name, unit=basic_models.ProductUnitEnum.METER, ) def test_printing_order_ids_returns_job_ids(self): """测试获取订单下所有 PrintingJob ID。""" order = printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric='测试布料', width='150cm', ) product = self._create_product_for_job() job1 = printing_models.PrintingJob.objects.create( merchant=self.merchant, printing_order=order, product=product, quantity=10, unit='米', ) job2 = printing_models.PrintingJob.objects.create( merchant=self.merchant, printing_order=order, product=product, quantity=20, unit='米', ) response = self.client.get(f'/api/v1/printing-orders/{order.id}/ids/') self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data, {'ids': [job1.id, job2.id]}) def test_printing_order_ids_returns_empty_list_when_no_jobs(self): """订单存在但没有任务时返回空数组。""" order = printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric='测试布料', width='150cm', ) response = self.client.get(f'/api/v1/printing-orders/{order.id}/ids/') self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data, {'ids': []}) def test_printing_order_ids_returns_404_when_order_missing(self): """订单不存在时返回 404。""" response = self.client.get('/api/v1/printing-orders/99999999/ids/') self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND) def test_filter_by_customer(self): """测试按客户过滤""" customer2 = basic_models.Customer.objects.create( merchant=self.merchant, name='客户2', mobile='13900139001', created_by=self.employee ) printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric='布料1', width='150cm' ) printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=customer2, fabric='布料2', width='160cm' ) response = self.client.get(f'/api/v1/printing-orders/?customer={self.customer.id}') self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data['count'], 1) def test_filter_by_urgent(self): """测试按紧急状态过滤""" printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric='布料1', width='150cm', is_urgent=True ) printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric='布料2', width='160cm', is_urgent=False ) response = self.client.get('/api/v1/printing-orders/?is_urgent=true') self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data['count'], 1) self.assertTrue(response.data['results'][0]['is_urgent']) def test_search_by_fabric(self): """测试按布料搜索""" printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric='纯棉布料', width='150cm' ) printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric='涤纶布料', width='160cm' ) response = self.client.get('/api/v1/printing-orders/?search=纯棉') self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data['count'], 1) self.assertIn('纯棉', response.data['results'][0]['fabric']) def test_ordering(self): """测试排序""" order1 = printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric='布料1', width='150cm' ) order2 = printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric='布料2', width='160cm' ) # 按 ID 升序 response = self.client.get('/api/v1/printing-orders/?ordering=id') self.assertEqual(response.status_code, status.HTTP_200_OK) results = response.data['results'] self.assertEqual(results[0]['id'], order1.id) self.assertEqual(results[1]['id'], order2.id) # 按 ID 降序 response = self.client.get('/api/v1/printing-orders/?ordering=-id') results = response.data['results'] self.assertEqual(results[0]['id'], order2.id) self.assertEqual(results[1]['id'], order1.id) def test_human_id_generation(self): """测试 human_id 自动生成""" from datetime import date order = printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric='测试布料', width='150cm' ) # human_id 格式: YYYYMMDD000001 self.assertIsNotNone(order.human_id) self.assertTrue(len(order.human_id) >= 14) today = date.today().strftime('%Y%m%d') self.assertTrue(order.human_id.startswith(today)) def test_create_order_with_default_process(self): """测试创建订单使用默认流程""" data = { 'customer': self.customer.id, 'fabric': '测试布料', 'width': '150cm', } response = self.client.post('/api/v1/printing-orders/', data, format='json') self.assertEqual(response.status_code, status.HTTP_201_CREATED) # 验证使用了默认流程 order = printing_models.PrintingOrder.objects.get(id=response.data['id']) self.assertEqual(order.process.id, self.process.id) def test_create_order_with_custom_process(self): """测试创建订单指定自定义流程""" custom_process = stateflow_models.Process.objects.create(name='自定义流程') data = { 'customer': self.customer.id, 'fabric': '测试布料', 'width': '150cm', 'process': custom_process.id, } response = self.client.post('/api/v1/printing-orders/', data, format='json') self.assertEqual(response.status_code, status.HTTP_201_CREATED) order = printing_models.PrintingOrder.objects.get(id=response.data['id']) self.assertEqual(order.process.id, custom_process.id) def test_order_progress_in_list(self): """测试订单列表包含进度字段""" order = printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric='测试布料', width='150cm', process=self.process, ) response = self.client.get('/api/v1/printing-orders/') self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertIn('progress', response.data['results'][0]) self.assertEqual(response.data['results'][0]['progress'], 0) # 没有任务时为0 def test_update_process_when_no_jobs(self): """测试没有任务时可以修改流程""" order = printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric='测试布料', width='150cm', process=self.process, ) new_process = stateflow_models.Process.objects.create(name='新流程') data = { 'process': new_process.id, } response = self.client.patch( f'/api/v1/printing-orders/{order.id}/', data, format='json' ) self.assertEqual(response.status_code, status.HTTP_200_OK) order.refresh_from_db() self.assertEqual(order.process.id, new_process.id) def test_update_process_with_jobs_not_started_relinks_jobs(self): """测试：有任务但均未开始时允许改流程，并对所有 jobs 重建/绑定新的 business_object""" order = printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric='测试布料', width='150cm', process=self.process, ) category = basic_models.ProductCategory.objects.create( name='测试类别2', merchant=self.merchant, ) product = basic_models.Product.objects.create( name='测试产品2', category=category, merchant=self.merchant, ) job = printing_models.PrintingJob.objects.create( printing_order=order, product=product, quantity=10, unit='件', ) # 绑定一个旧的 business_object（未推进过，未开始） old_bo = stateflow_models.BusinessObject.objects.create( name=f'PrintingJob-{job.id}', process=self.process, ) job.business_object = old_bo job.save() new_process = stateflow_models.Process.objects.create(name='新流程2') data = {'process': new_process.id} response = self.client.patch(f'/api/v1/printing-orders/{order.id}/', data, format='json') self.assertEqual(response.status_code, status.HTTP_200_OK) order.refresh_from_db() self.assertEqual(order.process.id, new_process.id) job.refresh_from_db() self.assertIsNotNone(job.business_object) self.assertNotEqual(job.business_object.id, old_bo.id) self.assertEqual(job.business_object.process.id, new_process.id) def test_cannot_update_process_when_job_started(self): """测试有已开始的任务时不能修改流程""" order = printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric='测试布料', width='150cm', process=self.process, ) # 创建产品类别和产品 category = basic_models.ProductCategory.objects.create( name='测试类别', merchant=self.merchant, ) product = basic_models.Product.objects.create( name='测试产品', category=category, merchant=self.merchant, ) # 创建任务 job = printing_models.PrintingJob.objects.create( printing_order=order, product=product, quantity=10, unit='件', size='100x200', pieces=5, ) # 创建 BusinessObject 并推进状态 business_object = stateflow_models.BusinessObject.objects.create( name=f'PrintingJob-{job.id}', process=self.process, ) job.business_object = business_object job.save() from stateflow.services import advance_to_next_state advance_to_next_state(business_object, self.user) # 尝试修改流程 new_process = stateflow_models.Process.objects.create(name='新流程') data = { 'process': new_process.id, } response = self.client.patch( f'/api/v1/printing-orders/{order.id}/', data, format='json' ) self.assertEqual(response.status_code, status.HTTP_400_BAD_REQUEST) self.assertIn('已开始', str(response.data)) def test_jobs_status_summary_in_list(self): """测试订单列表包含 jobs_status_summary 字段（PrintingJob 状态汇总）""" # 创建订单 order = printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric='测试布料', width='150cm', process=self.process, ) # 创建产品类别和产品 category = basic_models.ProductCategory.objects.create( name='测试类别', merchant=self.merchant, ) product = basic_models.Product.objects.create( name='测试产品', category=category, merchant=self.merchant, ) # 创建 3 个任务 jobs = [] for i in range(3): business_object = stateflow_models.BusinessObject.objects.create( name=f'PrintingJob-test-{i}', process=self.process, ) job = printing_models.PrintingJob.objects.create( printing_order=order, product=product, quantity=10, unit='件', business_object=business_object, ) jobs.append(job) # 推进第一个任务的状态（完成第一个节点） from stateflow.services import advance_to_next_state advance_to_next_state(jobs[0].business_object, self.user) # 推进第二个任务的状态两次（完成前两个节点） advance_to_next_state(jobs[1].business_object, self.user) advance_to_next_state(jobs[1].business_object, self.user) # 第三个任务保持初始状态（待印染） # 获取订单列表 response = self.client.get('/api/v1/printing-orders/') self.assertEqual(response.status_code, status.HTTP_200_OK) # 检查 jobs_status_summary 字段存在 self.assertIn('jobs_status_summary', response.data['results'][0]) summary = response.data['results'][0]['jobs_status_summary'] # 检查返回格式（列表） self.assertIsInstance(summary, list) # 汇总结果应该包含不同状态的计数 # - job0: 完成了第一个节点，下一个待执行是 "印染中" # - job1: 完成了前两个节点，下一个待执行是 "已完成" # - job2: 未开始，下一个待执行是 "待印染" state_names = {item['state_name']: item['count'] for item in summary} # 验证有3个任务被正确统计 total_count = sum(item['count'] for item in summary) self.assertEqual(total_count, 3) # 验证返回的每个项都包含必要字段 for item in summary: self.assertIn('state_name', item) self.assertIn('state_id', item) self.assertIn('count', item) def test_jobs_status_summary_empty_when_no_jobs(self): """测试没有任务时 jobs_status_summary 为空列表""" order = printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer, fabric='无任务订单', width='150cm', process=self.process, ) # 获取特定订单详情（避免被其他测试数据影响） response = self.client.get(f'/api/v1/printing-orders/{order.id}/') self.assertEqual(response.status_code, status.HTTP_200_OK) # 详情接口不包含 jobs_status_summary（这是列表专用字段） # 改用列表接口并按特定条件过滤 response = self.client.get('/api/v1/printing-orders/', {'fabric': '无任务订单'}) self.assertEqual(response.status_code, status.HTTP_200_OK) self.assertEqual(response.data['count'], 1) summary = response.data['results'][0]['jobs_status_summary'] self.assertEqual(summary, []) class CustomerVisibilityFilterTestCase(TestCase): """测试客户可见性过滤""" def setUp(self): from django.core.cache import cache cache.clear() self.client = APIClient() # 创建商户 self.merchant = basic_models.Merchant.objects.create( name='测试印花厂', type=basic_models.MerchantTypeEnum.FACTORY ) # 创建第二个商户（用于测试 merchant 隔离） self.merchant2 = 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', status=basic_models.EmployeeStatusEnum.ACTIVE ) # 创建第二个员工（同商户，用于测试客户可见性） self.user2 = User.objects.create_user( username='testuser2', password='testpass123', email='test2@example.com' ) self.employee2 = basic_models.Employee.objects.create( sys_user=self.user2, merchant=self.merchant, name='测试员工2', mobile='13800138001', status=basic_models.EmployeeStatusEnum.ACTIVE ) # 创建客户（employee 创建的，employee 可见） self.customer_by_emp1 = basic_models.Customer.objects.create( merchant=self.merchant, name='员工1的客户', mobile='13900139000', created_by=self.employee ) # 创建客户（employee2 创建的，默认 employee 不可见） self.customer_by_emp2 = basic_models.Customer.objects.create( merchant=self.merchant, name='员工2的客户', mobile='13900139001', created_by=self.employee2 ) # 创建 employee 可见客户的订单 self.order_visible = printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer_by_emp1, fabric='可见订单', width='150cm' ) # 创建 employee 不可见客户的订单 self.order_invisible = printing_models.PrintingOrder.objects.create( merchant=self.merchant, customer=self.customer_by_emp2, fabric='不可见订单', width='150cm' ) # 创建其他商户的订单（merchant 隔离测试） self.customer_other_merchant = basic_models.Customer.objects.create( merchant=self.merchant2, name='其他商户客户', mobile='13900139002' ) self.order_other_merchant = printing_models.PrintingOrder.objects.create( merchant=self.merchant2, customer=self.customer_other_merchant, fabric='其他商户订单', width='150cm' ) # 给用户添加基础权限（不含 view_all） view_perm = Permission.objects.get(codename='view_printingorder') self.user.user_permissions.add(view_perm) self.user2.user_permissions.add(view_perm) self.client.force_authenticate(user=self.user) def test_list_only_visible_orders(self): """测试普通员工只能看到可见客户的订单""" response = self.client.get('/api/v1/printing-orders/') self.assertEqual(response.status_code, status.HTTP_200_OK) # 应该只能看到自己创建的客户的订单 = 1 self.assertEqual(response.data['count'], 1) fabrics = [r['fabric'] for r in response.data['results']] self.assertIn('可见订单', fabrics) self.assertNotIn('不可见订单', fabrics) self.assertNotIn('其他商户订单', fabrics) def test_retrieve_visible_order(self): """测试可以获取可见订单详情""" response = self.client.get(f'/api/v1/printing-orders/{self.order_visible.id}/') self.assertEqual(response.status_code, status.HTTP_200_OK) def test_retrieve_invisible_order_404(self): """测试无法获取不可见订单详情""" response = self.client.get(f'/api/v1/printing-orders/{self.order_invisible.id}/') self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND) def test_retrieve_other_merchant_order_404(self): """测试无法获取其他商户的订单""" response = self.client.get(f'/api/v1/printing-orders/{self.order_other_merchant.id}/') self.assertEqual(response.status_code, status.HTTP_404_NOT_FOUND) def test_customer_in_visible_employees_can_see(self): """测试被加入 visible_employees 后可以看到订单""" # 将 employee 加入 customer_by_emp2 的可见员工列表 self.customer_by_emp2.visible_employees.add(self.employee) response = self.client.get('/api/v1/printing-orders/') self.assertEqual(response.status_code, status.HTTP_200_OK) # 现在应该能看到 2 个订单（自己创建的 + 被加入可见列表的） self.assertEqual(response.data['count'], 2) fabrics = [r['fabric'] for r in response.data['results']] self.assertIn('不可见订单', fabrics) def test_view_all_permission_bypasses_filter(self): """测试 view_all 权限可以突破客户可见性限制""" # 添加 view_all 权限 view_all_perm = Permission.objects.get(codename='view_all_printingorders') self.user.user_permissions.add(view_all_perm) response = self.client.get('/api/v1/printing-orders/') self.assertEqual(response.status_code, status.HTTP_200_OK) # 应该能看到本商户的所有订单（2个），但不能看到其他商户的 self.assertEqual(response.data['count'], 2) fabrics = [r['fabric'] for r in response.data['results']] self.assertIn('不可见订单', fabrics) self.assertNotIn('其他商户订单', fabrics) def test_superuser_sees_all(self): """测试超级用户可以看到所有订单""" self.user.is_superuser = True self.user.save() response = self.client.get('/api/v1/printing-orders/') self.assertEqual(response.status_code, status.HTTP_200_OK) # 超级用户可以看到所有订单（包括其他商户） self.assertEqual(response.data['count'], 3)