feat: added process_node parameters view for plate order

api_v2/tests.py

@@ -479,3 +479,195 @@ class BusinessObjectCloneV2APITest(TestCase):
             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('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)