forked from erp-dev/erp
65 lines
2.5 KiB
Python
65 lines
2.5 KiB
Python
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)
|
||
|