fix bug worker
This commit is contained in:
1 parent
4525ec3e6a
commit
81335b1439
2 files changed
+24
-1
No files matched your search
@@ -94,7 +94,8 @@ def process_due_announcements(*, job_id=None):
|
|||||||
).update(status=AnnouncementJob.Status.PENDING, next_attempt_at=now)
|
).update(status=AnnouncementJob.Status.PENDING, next_attempt_at=now)
|
||||||
with transaction.atomic():
|
with transaction.atomic():
|
||||||
queryset = (
|
queryset = (
|
||||||
AnnouncementJob.objects.select_for_update(skip_locked=True)
|
# Nullable message/ticket joins must not be included in FOR UPDATE on PostgreSQL.
|
||||||
|
AnnouncementJob.objects.select_for_update(skip_locked=True, of=("self",))
|
||||||
.filter(status=AnnouncementJob.Status.PENDING, next_attempt_at__lte=now)
|
.filter(status=AnnouncementJob.Status.PENDING, next_attempt_at__lte=now)
|
||||||
.select_related("message__ticket__customer", "ticket__customer")
|
.select_related("message__ticket__customer", "ticket__customer")
|
||||||
.annotate(priority=Case(When(source=AnnouncementJob.Source.MANUAL, then=Value(0)), default=Value(1), output_field=IntegerField()))
|
.annotate(priority=Case(When(source=AnnouncementJob.Source.MANUAL, then=Value(0)), default=Value(1), output_field=IntegerField()))
|
||||||
|
|||||||
@@ -84,6 +84,28 @@ def test_due_unread_message_calls_once_with_optional_api_key(announcement_data,
|
|||||||
assert post.call_args.kwargs["headers"] == ({"Idempotency-Key": job.idempotency_key, "X-API-Key": api_key} if api_key else {"Idempotency-Key": job.idempotency_key})
|
assert post.call_args.kwargs["headers"] == ({"Idempotency-Key": job.idempotency_key, "X-API-Key": api_key} if api_key else {"Idempotency-Key": job.idempotency_key})
|
||||||
|
|
||||||
|
|
||||||
|
def test_worker_locks_only_announcement_rows_with_nullable_relations(announcement_data, monkeypatch):
|
||||||
|
category, staff, customer, _ = announcement_data
|
||||||
|
ticket = create_ticket(customer=customer, created_by=staff, category=category, subject="پیگیری تماس", body="پیام نخست")
|
||||||
|
job = AnnouncementJob.objects.get(message__ticket=ticket)
|
||||||
|
due(job)
|
||||||
|
manager = AnnouncementJob.objects
|
||||||
|
original = manager.select_for_update
|
||||||
|
lock_targets = []
|
||||||
|
|
||||||
|
def capture_lock(*args, **kwargs):
|
||||||
|
queryset = original(*args, **kwargs)
|
||||||
|
lock_targets.append(queryset.query.select_for_update_of)
|
||||||
|
return queryset
|
||||||
|
|
||||||
|
monkeypatch.setattr(manager, "select_for_update", capture_lock)
|
||||||
|
with patch("apps.integrations.announcements.requests.post", return_value=Mock(content=b"")):
|
||||||
|
call_command("process_notifications", once=True)
|
||||||
|
job.refresh_from_db()
|
||||||
|
assert job.status == AnnouncementJob.Status.SENT
|
||||||
|
assert lock_targets == [("self",), ("self",)]
|
||||||
|
|
||||||
|
|
||||||
def test_customer_read_cancels_only_that_message(announcement_data):
|
def test_customer_read_cancels_only_that_message(announcement_data):
|
||||||
category, staff, customer, _ = announcement_data
|
category, staff, customer, _ = announcement_data
|
||||||
ticket = create_ticket(customer=customer, created_by=staff, category=category, subject="پیگیری مشتری", body="پیام نخست")
|
ticket = create_ticket(customer=customer, created_by=staff, category=category, subject="پیگیری مشتری", body="پیام نخست")
|
||||||
|
|||||||
Reference in new issue
Block a user