feat: task for auto upload plate_image to tencent each day 03:00

printing/test_tiia_upload_task.py

@@ -0,0 +1,64 @@
+from datetime import datetime, time, timedelta
+from unittest.mock import patch
+
+from django.test import TestCase
+from django.utils import timezone
+
+from basic_info import models as basic_models
+from printing import models as printing_models
+
+
+class PlateOrderTiiaUploadTaskTestCase(TestCase):
+    def setUp(self):
+        self.merchant = basic_models.Merchant.objects.create(
+            name='测试商户',
+            type=basic_models.MerchantTypeEnum.FACTORY,
+        )
+        self.customer = basic_models.Customer.objects.create(
+            merchant=self.merchant,
+            name='客户A',
+            mobile='13900000000',
+            area='A',
+        )
+
+    @patch('api_v1.utils.tencentcloud_tiia.time.sleep', lambda *_args, **_kwargs: None)
+    @patch('api_v1.utils.tencentcloud_tiia.upload_plate_order_images_to_tencent_tiia')
+    def test_task_records_failure_and_continues(self, mocked_upload_plate_order):
+        # 准备：创建 2 个 plate_order，并把 created_at 改到“昨天”
+        po_ok = printing_models.PlateOrder.objects.create(
+            merchant=self.merchant,
+            customer=self.customer,
+            plate_image=[{'url': 'https://example.com/a.png'}],
+        )
+        po_bad = printing_models.PlateOrder.objects.create(
+            merchant=self.merchant,
+            customer=self.customer,
+            plate_image=[{'url': 'https://example.com/b.png'}],
+        )
+
+        yesterday = timezone.localdate() - timedelta(days=1)
+        dt = timezone.make_aware(datetime.combine(yesterday, time.min))
+        printing_models.PlateOrder.objects.filter(id__in=[po_ok.id, po_bad.id]).update(created_at=dt)
+
+        def side_effect(*, plate_order_id: int, rate_limiter=None):
+            if plate_order_id == po_bad.id:
+                raise ValueError('boom')
+            return [{'image_url': 'https://example.com/a.png', 'ok': True, 'response': {'ok': 1}, 'error': None}]
+
+        mocked_upload_plate_order.side_effect = side_effect
+
+        from printing.tasks import upload_yesterday_plate_order_images_to_tencent_tiia
+
+        payload = upload_yesterday_plate_order_images_to_tencent_tiia()
+
+        self.assertEqual(payload['total_orders'], 2)
+        self.assertEqual(payload['failed_count'], 1)
+        self.assertIn(po_bad.id, payload['failed_ids'])
+
+        # failure 表应落库
+        from printing.models import PlateOrderTiiaUploadFailure
+
+        rec = PlateOrderTiiaUploadFailure.objects.filter(run_date=yesterday, plate_order_id=po_bad.id).first()
+        self.assertIsNotNone(rec)
+        self.assertGreaterEqual(rec.attempts, 1)
+