forked from erp-dev/erp
99 lines
3.4 KiB
Python
99 lines
3.4 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 shipment import handlers
|
|
from shipment import models as shipment_models
|
|
from shipment import tasks as shipment_tasks
|
|
|
|
|
|
class ShipmentCreatedWeComServiceTestCase(TestCase):
|
|
def setUp(self):
|
|
self.user = get_user_model().objects.create_user(username="shipment-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="测试客户",
|
|
)
|
|
|
|
def test_send_shipment_created_wecom_dry_run(self):
|
|
shipment = shipment_models.Shipment.objects.create(
|
|
merchant=self.merchant,
|
|
customer=self.customer,
|
|
shipment_date="2026-01-14",
|
|
created_by=self.user,
|
|
)
|
|
|
|
from shipment.services import send_shipment_created_wecom
|
|
|
|
payload = send_shipment_created_wecom(
|
|
shipment_id=shipment.id,
|
|
created_by_id=self.user.id,
|
|
dry_run=True,
|
|
)
|
|
|
|
self.assertEqual(payload["shipment_id"], shipment.id)
|
|
self.assertEqual(payload["customer_name"], self.customer.name)
|
|
self.assertEqual(payload["shipment_date"], "2026-01-14")
|
|
self.assertEqual(payload["items_count"], 0)
|
|
self.assertEqual(payload["sender_label"], "出货员工")
|
|
|
|
|
|
@override_settings(
|
|
CELERY_TASK_ALWAYS_EAGER=True,
|
|
CELERY_TASK_EAGER_PROPAGATES=True,
|
|
)
|
|
class ShipmentCreatedWeComTaskTestCase(TestCase):
|
|
def test_task_delegates_to_service(self):
|
|
with patch("shipment.tasks.send_shipment_created_wecom") as mock_sync:
|
|
mock_sync.return_value = {"shipment_id": 123, "ok": True}
|
|
|
|
async_result = shipment_tasks.notify_shipment_created_wecom.delay(
|
|
shipment_id=123,
|
|
created_by_id=456,
|
|
)
|
|
payload = async_result.get(timeout=5)
|
|
|
|
mock_sync.assert_called_once_with(
|
|
shipment_id=123,
|
|
created_by_id=456,
|
|
)
|
|
self.assertEqual(payload["shipment_id"], 123)
|
|
self.assertIn("task_id", payload)
|
|
|
|
|
|
@override_settings(
|
|
TESTING=False,
|
|
SHIPMENT_CREATED_WECOM_NOTIFY_ENABLED=True,
|
|
)
|
|
class ShipmentCreatedHandlerTestCase(TransactionTestCase):
|
|
def test_handler_enqueues_task_on_commit(self):
|
|
shipment = type("Shipment", (), {"id": 321})()
|
|
user = type("User", (), {"id": 654})()
|
|
|
|
with patch("shipment.tasks.notify_shipment_created_wecom.delay") as mock_delay:
|
|
with transaction.atomic():
|
|
handlers.on_shipment_created(
|
|
sender=shipment_models.Shipment,
|
|
instance=shipment,
|
|
created_by=user,
|
|
)
|
|
self.assertFalse(mock_delay.called)
|
|
|
|
mock_delay.assert_called_once_with(
|
|
shipment_id=321,
|
|
created_by_id=654,
|
|
)
|