forked from erp-dev/erp
137 lines
4.8 KiB
Python
137 lines
4.8 KiB
Python
from unittest.mock import patch
|
|
|
|
from django.contrib.auth import get_user_model
|
|
from django.db import transaction
|
|
from django.test import TestCase, TransactionTestCase, override_settings
|
|
|
|
from basic_info import models as basic_models
|
|
from printing import handlers
|
|
from printing import models as printing_models
|
|
from printing import tasks as printing_tasks
|
|
from shipment import models as shipment_models
|
|
|
|
|
|
class PrintingJobProductionCompletedWeComServiceTestCase(TestCase):
|
|
def setUp(self):
|
|
self.user = get_user_model().objects.create_user(username="notify-user", password="pass")
|
|
self.merchant = basic_models.Merchant.objects.create(
|
|
name="测试印花厂",
|
|
type=basic_models.MerchantTypeEnum.FACTORY,
|
|
)
|
|
basic_models.Employee.objects.create(
|
|
sys_user=self.user,
|
|
merchant=self.merchant,
|
|
name="测试员工",
|
|
status=basic_models.EmployeeStatusEnum.ACTIVE,
|
|
)
|
|
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,
|
|
category=self.category,
|
|
name="测试产品",
|
|
)
|
|
|
|
def test_send_printing_job_production_completed_wecom_fallbacks_to_order_id(self):
|
|
order = printing_models.PrintingOrder.objects.create(
|
|
merchant=self.merchant,
|
|
customer=self.customer,
|
|
fabric="棉布",
|
|
width="150cm",
|
|
external_order_id="",
|
|
)
|
|
job = printing_models.PrintingJob.objects.create(
|
|
merchant=self.merchant,
|
|
printing_order=order,
|
|
product=self.product,
|
|
quantity=10,
|
|
unit="米",
|
|
)
|
|
shipment_models.SalesItem.objects.create(
|
|
merchant=self.merchant,
|
|
name="未出货销售品1",
|
|
quantity="50.50",
|
|
unit=shipment_models.UnitChoices.METER,
|
|
printing_job_id=job.id,
|
|
created_by=self.user,
|
|
)
|
|
shipment_models.SalesItem.objects.create(
|
|
merchant=self.merchant,
|
|
name="未出货销售品2",
|
|
quantity="38.00",
|
|
unit=shipment_models.UnitChoices.METER,
|
|
printing_job_id=job.id,
|
|
created_by=self.user,
|
|
)
|
|
|
|
from printing.services import send_printing_job_production_completed_wecom
|
|
|
|
payload = send_printing_job_production_completed_wecom(
|
|
printing_job_id=job.id,
|
|
triggered_by_id=self.user.id,
|
|
dry_run=True,
|
|
)
|
|
|
|
self.assertEqual(payload["printing_order_identifier"], str(order.id))
|
|
self.assertEqual(payload["product_name"], self.product.name)
|
|
self.assertEqual(payload["unshipped_sales_items_count"], 2)
|
|
self.assertEqual(payload["unshipped_sales_items_quantity"], "88.50")
|
|
self.assertEqual(payload["sender_label"], "测试员工")
|
|
|
|
|
|
@override_settings(
|
|
CELERY_TASK_ALWAYS_EAGER=True,
|
|
CELERY_TASK_EAGER_PROPAGATES=True,
|
|
)
|
|
class PrintingJobProductionCompletedWeComTaskTestCase(TestCase):
|
|
def test_task_delegates_to_service(self):
|
|
with patch(
|
|
"printing.tasks.send_printing_job_production_completed_wecom"
|
|
) as mock_sync:
|
|
mock_sync.return_value = {"printing_job_id": 123, "ok": True}
|
|
|
|
async_result = printing_tasks.notify_printing_job_production_completed_wecom.delay(
|
|
printing_job_id=123,
|
|
triggered_by_id=456,
|
|
)
|
|
payload = async_result.get(timeout=5)
|
|
|
|
mock_sync.assert_called_once_with(
|
|
printing_job_id=123,
|
|
triggered_by_id=456,
|
|
)
|
|
self.assertEqual(payload["printing_job_id"], 123)
|
|
self.assertIn("task_id", payload)
|
|
|
|
|
|
@override_settings(
|
|
TESTING=False,
|
|
PRINTING_JOB_PRODUCTION_COMPLETED_WECOM_NOTIFY_ENABLED=True,
|
|
)
|
|
class PrintingJobProductionCompletedHandlerTestCase(TransactionTestCase):
|
|
def test_handler_enqueues_task_on_commit(self):
|
|
job = type("Job", (), {"id": 321})()
|
|
user = type("User", (), {"id": 654})()
|
|
|
|
with patch(
|
|
"printing.tasks.notify_printing_job_production_completed_wecom.delay"
|
|
) as mock_delay:
|
|
with transaction.atomic():
|
|
handlers.on_printing_job_production_completed(
|
|
sender=printing_models.PrintingJob,
|
|
instance=job,
|
|
triggered_by=user,
|
|
)
|
|
self.assertFalse(mock_delay.called)
|
|
|
|
mock_delay.assert_called_once_with(
|
|
printing_job_id=321,
|
|
triggered_by_id=654,
|
|
)
|