feat: added wecom notify when printing_job production was completed(new status) and shipment was created

api_v1/views/printing/test_services.py

@@ -7,6 +7,7 @@ from django.conf import settings
 from basic_info.models import Merchant, MerchantTypeEnum, Employee, Customer, Product
 from stateflow.models import State, Process
 from printing.models import PrintingOrder, PrintingJob
+from printing.signals import printing_job_production_completed
 from api_v1.views.printing.services import PrintingOrderService, PrintingJobService
 
 
@@ -380,7 +381,71 @@ class PrintingJobServiceTest(TestCase):
         self.assertTrue(success)
         self.assertEqual(updated_job.quantity, 20)
         self.assertEqual(updated_job.pieces, 10)
-    
+
+    def test_create_printing_job_ignores_is_production_completed(self):
+        """测试创建任务时忽略 is_production_completed 输入"""
+        data = {
+            'printing_order': self.order,
+            'product': self.product,
+            'quantity': 10,
+            'unit': '件',
+            'is_production_completed': True,
+        }
+
+        job = PrintingJobService.create_printing_job(data, self.user)
+
+        self.assertFalse(job.is_production_completed)
+
+    def test_update_printing_job_ignores_is_production_completed(self):
+        """测试更新任务时忽略 is_production_completed 输入"""
+        job = PrintingJob.objects.create(
+            printing_order=self.order,
+            product=self.product,
+            quantity=10,
+            unit='件',
+            is_production_completed=False,
+        )
+
+        success, _, updated_job = PrintingJobService.update_printing_job(
+            job,
+            {'is_production_completed': True, 'quantity': 20},
+            self.user,
+        )
+
+        updated_job.refresh_from_db()
+        self.assertTrue(success)
+        self.assertEqual(updated_job.quantity, 20)
+        self.assertFalse(updated_job.is_production_completed)
+
+    def test_make_printing_job_production_completed_sets_flag_and_sends_signal(self):
+        """测试显式完成生产会更新字段并触发 signal"""
+        job = PrintingJob.objects.create(
+            printing_order=self.order,
+            product=self.product,
+            quantity=10,
+            unit='件',
+        )
+
+        received = {}
+        uid = 'test.printing_job_production_completed'
+
+        def _receiver(sender, **kwargs):
+            received['sender'] = sender
+            received['instance_id'] = kwargs['instance'].id
+            received['triggered_by'] = kwargs.get('triggered_by')
+
+        printing_job_production_completed.connect(_receiver, dispatch_uid=uid, weak=False)
+        try:
+            updated_job = PrintingJobService.make_printing_job_production_completed(job, self.user)
+        finally:
+            printing_job_production_completed.disconnect(dispatch_uid=uid)
+
+        updated_job.refresh_from_db()
+        self.assertTrue(updated_job.is_production_completed)
+        self.assertEqual(received['sender'], PrintingJob)
+        self.assertEqual(received['instance_id'], job.id)
+        self.assertEqual(received['triggered_by'], self.user)
+
     def test_get_job_status(self):
         """测试获取任务状态"""
         job = PrintingJob.objects.create(
@@ -397,9 +462,11 @@ class PrintingJobServiceTest(TestCase):
         self.assertIn('status', status_info)
         self.assertIn('status_id', status_info)
         self.assertIn('is_completed', status_info)
+        self.assertIn('is_production_completed', status_info)
         self.assertIn('has_started', status_info)
         
         # 未开始状态（现在返回空字符串或第一个状态名称）
         self.assertIn(status_info['status'], ['', self.state1.name])
         self.assertFalse(status_info['is_completed'])
+        self.assertFalse(status_info['is_production_completed'])
         self.assertFalse(status_info['has_started'])