from decimal import Decimal import datetime from django.contrib.auth import get_user_model from django.contrib.auth.models import Group, Permission from django.test import TestCase from django.utils import timezone from rest_framework.test import APIClient, APIRequestFactory from basic_info import models as basic_models from printing import models as printing_models from business import models as business_models from api_v2.views.printing import PrintingJobByCustomerView from django.contrib.contenttypes.models import ContentType from stateflow import models as stateflow_models class QuickCreateEmployeeUserAPITest(TestCase): def setUp(self): self.client = APIClient() self.merchant = basic_models.Merchant.objects.create( name='测试商户', type=basic_models.MerchantTypeEnum.FACTORY, ) self.admin_user = get_user_model().objects.create_user(username='admin', password='pass12345') self.admin_employee = basic_models.Employee.objects.create( merchant=self.merchant, sys_user=self.admin_user, name='管理员', ) self.client.force_authenticate(user=self.admin_user) self.url = '/api/v2/users/quick-create/' self.payload = { 'username': 'new_user', 'password': 'strongPass#1', 'display_name': '新员工', 'merchant_id': self.merchant.id, 'mobile': '13800000000', } def test_quick_create_employee_user_success(self): response = self.client.post(self.url, self.payload, format='json') self.assertEqual(response.status_code, 201) data = response.data self.assertIn('user', data) self.assertIn('employee', data) self.assertEqual(data['user']['username'], self.payload['username']) created_user = get_user_model().objects.get(username=self.payload['username']) self.assertEqual(created_user.employee.merchant, self.merchant) self.assertEqual(created_user.employee.name, self.payload['display_name']) def test_quick_create_employee_user_duplicate_username(self): get_user_model().objects.create_user(username=self.payload['username'], password='pass12345') response = self.client.post(self.url, self.payload, format='json') self.assertEqual(response.status_code, 400) self.assertIn('用户名已存在', str(response.data)) def test_quick_create_employee_user_invalid_merchant(self): payload = {**self.payload, 'merchant_id': 9999} response = self.client.post(self.url, payload, format='json') self.assertEqual(response.status_code, 400) self.assertIn('商户不存在', str(response.data)) def test_quick_create_employee_user_with_role(self): """测试创建用户时指定角色""" role = Group.objects.create(name='管理员') payload = {**self.payload, 'role_id': role.id} response = self.client.post(self.url, payload, format='json') self.assertEqual(response.status_code, 201) self.assertEqual(response.data['user']['role_id'], role.id) created_user = get_user_model().objects.get(username=self.payload['username']) self.assertIn(role, created_user.groups.all()) def test_quick_create_employee_user_with_invalid_role(self): """测试创建用户时指定不存在的角色""" payload = {**self.payload, 'role_id': 99999} response = self.client.post(self.url, payload, format='json') self.assertEqual(response.status_code, 400) self.assertIn('角色不存在', str(response.data)) class RoleListV2APITest(TestCase): def setUp(self): self.client = APIClient() self.user = get_user_model().objects.create_user(username='test_user', password='pass12345') self.client.force_authenticate(user=self.user) self.url = '/api/v2/roles/' def test_role_list_returns_all_roles(self): """测试获取所有角色""" role1 = Group.objects.create(name='管理员') role2 = Group.objects.create(name='普通用户') response = self.client.get(self.url) self.assertEqual(response.status_code, 200) self.assertEqual(len(response.data), 2) names = [r['name'] for r in response.data] self.assertIn('管理员', names) self.assertIn('普通用户', names) # 检查返回字段只有 id 和 name for role in response.data: self.assertEqual(set(role.keys()), {'id', 'name'}) def test_role_list_empty(self): """测试没有角色时返回空列表""" response = self.client.get(self.url) self.assertEqual(response.status_code, 200) self.assertEqual(response.data, []) def test_role_list_requires_authentication(self): """测试未认证用户无法访问""" client = APIClient() response = client.get(self.url) self.assertEqual(response.status_code, 401) class HealthCheckV2APITest(TestCase): def setUp(self): self.client = APIClient() def test_health_check_returns_200(self): resp = self.client.get('/api/v2/health/') self.assertEqual(resp.status_code, 200) self.assertEqual(resp.data.get('status'), 'ok') class PrintingJobByCustomerAPITest(TestCase): def setUp(self): self.factory = APIRequestFactory() self.client = APIClient() self.merchant = basic_models.Merchant.objects.create( name='印染商户', type=basic_models.MerchantTypeEnum.FACTORY, ) self.customer = basic_models.Customer.objects.create( merchant=self.merchant, name='客户X', created_by=None, ) category = basic_models.ProductCategory.objects.create( merchant=self.merchant, name='品类', product_prefix='FAB', ) self.product = basic_models.Product.objects.create( merchant=self.merchant, category=category, name='产品A', human_id='FAB-001', width_size=Decimal('150.00'), color='红色', unit=basic_models.ProductUnitEnum.METER, ) self.other_product = basic_models.Product.objects.create( merchant=self.merchant, category=category, name='产品B', human_id='FAB-002', width_size=Decimal('160.00'), color='蓝色', unit=basic_models.ProductUnitEnum.METER, ) self.printing_order = printing_models.PrintingOrder.objects.create( customer=self.customer, fabric='棉', width='150cm', ) self.printing_order_other = printing_models.PrintingOrder.objects.create( customer=self.customer, fabric='麻', width='160cm', ) tz = timezone.get_default_timezone() in_range = timezone.make_aware(datetime.datetime(2025, 12, 5, 10, 0, 0), tz) out_range = timezone.make_aware(datetime.datetime(2025, 11, 20, 10, 0, 0), tz) self.job_in_range = printing_models.PrintingJob.objects.create( printing_order=self.printing_order, product=self.product, quantity=10, unit='米', ) printing_models.PrintingJob.objects.filter(id=self.job_in_range.id).update(created_at=in_range) self.job_out_range = printing_models.PrintingJob.objects.create( printing_order=self.printing_order_other, product=self.other_product, quantity=20, unit='米', ) printing_models.PrintingJob.objects.filter(id=self.job_out_range.id).update(created_at=out_range) # 关联销售单，验证 billed_quantity warehouse = basic_models.WareHouse.objects.create( merchant=self.merchant, name='仓库A', mode=basic_models.WareHouseModeEnum.UNRESTRICTED, ) operator = basic_models.Employee.objects.create( merchant=self.merchant, name='操作员', ) sales_order = business_models.SalesOrder.objects.create( merchant=self.merchant, customer=self.customer, sales_date=datetime.date(2025, 12, 5), operator=operator, warehouse=warehouse, ) business_models.SalesOrderItem.objects.create( sales_order=sales_order, product=self.product, price=Decimal('10'), quantity=Decimal('12.5'), unit='米', empty_diff_percent=Decimal('0'), num_of_rolls=1, printing_job=self.job_in_range, ) business_models.SalesOrderItem.objects.create( sales_order=sales_order, product=self.product, price=Decimal('8'), quantity=Decimal('7.5'), unit='米', empty_diff_percent=Decimal('0'), num_of_rolls=1, printing_job=self.job_in_range, ) self.view = PrintingJobByCustomerView.as_view() def _get(self, params): request = self.factory.get('/api/v2/printing/jobs/', params) return self.view(request) def test_basic_date_and_customer_filter(self): resp = self._get({ 'customer_id': self.customer.id, 'date_from': '2025-12-01', 'date_to': '2025-12-10', }) self.assertEqual(resp.status_code, 200) self.assertEqual(len(resp.data), 1) self.assertEqual(resp.data[0]['id'], self.job_in_range.id) self.assertEqual(resp.data[0]['billed_quantity'], '20.00') def test_filter_by_printing_order(self): resp = self._get({ 'customer_id': self.customer.id, 'date_from': '2025-12-01', 'date_to': '2025-12-10', 'printing_order': self.printing_order.id, }) self.assertEqual(len(resp.data), 1) self.assertEqual(resp.data[0]['printing_order'], self.printing_order.id) def test_filter_by_product_fields(self): resp = self._get({ 'customer_id': self.customer.id, 'date_from': '2025-12-01', 'date_to': '2025-12-10', 'product_id': self.product.id, 'product_name': '产品A', 'product_human_id': 'FAB-001', 'product_width_size': '150', 'product_color': '红', }) self.assertEqual(len(resp.data), 1) data = resp.data[0] self.assertEqual(data['product']['id'], self.product.id) self.assertEqual(data['billed_quantity'], '20.00') class PrintingJobBatchAdvanceV2APITest(TestCase): def setUp(self): self.client = APIClient() # 工厂用户（满足 IsPrintingFactory） self.merchant = basic_models.Merchant.objects.create( name='印染工厂', type=basic_models.MerchantTypeEnum.FACTORY, ) self.user = get_user_model().objects.create_user(username='factory_user', password='pass12345') self.employee = basic_models.Employee.objects.create( merchant=self.merchant, sys_user=self.user, name='工厂员工', ) self.client.force_authenticate(user=self.user) # 基础客户/产品 self.customer = basic_models.Customer.objects.create( merchant=self.merchant, name='客户A', created_by=None, ) category = basic_models.ProductCategory.objects.create( merchant=self.merchant, name='品类', product_prefix='FAB', ) self.product = basic_models.Product.objects.create( merchant=self.merchant, category=category, name='产品A', human_id='FAB-100', width_size=Decimal('150.00'), color='红色', unit=basic_models.ProductUnitEnum.METER, ) # stateflow：流程 + 3个节点 + 节点参数（第一步必填） self.state1 = stateflow_models.State.objects.create(name='待处理1', description='第一个节点') self.state2 = stateflow_models.State.objects.create(name='待处理2', description='第二个节点') self.state3 = stateflow_models.State.objects.create(name='待处理3', description='第三个节点') self.param_required = stateflow_models.StateParameter.objects.create( key='temperature', value='', is_required=True, description='温度（必填）', ) self.state1.parameters.add(self.param_required) self.process = stateflow_models.Process.objects.create(name='印染流程') stateflow_models.ProcessNode.objects.create(process=self.process, state=self.state1, order=0) stateflow_models.ProcessNode.objects.create(process=self.process, state=self.state2, order=1) stateflow_models.ProcessNode.objects.create(process=self.process, state=self.state3, order=2) # printing：订单 + 两条明细（每条明细都有 business_object） self.printing_order = printing_models.PrintingOrder.objects.create( customer=self.customer, fabric='棉', width='150cm', process=self.process, ) bo1 = stateflow_models.BusinessObject.objects.create( name='BO-1', process=self.process, ) bo2 = stateflow_models.BusinessObject.objects.create( name='BO-2', process=self.process, ) self.job1 = printing_models.PrintingJob.objects.create( printing_order=self.printing_order, product=self.product, quantity=10, unit='米', business_object=bo1, ) self.job2 = printing_models.PrintingJob.objects.create( printing_order=self.printing_order, product=self.product, quantity=20, unit='米', business_object=bo2, ) def test_preview_success_returns_next_state_and_parameters(self): resp = self.client.post( '/api/v2/printing-jobs/batch-advance/preview/', {'printing_job_ids': [self.job1.id, self.job2.id]}, format='json' ) self.assertEqual(resp.status_code, 200) self.assertEqual(resp.data['printing_order_id'], self.printing_order.id) self.assertEqual(set(resp.data['printing_job_ids']), {self.job1.id, self.job2.id}) next_state = resp.data['next_state'] self.assertEqual(next_state['id'], self.state1.id) self.assertEqual(next_state['order'], 0) self.assertTrue(any(p['key'] == 'temperature' and p['is_required'] for p in next_state['parameters'])) def test_preview_fails_when_jobs_not_same_printing_order(self): other_order = printing_models.PrintingOrder.objects.create( customer=self.customer, fabric='麻', width='160cm', process=self.process, ) bo3 = stateflow_models.BusinessObject.objects.create(name='BO-3', process=self.process) job3 = printing_models.PrintingJob.objects.create( printing_order=other_order, product=self.product, quantity=5, unit='米', business_object=bo3, ) resp = self.client.post( '/api/v2/printing-jobs/batch-advance/preview/', {'printing_job_ids': [self.job1.id, job3.id]}, format='json' ) self.assertEqual(resp.status_code, 400) self.assertIn('printing_order', resp.data.get('detail', '')) def test_submit_success_advances_all_and_creates_batch_record(self): resp = self.client.post( '/api/v2/printing-jobs/batch-advance/', { 'printing_job_ids': [self.job1.id, self.job2.id], 'parameters': {'temperature': '25.5'} }, format='json' ) self.assertEqual(resp.status_code, 200) self.assertIn('batch_id', resp.data) # stateflow：两条 business_object 都应生成 1 条完成日志，且完成的是 state1 self.job1.refresh_from_db() self.job2.refresh_from_db() # 同时应修复绑定：BusinessObject.content_type/object_id 指向 PrintingJob from django.contrib.contenttypes.models import ContentType ct_job = ContentType.objects.get_for_model(printing_models.PrintingJob) self.job1.business_object.refresh_from_db() self.job2.business_object.refresh_from_db() self.assertEqual(self.job1.business_object.content_type_id, ct_job.id) self.assertEqual(self.job1.business_object.object_id, self.job1.id) self.assertEqual(self.job2.business_object.content_type_id, ct_job.id) self.assertEqual(self.job2.business_object.object_id, self.job2.id) self.assertEqual(self.job1.business_object.state_logs.filter(is_cancelled=False).count(), 1) self.assertEqual(self.job2.business_object.state_logs.filter(is_cancelled=False).count(), 1) self.assertEqual(self.job1.business_object.state_logs.first().state_id, self.state1.id) self.assertEqual(self.job2.business_object.state_logs.first().state_id, self.state1.id) # printing：批量记录应存在且关联 jobs record = printing_models.PrintingJobBatchAdvanceRecord.objects.get(id=resp.data['batch_id']) self.assertEqual(record.created_by_id, self.user.id) self.assertEqual(record.printing_order_id, self.printing_order.id) self.assertEqual(record.state_id, self.state1.id) self.assertEqual(record.parameters.get('temperature'), '25.5') self.assertEqual(set(record.printing_jobs.values_list('id', flat=True)), {self.job1.id, self.job2.id}) def test_submit_fails_when_missing_required_parameter(self): resp = self.client.post( '/api/v2/printing-jobs/batch-advance/', { 'printing_job_ids': [self.job1.id, self.job2.id], 'parameters': {} # 缺少 temperature }, format='json' ) self.assertEqual(resp.status_code, 400) self.assertIn('缺失必填参数', resp.data.get('detail', '')) # 未产生任何日志/批量记录 self.assertEqual(self.job1.business_object.state_logs.count(), 0) self.assertEqual(self.job2.business_object.state_logs.count(), 0) self.assertEqual(printing_models.PrintingJobBatchAdvanceRecord.objects.count(), 0) class BusinessObjectCloneV2APITest(TestCase): def setUp(self): self.client = APIClient() self.user = get_user_model().objects.create_user(username='u1', password='pass12345') self.client.force_authenticate(user=self.user) self.state1 = stateflow_models.State.objects.create(name='S1') self.state2 = stateflow_models.State.objects.create(name='S2') self.process = stateflow_models.Process.objects.create(name='P1') stateflow_models.ProcessNode.objects.create(process=self.process, state=self.state1, order=0) stateflow_models.ProcessNode.objects.create(process=self.process, state=self.state2, order=1) self.process_src = stateflow_models.Process.objects.create(name='P-src') self.process_dst = stateflow_models.Process.objects.create(name='P-dst') ct = ContentType.objects.get_for_model(stateflow_models.Process) self.bo = stateflow_models.BusinessObject.objects.create( name='BO', process=self.process, content_type=ct, object_id=self.process_src.id, ) # 创建一条日志 + 参数记录，确保“克隆包含工艺参数” log = stateflow_models.StateFlowRecord.objects.create( business_object=self.bo, state=self.state1, completed_by=self.user, ) stateflow_models.StateLogParameterRecord.objects.create( state_log=log, parameters={'temperature': '25.5'}, remark='r1', ) def test_clone_business_object_api_returns_new_id(self): new_object_id = self.process_dst.id resp = self.client.post( '/api/v2/stateflow/business-objects/clone/', { 'business_object_id': self.bo.id, 'content_type': self.bo.content_type_id, 'object_id': new_object_id, }, format='json' ) self.assertEqual(resp.status_code, 200) self.assertIn('business_object_id', resp.data) new_id = resp.data['business_object_id'] self.assertNotEqual(new_id, self.bo.id) cloned = stateflow_models.BusinessObject.objects.get(id=new_id) self.assertEqual(cloned.process_id, self.bo.process_id) self.assertEqual(cloned.state_logs.count(), self.bo.state_logs.count()) self.assertEqual(cloned.content_type_id, self.bo.content_type_id) self.assertEqual(cloned.object_id, new_object_id) def test_clone_business_object_api_binds_to_plate_order_business_object(self): """ printing 工作流依赖：新建 PlateOrder 后通过 clone API 复制流程进度， 后端应自动把克隆后的 BusinessObject 绑定到目标 PlateOrder.business_object。 """ merchant = basic_models.Merchant.objects.create( name='M1', type=basic_models.MerchantTypeEnum.FACTORY, ) customer = basic_models.Customer.objects.create( merchant=merchant, name='C1', created_by=None, ) # source PlateOrder（会自动创建 business_object） source_po = printing_models.PlateOrder.objects.create( customer=customer, process=self.process.id, design_code='SRC', style_name='S', ) self.assertIsNotNone(source_po.business_object_id) # 给 source 写一条日志 + 参数，确保克隆后能看到“节点与参数” log = stateflow_models.StateFlowRecord.objects.create( business_object=source_po.business_object, state=self.state1, completed_by=self.user, ) stateflow_models.StateLogParameterRecord.objects.create( state_log=log, parameters={'temperature': '25.5'}, remark='r1', ) # target PlateOrder（创建时也会自动创建一个“空 business_object”） target_po = printing_models.PlateOrder.objects.create( customer=customer, process=self.process.id, design_code='DST', style_name='D', ) self.assertIsNotNone(target_po.business_object_id) old_target_bo_id = target_po.business_object_id ct_po = ContentType.objects.get_for_model(printing_models.PlateOrder) resp = self.client.post( '/api/v2/stateflow/business-objects/clone/', { 'business_object_id': source_po.business_object_id, 'content_type': ct_po.id, 'object_id': target_po.id, }, format='json' ) self.assertEqual(resp.status_code, 200, resp.data) new_bo_id = resp.data['business_object_id'] target_po.refresh_from_db() self.assertEqual(target_po.business_object_id, new_bo_id) # 目标 PlateOrder 的旧“空 BO”应被安全清理（避免残留多份绑定） self.assertFalse(stateflow_models.BusinessObject.objects.filter(id=old_target_bo_id).exists()) # 克隆后的日志与参数应存在 self.assertEqual(target_po.business_object.state_logs.count(), 1) cloned_log = target_po.business_object.state_logs.first() self.assertEqual(cloned_log.state_id, self.state1.id) self.assertEqual(cloned_log.parameter_records.count(), 1) self.assertEqual(cloned_log.parameter_records.first().parameters.get('temperature'), '25.5') def test_clone_business_object_api_404(self): resp = self.client.post( '/api/v2/stateflow/business-objects/clone/', { 'business_object_id': 999999, 'content_type': self.bo.content_type_id, 'object_id': 222, }, format='json' ) self.assertEqual(resp.status_code, 404) def test_clone_business_object_api_rejects_mismatched_content_type(self): # 用另一个 ContentType（比如 State） ct_state = ContentType.objects.get_for_model(stateflow_models.State) self.assertNotEqual(ct_state.id, self.bo.content_type_id) resp = self.client.post( '/api/v2/stateflow/business-objects/clone/', { 'business_object_id': self.bo.id, 'content_type': ct_state.id, 'object_id': 222, }, format='json' ) self.assertEqual(resp.status_code, 400) self.assertIn('content_type', resp.data.get('detail', '')) def test_clone_business_object_api_unauthorized(self): client = APIClient() resp = client.post( '/api/v2/stateflow/business-objects/clone/', { 'business_object_id': self.bo.id, 'content_type': self.bo.content_type_id, 'object_id': 222, }, format='json' ) self.assertEqual(resp.status_code, 401) class PlateOrderByProcessNodeV2APITest(TestCase): def setUp(self): self.client = APIClient() # 工厂用户（满足 IsPrintingFactory） self.merchant = basic_models.Merchant.objects.create( name='开版工厂', type=basic_models.MerchantTypeEnum.FACTORY, ) self.user = get_user_model().objects.create_user(username='plate_factory_user', password='pass12345') basic_models.Employee.objects.create( merchant=self.merchant, sys_user=self.user, name='开版员工', ) self.client.force_authenticate(user=self.user) self.customer = basic_models.Customer.objects.create( merchant=self.merchant, name='客户P', created_by=None, ) # stateflow：流程 + 3个节点 self.state1 = stateflow_models.State.objects.create(name='节点1') self.state2 = stateflow_models.State.objects.create(name='节点2') self.state3 = stateflow_models.State.objects.create(name='节点3') self.param2 = stateflow_models.StateParameter.objects.create(key='temperature', value='25') self.state2.parameters.add(self.param2) self.process = stateflow_models.Process.objects.create(name='开版流程') self.node1 = stateflow_models.ProcessNode.objects.create(process=self.process, state=self.state1, order=0) self.node2 = stateflow_models.ProcessNode.objects.create(process=self.process, state=self.state2, order=1) self.node3 = stateflow_models.ProcessNode.objects.create(process=self.process, state=self.state3, order=2) # PlateOrder：构造不同“下一待执行节点” self.po_state1 = printing_models.PlateOrder.objects.create( customer=self.customer, process=self.process.id, design_code='PO-A', style_name='款式A', ) self.po_state2_old = printing_models.PlateOrder.objects.create( customer=self.customer, process=self.process.id, design_code='CODE-X', # 不包含数字，便于测试 search=id style_name='款式B', ) stateflow_models.StateFlowRecord.objects.create( business_object=self.po_state2_old.business_object, state=self.state1, completed_by=self.user, is_cancelled=False, ) # 订单维度参数：模拟该订单曾在 state2 提交过参数，但该状态记录已撤销（因此当前仍处于 state2 待执行） cancelled_log = stateflow_models.StateFlowRecord.objects.create( business_object=self.po_state2_old.business_object, state=self.state2, completed_by=self.user, is_cancelled=True, ) stateflow_models.StateLogParameterRecord.objects.create( state_log=cancelled_log, parameters={'temperature': '30'}, ) self.po_state2_new = printing_models.PlateOrder.objects.create( customer=self.customer, process=self.process.id, design_code='ABC-999', style_name='款式C', ) stateflow_models.StateFlowRecord.objects.create( business_object=self.po_state2_new.business_object, state=self.state1, completed_by=self.user, is_cancelled=False, ) self.po_state3 = printing_models.PlateOrder.objects.create( customer=self.customer, process=self.process.id, design_code='PO-D', style_name='款式D', ) stateflow_models.StateFlowRecord.objects.create( business_object=self.po_state3.business_object, state=self.state1, completed_by=self.user, is_cancelled=False, ) stateflow_models.StateFlowRecord.objects.create( business_object=self.po_state3.business_object, state=self.state2, completed_by=self.user, is_cancelled=False, ) # 固定 created_at 用于排序断言 tz = timezone.get_default_timezone() t_old = timezone.make_aware(datetime.datetime(2025, 12, 1, 10, 0, 0), tz) t_new = timezone.make_aware(datetime.datetime(2025, 12, 2, 10, 0, 0), tz) printing_models.PlateOrder.objects.filter(id=self.po_state2_old.id).update(created_at=t_old) printing_models.PlateOrder.objects.filter(id=self.po_state2_new.id).update(created_at=t_new) self.url = '/api/v2/plate-orders/by-process-node/' def test_filter_by_process_node_returns_only_matching_next_pending(self): resp = self.client.get(self.url, {'process_node_id': self.node2.id}) self.assertEqual(resp.status_code, 200) self.assertEqual(resp.data['process_node']['id'], self.node2.id) self.assertEqual(resp.data['process_node']['state_id'], self.state2.id) self.assertTrue(any(p['key'] == 'temperature' and p['value'] == '25' for p in resp.data['parameters'])) ids = [item['id'] for item in resp.data['results']] self.assertEqual(set(ids), {self.po_state2_old.id, self.po_state2_new.id}) # results 必须包含“订单维度”的参数值（同一节点下不同订单可不同） by_id = {item['id']: item for item in resp.data['results']} self.assertIn('created_by', by_id[self.po_state2_old.id]) self.assertIn('created_by', by_id[self.po_state2_new.id]) self.assertIn('process_parameters', by_id[self.po_state2_old.id]) self.assertIn('process_parameters', by_id[self.po_state2_new.id]) # po_state2_old：来自已撤销的 state2 提交记录 self.assertTrue( any(p['key'] == 'temperature' and p['value'] == '30' for p in by_id[self.po_state2_old.id]['process_parameters']) ) # po_state2_new：从未提交过 state2 参数，value 应为 null self.assertTrue( any(p['key'] == 'temperature' and p['value'] is None for p in by_id[self.po_state2_new.id]['process_parameters']) ) def test_process_parameters_schema_and_keys_align_with_template(self): resp = self.client.get(self.url, {'process_node_id': self.node2.id}) self.assertEqual(resp.status_code, 200) template_keys = [p['key'] for p in resp.data['parameters']] self.assertGreater(len(template_keys), 0) for item in resp.data['results']: self.assertIn('process_parameters', item) self.assertIsInstance(item['process_parameters'], list) keys = [p.get('key') for p in item['process_parameters']] self.assertEqual(keys, template_keys) # 严格 schema：每项只能有 key/value 两个字段（便于前端做稳定类型定义） for p in item['process_parameters']: self.assertEqual(set(p.keys()), {'key', 'value'}) def test_when_state_has_no_parameters_returns_empty_process_parameters(self): # node1 对应 state1：该测试 setUp 中未配置任何参数 resp = self.client.get(self.url, {'process_node_id': self.node1.id}) self.assertEqual(resp.status_code, 200) self.assertEqual(resp.data['parameters'], []) self.assertEqual(len(resp.data['results']), 1) self.assertEqual(resp.data['results'][0]['id'], self.po_state1.id) self.assertEqual(resp.data['results'][0]['process_parameters'], []) def test_default_ordering_is_minus_created_at(self): resp = self.client.get(self.url, {'process_node_id': self.node2.id}) self.assertEqual(resp.status_code, 200) ids = [item['id'] for item in resp.data['results']] self.assertEqual(ids[0], self.po_state2_new.id) self.assertEqual(ids[1], self.po_state2_old.id) def test_ordering_param_created_at_asc(self): resp = self.client.get(self.url, {'process_node_id': self.node2.id, 'ordering': 'created_at'}) self.assertEqual(resp.status_code, 200) ids = [item['id'] for item in resp.data['results']] self.assertEqual(ids[0], self.po_state2_old.id) self.assertEqual(ids[1], self.po_state2_new.id) def test_search_supports_design_code_icontains(self): resp = self.client.get(self.url, {'process_node_id': self.node2.id, 'search': 'abc'}) self.assertEqual(resp.status_code, 200) ids = [item['id'] for item in resp.data['results']] self.assertEqual(ids, [self.po_state2_new.id]) def test_search_supports_pk_without_extra_param(self): resp = self.client.get(self.url, {'process_node_id': self.node2.id, 'search': str(self.po_state2_old.id)}) self.assertEqual(resp.status_code, 200) ids = [item['id'] for item in resp.data['results']] self.assertEqual(ids, [self.po_state2_old.id]) def test_pagination_limit_offset(self): resp = self.client.get(self.url, {'process_node_id': self.node2.id, 'limit': 1, 'offset': 0}) self.assertEqual(resp.status_code, 200) self.assertEqual(resp.data['count'], 2) self.assertEqual(len(resp.data['results']), 1) class PlateOrderByProcessV2APITest(TestCase): def setUp(self): self.client = APIClient() # 工厂用户（满足 IsPrintingFactory） self.merchant = basic_models.Merchant.objects.create( name='开版工厂2', type=basic_models.MerchantTypeEnum.FACTORY, ) self.user = get_user_model().objects.create_user(username='plate_process_user', password='pass12345') basic_models.Employee.objects.create( merchant=self.merchant, sys_user=self.user, name='开版员工2', ) self.client.force_authenticate(user=self.user) self.customer = basic_models.Customer.objects.create( merchant=self.merchant, name='客户PP', created_by=None, ) # stateflow：流程 + 3个节点（其中 2 个节点有参数，1 个无参数） 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.param_len = stateflow_models.StateParameter.objects.create(key='length', value=None) self.param_temp = stateflow_models.StateParameter.objects.create(key='temperature', value=None) self.state1.parameters.add(self.param_len) self.state2.parameters.add(self.param_temp) self.process = stateflow_models.Process.objects.create(name='开版流程2') self.node1 = stateflow_models.ProcessNode.objects.create(process=self.process, state=self.state1, order=0) self.node2 = stateflow_models.ProcessNode.objects.create(process=self.process, state=self.state2, order=1) self.node3 = stateflow_models.ProcessNode.objects.create(process=self.process, state=self.state3, order=2) # PlateOrder：同一个 process，不同 created_at self.po_no_exec = printing_models.PlateOrder.objects.create( customer=self.customer, process=self.process.id, design_code='CODE-X', # 不包含数字，便于测试 plate_order=pk style_name='款式1', ) self.po_exec = printing_models.PlateOrder.objects.create( customer=self.customer, process=self.process.id, design_code='PO-EXEC', style_name='款式2', ) # 执行 state1（非撤销）并提交参数 log1 = stateflow_models.StateFlowRecord.objects.create( business_object=self.po_exec.business_object, state=self.state1, completed_by=self.user, is_cancelled=False, ) stateflow_models.StateLogParameterRecord.objects.create( state_log=log1, parameters={'length': '5米'}, ) # 执行 state2（撤销）并提交参数（默认不带撤销参数 -> 应视为未执行） cancelled_log2 = stateflow_models.StateFlowRecord.objects.create( business_object=self.po_exec.business_object, state=self.state2, completed_by=self.user, is_cancelled=True, ) stateflow_models.StateLogParameterRecord.objects.create( state_log=cancelled_log2, parameters={'temperature': '30'}, ) self.po_out_of_range = printing_models.PlateOrder.objects.create( customer=self.customer, process=self.process.id, design_code='OUT', style_name='款式3', ) # 固定 created_at tz = timezone.get_default_timezone() t1 = timezone.make_aware(datetime.datetime(2025, 12, 1, 10, 0, 0), tz) t2 = timezone.make_aware(datetime.datetime(2025, 12, 2, 10, 0, 0), tz) t0 = timezone.make_aware(datetime.datetime(2025, 11, 20, 10, 0, 0), tz) printing_models.PlateOrder.objects.filter(id=self.po_no_exec.id).update(created_at=t1) printing_models.PlateOrder.objects.filter(id=self.po_exec.id).update(created_at=t2) printing_models.PlateOrder.objects.filter(id=self.po_out_of_range.id).update(created_at=t0) self.url = '/api/v2/plate-orders/by-process/' def test_list_by_process_and_date_range(self): resp = self.client.get( self.url, { 'process_id': self.process.id, 'date_from': '2025-12-01', 'date_to': '2025-12-02', } ) self.assertEqual(resp.status_code, 200) self.assertEqual(resp.data['process']['id'], self.process.id) ids = [item['id'] for item in resp.data['results']] self.assertEqual(set(ids), {self.po_no_exec.id, self.po_exec.id}) # 序列化器字段：created_by 必须存在（允许为 null） for item in resp.data['results']: self.assertIn('created_by', item) def test_default_ordering_is_minus_created_at(self): resp = self.client.get( self.url, { 'process_id': self.process.id, 'date_from': '2025-12-01', 'date_to': '2025-12-02', } ) self.assertEqual(resp.status_code, 200) ids = [item['id'] for item in resp.data['results']] self.assertEqual(ids[0], self.po_exec.id) self.assertEqual(ids[1], self.po_no_exec.id) def test_plate_order_param_supports_design_code_icontains(self): resp = self.client.get( self.url, { 'process_id': self.process.id, 'date_from': '2025-12-01', 'date_to': '2025-12-02', 'plate_order': 'exec', } ) self.assertEqual(resp.status_code, 200) ids = [item['id'] for item in resp.data['results']] self.assertEqual(ids, [self.po_exec.id]) def test_plate_order_param_supports_pk_without_extra_param(self): resp = self.client.get( self.url, { 'process_id': self.process.id, 'date_from': '2025-12-01', 'date_to': '2025-12-02', 'plate_order': str(self.po_no_exec.id), } ) self.assertEqual(resp.status_code, 200) ids = [item['id'] for item in resp.data['results']] self.assertEqual(ids, [self.po_no_exec.id]) def test_process_params_includes_all_nodes_and_is_executed(self): resp = self.client.get( self.url, { 'process_id': self.process.id, 'date_from': '2025-12-01', 'date_to': '2025-12-02', } ) self.assertEqual(resp.status_code, 200) by_id = {item['id']: item for item in resp.data['results']} po_exec = by_id[self.po_exec.id] po_no_exec = by_id[self.po_no_exec.id] # 每条 PlateOrder 都必须有 process_params，且包含所有节点（按 order） self.assertIn('process_params', po_exec) self.assertEqual([n['order'] for n in po_exec['process_params']], [0, 1, 2]) self.assertEqual([n['node_name'] for n in po_exec['process_params']], ['画图', '调色', '套样']) # po_exec：state1 已执行；state2 只有撤销记录 -> 默认视为未执行；state3 未执行 n1, n2, n3 = po_exec['process_params'] self.assertTrue(n1['is_executed']) self.assertFalse(n2['is_executed']) self.assertFalse(n3['is_executed']) self.assertEqual([p['key'] for p in n1['params']], ['length']) self.assertEqual(n1['params'][0]['value'], '5米') self.assertEqual([p['key'] for p in n2['params']], ['temperature']) self.assertIsNone(n2['params'][0]['value']) self.assertEqual(n3['params'], []) # po_no_exec：全部未执行，参数 value 全为 null self.assertEqual([n['order'] for n in po_no_exec['process_params']], [0, 1, 2]) self.assertFalse(po_no_exec['process_params'][0]['is_executed']) self.assertIsNone(po_no_exec['process_params'][0]['params'][0]['value']) def test_invalid_ordering_returns_400(self): resp = self.client.get( self.url, { 'process_id': self.process.id, 'date_from': '2025-12-01', 'date_to': '2025-12-02', 'ordering': 'unknown', } ) self.assertEqual(resp.status_code, 400) class PlateOrderByStateStatusV2APITest(TestCase): def setUp(self): self.client = APIClient() self.merchant = basic_models.Merchant.objects.create( name='开版工厂3', type=basic_models.MerchantTypeEnum.FACTORY, ) self.user = get_user_model().objects.create_user(username='state_status_user', password='pass12345') basic_models.Employee.objects.create( merchant=self.merchant, sys_user=self.user, name='开版员工3', ) self.client.force_authenticate(user=self.user) self.customer = basic_models.Customer.objects.create( merchant=self.merchant, name='客户SS', created_by=None, ) self.state_prepare = stateflow_models.State.objects.create(name='画图') self.state_target = stateflow_models.State.objects.create(name='调色') self.state_unlinked = stateflow_models.State.objects.create(name='未绑定') self.param_temp = stateflow_models.StateParameter.objects.create(key='temperature', value='25') self.state_target.parameters.add(self.param_temp) self.process = stateflow_models.Process.objects.create(name='主流程') self.node_prepare = stateflow_models.ProcessNode.objects.create(process=self.process, state=self.state_prepare, order=0) self.node_target = stateflow_models.ProcessNode.objects.create(process=self.process, state=self.state_target, order=1) self.other_process = stateflow_models.Process.objects.create(name='其他流程') stateflow_models.ProcessNode.objects.create(process=self.other_process, state=self.state_target, order=0) self.po_not_started = printing_models.PlateOrder.objects.create( customer=self.customer, process=self.process.id, design_code='PO-NOT', style_name='款式N', ) self.po_no_logs = printing_models.PlateOrder.objects.create( customer=self.customer, process=self.process.id, design_code='PO-ZERO', style_name='款式Z', ) self.po_completed_old = printing_models.PlateOrder.objects.create( customer=self.customer, process=self.process.id, design_code='PO-C-OLD', style_name='款式CO', ) self.po_completed_new = printing_models.PlateOrder.objects.create( customer=self.customer, process=self.process.id, design_code='PO-C-NEW', style_name='款式CN', ) self.po_cancelled = printing_models.PlateOrder.objects.create( customer=self.customer, process=self.process.id, design_code='PO-CAN', style_name='款式CA', ) self.po_other_process = printing_models.PlateOrder.objects.create( customer=self.customer, process=self.other_process.id, design_code='PO-OTHER', style_name='款式OP', ) self._complete_state(self.po_not_started, self.state_prepare) self._complete_state(self.po_completed_old, self.state_prepare) self._complete_state(self.po_completed_new, self.state_prepare) self._complete_state(self.po_cancelled, self.state_prepare) self._complete_state(self.po_completed_old, self.state_target, parameters={'temperature': '28'}) self._complete_state(self.po_completed_new, self.state_target, parameters={'temperature': '32'}) self._complete_state(self.po_cancelled, self.state_target, is_cancelled=True, parameters={'temperature': '99'}) self._complete_state(self.po_other_process, self.state_target, parameters={'temperature': '88'}) tz = timezone.get_default_timezone() t_old = timezone.make_aware(datetime.datetime(2025, 12, 1, 9, 0, 0), tz) t_new = timezone.make_aware(datetime.datetime(2025, 12, 3, 9, 0, 0), tz) printing_models.PlateOrder.objects.filter(id=self.po_completed_old.id).update(created_at=t_old) printing_models.PlateOrder.objects.filter(id=self.po_completed_new.id).update(created_at=t_new) self.url = '/api/v2/plate-orders/by-state-status/' def _complete_state(self, plate_order, state, *, is_cancelled=False, parameters=None): log = stateflow_models.StateFlowRecord.objects.create( business_object=plate_order.business_object, state=state, completed_by=self.user, is_cancelled=is_cancelled, ) if parameters is not None: stateflow_models.StateLogParameterRecord.objects.create( state_log=log, parameters=parameters, ) return log def test_completed_status_returns_sorted_plate_orders(self): resp = self.client.get( self.url, { 'process_id': self.process.id, 'state_id': self.state_target.id, 'status': 'completed', } ) self.assertEqual(resp.status_code, 200) self.assertEqual(resp.data['state']['id'], self.state_target.id) self.assertEqual(resp.data['status'], 'completed') ids = [item['id'] for item in resp.data['results']] self.assertEqual(ids, [self.po_completed_new.id, self.po_completed_old.id]) template_keys = [p['key'] for p in resp.data['state_parameters']] self.assertEqual(template_keys, ['temperature']) by_id = {item['id']: item for item in resp.data['results']} self.assertEqual(by_id[self.po_completed_new.id]['state_parameters'][0]['value'], '32') self.assertEqual(by_id[self.po_completed_old.id]['state_parameters'][0]['value'], '28') self.assertIsNotNone(by_id[self.po_completed_new.id]['state_log']) self.assertFalse(by_id[self.po_completed_new.id]['state_log']['is_cancelled']) def test_default_status_is_completed(self): resp = self.client.get( self.url, { 'process_id': self.process.id, 'state_id': self.state_target.id, } ) self.assertEqual(resp.status_code, 200) ids = [item['id'] for item in resp.data['results']] self.assertEqual(set(ids), {self.po_completed_new.id, self.po_completed_old.id}) def test_state_id_with_not_started_status_is_invalid(self): resp = self.client.get( self.url, { 'process_id': self.process.id, 'state_id': self.state_target.id, 'status': 'not_started', } ) self.assertEqual(resp.status_code, 400) def test_cancelled_status_returns_orders_with_cancelled_logs(self): resp = self.client.get( self.url, { 'process_id': self.process.id, 'state_id': self.state_target.id, 'status': 'cancelled', } ) self.assertEqual(resp.status_code, 200) self.assertEqual(resp.data['results'][0]['id'], self.po_cancelled.id) self.assertTrue(resp.data['results'][0]['state_log']['is_cancelled']) def test_invalid_process_state_pair_returns_400(self): resp = self.client.get( self.url, { 'process_id': self.process.id, 'state_id': self.state_unlinked.id, 'status': 'completed', } ) self.assertEqual(resp.status_code, 400) self.assertIn('state 不属于该 process', resp.data['detail']) def test_ordering_created_at_and_pagination(self): resp = self.client.get( self.url, { 'process_id': self.process.id, 'state_id': self.state_target.id, 'status': 'completed', 'ordering': 'created_at', 'limit': 1, 'offset': 1, } ) self.assertEqual(resp.status_code, 200) self.assertEqual(resp.data['count'], 2) self.assertEqual(len(resp.data['results']), 1) self.assertEqual(resp.data['results'][0]['id'], self.po_completed_new.id) def test_other_process_records_are_excluded(self): resp = self.client.get( self.url, { 'process_id': self.process.id, 'state_id': self.state_target.id, 'status': 'completed', } ) ids = [item['id'] for item in resp.data['results']] self.assertNotIn(self.po_other_process.id, ids) def test_completed_status_only_returns_latest_state(self): resp = self.client.get( self.url, { 'process_id': self.process.id, 'state_id': self.state_prepare.id, 'status': 'completed', } ) self.assertEqual(resp.status_code, 200) ids = [item['id'] for item in resp.data['results']] self.assertEqual(set(ids), {self.po_not_started.id, self.po_cancelled.id}) self.assertNotIn(self.po_completed_old.id, ids) self.assertNotIn(self.po_completed_new.id, ids) def test_without_state_id_returns_orders_without_any_logs(self): resp = self.client.get( self.url, { 'process_id': self.process.id, } ) self.assertEqual(resp.status_code, 200) self.assertEqual(resp.data['status'], 'not_started') self.assertIsNone(resp.data['state']) self.assertEqual(resp.data['state_parameters'], []) ids = [item['id'] for item in resp.data['results']] self.assertEqual(ids, [self.po_no_logs.id]) self.assertEqual(resp.data['results'][0]['state_parameters'], []) self.assertIsNone(resp.data['results'][0]['state_log']) def test_without_state_id_only_allows_not_started_status(self): resp = self.client.get( self.url, { 'process_id': self.process.id, 'status': 'completed', } ) self.assertEqual(resp.status_code, 400) self.assertIn('state_id 为空时仅支持 status=not_started', resp.data['detail']) class PlateOrderBatchUpdateV2APITest(TestCase): def setUp(self): self.client = APIClient() # 工厂用户（满足 IsPrintingFactory） self.merchant = basic_models.Merchant.objects.create( name='印染工厂', type=basic_models.MerchantTypeEnum.FACTORY, ) self.user = get_user_model().objects.create_user(username='factory_user2', password='pass12345') basic_models.Employee.objects.create( merchant=self.merchant, sys_user=self.user, name='工厂员工2', ) self.client.force_authenticate(user=self.user) # 赋予 change 权限（批量更新必需） perm_change = Permission.objects.get(codename='change_plateorder') self.user.user_permissions.add(perm_change) self.customer = basic_models.Customer.objects.create( merchant=self.merchant, name='客户A', created_by=None, ) self.designer = basic_models.Employee.objects.create( merchant=self.merchant, name='设计师', ) self.po1 = printing_models.PlateOrder.objects.create( customer=self.customer, design_code='D1', urgency_level='正常', style_name='S1', ) self.po2 = printing_models.PlateOrder.objects.create( customer=self.customer, design_code='D2', urgency_level='正常', style_name='S2', ) def test_batch_update_success(self): resp = self.client.post( '/api/v2/plate-orders/batch-update/', { 'plate_order_ids': [self.po1.id, self.po2.id], 'data': {'urgency_level': '加急', 'designer': self.designer.id}, }, format='json', ) self.assertEqual(resp.status_code, 200) self.assertEqual(resp.data['updated_count'], 2) self.po1.refresh_from_db() self.po2.refresh_from_db() self.assertEqual(self.po1.urgency_level, '加急') self.assertEqual(self.po2.urgency_level, '加急') self.assertEqual(self.po1.designer_id, self.designer.id) self.assertEqual(self.po2.designer_id, self.designer.id) def test_batch_update_fails_when_missing_ids(self): resp = self.client.post( '/api/v2/plate-orders/batch-update/', { 'plate_order_ids': [self.po1.id, 999999], 'data': {'urgency_level': '加急'}, }, format='json', ) self.assertEqual(resp.status_code, 400) self.assertIn('不存在', resp.data.get('detail', '')) def test_batch_update_fails_when_unknown_field(self): resp = self.client.post( '/api/v2/plate-orders/batch-update/', { 'plate_order_ids': [self.po1.id], 'data': {'process': 123}, }, format='json', ) self.assertEqual(resp.status_code, 400) self.assertIn('不支持批量更新字段', resp.data.get('detail', '')) def test_batch_update_is_invalid_requires_permission(self): # 未授予 can_invalidate_plateorder resp = self.client.post( '/api/v2/plate-orders/batch-update/', { 'plate_order_ids': [self.po1.id], 'data': {'is_invalid': True}, }, format='json', ) self.assertEqual(resp.status_code, 403) perm_invalidate = Permission.objects.get(codename='can_invalidate_plateorder') self.user.user_permissions.add(perm_invalidate) resp2 = self.client.post( '/api/v2/plate-orders/batch-update/', { 'plate_order_ids': [self.po1.id], 'data': {'is_invalid': True}, }, format='json', ) self.assertEqual(resp2.status_code, 200) self.po1.refresh_from_db() self.assertTrue(self.po1.is_invalid) class MyVisiblePagesAPITest(TestCase): """测试获取当前用户可见页面 API""" def setUp(self): self.client = APIClient() self.merchant = basic_models.Merchant.objects.create( name='测试商户', type=basic_models.MerchantTypeEnum.FACTORY, ) self.user = get_user_model().objects.create_user(username='testuser', password='pass12345') # 创建前端页面 self.main_menu = basic_models.FrontendPage.objects.create( key='production', label='生产管理', page_type=basic_models.FrontendPageTypeEnum.MAIN, sort_order=100 ) self.sub_menu1 = basic_models.FrontendPage.objects.create( key='production-kanban', label='生产看板', page_type=basic_models.FrontendPageTypeEnum.SUB, parent=self.main_menu, sort_order=101 ) self.sub_menu2 = basic_models.FrontendPage.objects.create( key='printing-order-list', label='生产订单', page_type=basic_models.FrontendPageTypeEnum.SUB, parent=self.main_menu, sort_order=102 ) self.other_menu = basic_models.FrontendPage.objects.create( key='inventory', label='库存管理', page_type=basic_models.FrontendPageTypeEnum.MAIN, sort_order=200 ) # 创建岗位并分配页面 self.emp_type = basic_models.EmployeeType.objects.create( merchant=self.merchant, title='生产主管' ) self.emp_type.visible_pages.add(self.main_menu) # 创建员工并关联用户 self.employee = basic_models.Employee.objects.create( merchant=self.merchant, name='张三', position=self.emp_type, sys_user=self.user # 关联到用户 ) self.client.force_authenticate(user=self.user) def test_get_visible_pages_success(self): """测试成功获取当前用户可见页面""" response = self.client.get('/api/v2/me/visible-pages/') self.assertEqual(response.status_code, 200) data = response.data self.assertEqual(data['employee_id'], self.employee.id) self.assertEqual(data['employee_name'], '张三') self.assertEqual(data['position_id'], self.emp_type.id) self.assertEqual(data['position_title'], '生产主管') # 验证可见页面（主菜单 + 子菜单） visible_keys = data['visible_keys'] self.assertIn('production', visible_keys) self.assertIn('production-kanban', visible_keys) self.assertIn('printing-order-list', visible_keys) self.assertNotIn('inventory', visible_keys) # 验证页面详情 self.assertEqual(len(data['visible_pages']), 3) def test_get_visible_pages_with_sub_menu_only(self): """测试仅授予子菜单时只返回该子菜单""" # 重新配置岗位：只授予子菜单 self.emp_type.visible_pages.clear() self.emp_type.visible_pages.add(self.sub_menu1) response = self.client.get('/api/v2/me/visible-pages/') self.assertEqual(response.status_code, 200) visible_keys = response.data['visible_keys'] self.assertIn('production-kanban', visible_keys) self.assertNotIn('production', visible_keys) self.assertNotIn('printing-order-list', visible_keys) self.assertEqual(len(visible_keys), 1) def test_get_visible_pages_employee_no_position(self): """测试没有岗位的员工返回空列表""" # 移除员工岗位 self.employee.position = None self.employee.save() response = self.client.get('/api/v2/me/visible-pages/') self.assertEqual(response.status_code, 200) data = response.data self.assertEqual(data['employee_id'], self.employee.id) self.assertEqual(data['position_id'], None) self.assertEqual(data['visible_pages'], []) self.assertEqual(data['visible_keys'], []) def test_get_visible_pages_user_no_employee(self): """测试用户未关联员工时返回空列表""" # 创建一个没有关联员工的用户 user_no_employee = get_user_model().objects.create_user(username='noemployee', password='pass12345') self.client.force_authenticate(user=user_no_employee) response = self.client.get('/api/v2/me/visible-pages/') self.assertEqual(response.status_code, 200) data = response.data self.assertEqual(data['employee_id'], None) self.assertEqual(data['employee_name'], None) self.assertEqual(data['position_id'], None) self.assertEqual(data['visible_pages'], []) self.assertEqual(data['visible_keys'], []) def test_get_visible_pages_unauthenticated(self): """测试未登录时返回 401""" self.client.logout() response = self.client.get('/api/v2/me/visible-pages/') self.assertEqual(response.status_code, 401) class PrintingOrderBatchAdvanceRecordsAPITest(TestCase): """测试印染订单批量推进记录查询 API""" def setUp(self): self.client = APIClient() self.merchant = basic_models.Merchant.objects.create( name='测试商户', type=basic_models.MerchantTypeEnum.FACTORY, ) self.user = get_user_model().objects.create_user(username='testuser', password='pass12345') self.client.force_authenticate(user=self.user) # 创建客户 self.customer = basic_models.Customer.objects.create( merchant=self.merchant, name='测试客户' ) # 创建产品类别和产品 self.category = basic_models.ProductCategory.objects.create( merchant=self.merchant, name='测试类别' ) self.product = basic_models.Product.objects.create( merchant=self.merchant, name='测试产品', category=self.category, unit=basic_models.ProductUnitEnum.METER ) # 创建印染订单 self.printing_order = printing_models.PrintingOrder.objects.create( customer=self.customer, fabric='测试面料', width='150cm' ) # 创建印染明细 self.job1 = printing_models.PrintingJob.objects.create( printing_order=self.printing_order, product=self.product, quantity=100, unit='米' ) self.job2 = printing_models.PrintingJob.objects.create( printing_order=self.printing_order, product=self.product, quantity=200, unit='米' ) # 创建流程状态 self.state = stateflow_models.State.objects.create( name='染色', description='染色工序' ) # 创建批量推进记录 self.record = printing_models.PrintingJobBatchAdvanceRecord.objects.create( printing_order=self.printing_order, state=self.state, created_by=self.user, parameters={'temperature': '25.5', 'operator': '张三'} ) self.record.printing_jobs.add(self.job1, self.job2) def test_get_batch_advance_records_success(self): """测试成功获取批量推进记录""" response = self.client.get(f'/api/v2/printing-orders/{self.printing_order.id}/batch-advance-records/') self.assertEqual(response.status_code, 200) data = response.data self.assertEqual(data['printing_order_id'], self.printing_order.id) self.assertEqual(data['total_count'], 1) self.assertEqual(len(data['records']), 1) record = data['records'][0] self.assertEqual(record['id'], self.record.id) self.assertEqual(record['state_id'], self.state.id) self.assertEqual(record['state_name'], '染色') self.assertEqual(record['created_by'], self.user.id) self.assertEqual(record['created_by_username'], 'testuser') self.assertEqual(record['parameters'], {'temperature': '25.5', 'operator': '张三'}) self.assertEqual(record['printing_job_count'], 2) self.assertIn(self.job1.id, record['printing_job_ids']) self.assertIn(self.job2.id, record['printing_job_ids']) def test_get_batch_advance_records_empty(self): """测试没有批量推进记录时返回空列表""" # 创建另一个没有记录的订单 another_order = printing_models.PrintingOrder.objects.create( customer=self.customer, fabric='测试面料2', width='160cm' ) response = self.client.get(f'/api/v2/printing-orders/{another_order.id}/batch-advance-records/') self.assertEqual(response.status_code, 200) data = response.data self.assertEqual(data['printing_order_id'], another_order.id) self.assertEqual(data['total_count'], 0) self.assertEqual(data['records'], []) def test_get_batch_advance_records_order_not_found(self): """测试订单不存在时返回 404""" response = self.client.get('/api/v2/printing-orders/99999/batch-advance-records/') self.assertEqual(response.status_code, 404) self.assertIn('error', response.data) def test_get_batch_advance_records_multiple_records(self): """测试多条批量推进记录按时间倒序返回""" # 创建第二条记录 state2 = stateflow_models.State.objects.create(name='整理', description='整理工序') record2 = printing_models.PrintingJobBatchAdvanceRecord.objects.create( printing_order=self.printing_order, state=state2, created_by=self.user, parameters={'notes': '完成'} ) record2.printing_jobs.add(self.job1) response = self.client.get(f'/api/v2/printing-orders/{self.printing_order.id}/batch-advance-records/') self.assertEqual(response.status_code, 200) data = response.data self.assertEqual(data['total_count'], 2) # 最新的记录应该在前面 self.assertEqual(data['records'][0]['id'], record2.id) self.assertEqual(data['records'][1]['id'], self.record.id) def test_get_batch_advance_records_unauthenticated(self): """测试未登录时返回 401""" self.client.logout() response = self.client.get(f'/api/v2/printing-orders/{self.printing_order.id}/batch-advance-records/') self.assertEqual(response.status_code, 401) class PrintingJobBatchAddParametersAPITest(TestCase): """测试批量补充工艺参数 API""" def setUp(self): self.client = APIClient() self.merchant = basic_models.Merchant.objects.create( name='测试工厂', type=basic_models.MerchantTypeEnum.FACTORY, ) self.user = get_user_model().objects.create_user(username='factory_user', password='pass12345') self.employee = basic_models.Employee.objects.create( merchant=self.merchant, name='工厂员工', sys_user=self.user ) self.client.force_authenticate(user=self.user) # 创建客户 self.customer = basic_models.Customer.objects.create( merchant=self.merchant, name='测试客户' ) # 创建产品类别和产品 self.category = basic_models.ProductCategory.objects.create( merchant=self.merchant, name='测试类别' ) self.product = basic_models.Product.objects.create( merchant=self.merchant, name='测试产品', category=self.category, unit=basic_models.ProductUnitEnum.METER ) # 创建流程状态 self.state1 = stateflow_models.State.objects.create(name='染色', description='染色工序') self.state2 = stateflow_models.State.objects.create(name='整理', description='整理工序') # 创建流程 self.process = stateflow_models.Process.objects.create(name='印染流程') stateflow_models.ProcessNode.objects.create(process=self.process, state=self.state1, order=1) stateflow_models.ProcessNode.objects.create(process=self.process, state=self.state2, order=2) # 创建印染订单 self.printing_order = printing_models.PrintingOrder.objects.create( customer=self.customer, fabric='测试面料', width='150cm', process=self.process ) # 创建印染明细并绑定流程实例 self.job1 = printing_models.PrintingJob.objects.create( printing_order=self.printing_order, product=self.product, quantity=100, unit='米' ) self.job2 = printing_models.PrintingJob.objects.create( printing_order=self.printing_order, product=self.product, quantity=200, unit='米' ) # 创建业务对象并推进到 state1 from stateflow import services as stateflow_services from django.contrib.contenttypes.models import ContentType ct = ContentType.objects.get_for_model(printing_models.PrintingJob) self.bo1 = stateflow_models.BusinessObject.objects.create( name=f'PrintingJob-{self.job1.id}', process=self.process, content_type=ct, object_id=self.job1.id ) self.job1.business_object = self.bo1 self.job1.save() self.bo2 = stateflow_models.BusinessObject.objects.create( name=f'PrintingJob-{self.job2.id}', process=self.process, content_type=ct, object_id=self.job2.id ) self.job2.business_object = self.bo2 self.job2.save() # 推进到 state1（创建 state_log） stateflow_services.advance_to_next_state(self.bo1, self.user, temperature='25.0') stateflow_services.advance_to_next_state(self.bo2, self.user, temperature='25.0') def test_batch_add_parameters_success(self): """测试成功批量补充参数""" response = self.client.post( '/api/v2/printing-jobs/batch-add-parameters/', { 'printing_job_ids': [self.job1.id, self.job2.id], 'state_id': self.state1.id, 'parameters': {'temperature': '26.0', 'operator': '李四'}, 'remark': '补测数据' }, format='json' ) self.assertEqual(response.status_code, 200) data = response.data self.assertEqual(data['detail'], '批量补充参数成功') self.assertEqual(data['printing_order_id'], self.printing_order.id) self.assertEqual(set(data['printing_job_ids']), {self.job1.id, self.job2.id}) self.assertEqual(data['state_id'], self.state1.id) self.assertEqual(data['state_name'], '染色') self.assertEqual(data['affected_count'], 2) # 验证创建了批量操作记录 record = printing_models.PrintingJobBatchAdvanceRecord.objects.get(id=data['batch_record_id']) self.assertTrue(record.only_parameters) self.assertEqual(record.parameters, {'temperature': '26.0', 'operator': '李四'}) self.assertEqual(record.printing_jobs.count(), 2) def test_batch_add_parameters_empty_parameters(self): """测试参数为空时返回错误""" response = self.client.post( '/api/v2/printing-jobs/batch-add-parameters/', { 'printing_job_ids': [self.job1.id], 'state_id': self.state1.id, 'parameters': {}, }, format='json' ) self.assertEqual(response.status_code, 400) def test_batch_add_parameters_job_not_found(self): """测试 job 不存在时返回错误""" response = self.client.post( '/api/v2/printing-jobs/batch-add-parameters/', { 'printing_job_ids': [99999], 'state_id': self.state1.id, 'parameters': {'temperature': '26.0'}, }, format='json' ) self.assertEqual(response.status_code, 400) self.assertIn('missing_jobs', response.data) def test_batch_add_parameters_state_not_found(self): """测试 state 不存在时返回错误""" response = self.client.post( '/api/v2/printing-jobs/batch-add-parameters/', { 'printing_job_ids': [self.job1.id], 'state_id': 99999, 'parameters': {'temperature': '26.0'}, }, format='json' ) self.assertEqual(response.status_code, 400) def test_batch_add_parameters_missing_state_log(self): """测试 job 没有对应 state_log 时返回错误""" # state2 的 state_log 尚未创建（还没推进到那一步） response = self.client.post( '/api/v2/printing-jobs/batch-add-parameters/', { 'printing_job_ids': [self.job1.id], 'state_id': self.state2.id, # 还没推进到这个状态 'parameters': {'temperature': '26.0'}, }, format='json' ) self.assertEqual(response.status_code, 400) self.assertIn('missing_state_log', response.data) def test_batch_add_parameters_different_orders(self): """测试 jobs 属于不同订单时返回错误""" # 创建另一个订单和 job another_order = printing_models.PrintingOrder.objects.create( customer=self.customer, fabric='另一面料', width='160cm' ) another_job = printing_models.PrintingJob.objects.create( printing_order=another_order, product=self.product, quantity=300, unit='米' ) response = self.client.post( '/api/v2/printing-jobs/batch-add-parameters/', { 'printing_job_ids': [self.job1.id, another_job.id], 'state_id': self.state1.id, 'parameters': {'temperature': '26.0'}, }, format='json' ) self.assertEqual(response.status_code, 400) def test_batch_add_parameters_unauthenticated(self): """测试未登录时返回 401""" self.client.logout() response = self.client.post( '/api/v2/printing-jobs/batch-add-parameters/', { 'printing_job_ids': [self.job1.id], 'state_id': self.state1.id, 'parameters': {'temperature': '26.0'}, }, format='json' ) self.assertEqual(response.status_code, 401)