661 lines
36 KiB
Python
661 lines
36 KiB
Python
import pytest
|
|
import base64
|
|
from datetime import timedelta
|
|
from django.contrib.auth.models import Permission
|
|
from django.core.exceptions import PermissionDenied, ValidationError
|
|
from django.test import override_settings
|
|
from django.utils import timezone
|
|
from django.core.files.uploadedfile import SimpleUploadedFile
|
|
from apps.attachments.models import Attachment, AttachmentPolicy
|
|
from apps.accounts.models import User, normalize_mobile
|
|
from apps.notifications.models import NotificationJob
|
|
from apps.staff.models import OrganizationalUnit, StaffProfile
|
|
from apps.tickets.models import Ticket, TicketCategory
|
|
from apps.tickets.policies import TicketPolicy
|
|
from apps.tickets.services import claim_ticket, create_ticket, delete_ticket, reply_ticket
|
|
|
|
@pytest.fixture
|
|
def category(db):
|
|
unit = OrganizationalUnit.objects.create(name="واحد تست دامنه", slug="domain-test-unit", bale_chat_id="domain-bale-group")
|
|
unit.permissions.add(
|
|
Permission.objects.get(codename="view_ticket", content_type__app_label="tickets"),
|
|
Permission.objects.get(codename="add_message", content_type__app_label="messaging"),
|
|
)
|
|
return TicketCategory.objects.create(unit=unit, title="قیمت", slug="domain-test-price")
|
|
|
|
@pytest.fixture
|
|
def customer(db):
|
|
return User.objects.create_user(username="customer", mobile="09121234567", account_type="CUSTOMER")
|
|
|
|
def make_staff(username):
|
|
mobile = "09120000001" if username == "a" else "09120000002"
|
|
user = User.objects.create_user(username=username, mobile=mobile, account_type="STAFF")
|
|
unit = OrganizationalUnit.objects.filter(slug="domain-test-unit").first()
|
|
StaffProfile.objects.create(user=user, primary_unit=unit, display_name=username)
|
|
for code in ["view_ticket", "claim_ticket", "change_ticket"]:
|
|
user.user_permissions.add(Permission.objects.get(codename=code, content_type__app_label="tickets"))
|
|
return user
|
|
|
|
@pytest.mark.django_db
|
|
def test_mobile_normalization():
|
|
assert normalize_mobile("0912 123 4567") == "+989121234567"
|
|
|
|
@pytest.mark.django_db
|
|
def test_customer_reply_enqueues_bale_group_notification(category, customer):
|
|
ticket = create_ticket(customer=customer, category=category, subject="پیگیری سفارش", body="پیام نخست")
|
|
message = reply_ticket(ticket_id=ticket.pk, user=customer, body="پاسخ تازه مشتری")
|
|
job = NotificationJob.objects.get(event_type="CUSTOMER_TICKET_REPLY", ticket=ticket)
|
|
assert job.channel == NotificationJob.Channel.BALE
|
|
assert job.payload == {"message_id": message.pk,"message_count":1}
|
|
assert job.next_attempt_at>=timezone.now()+timedelta(seconds=595)
|
|
|
|
@pytest.mark.django_db
|
|
def test_customer_ticket_and_reply_bale_events_target_owning_unit_group(category, customer, monkeypatch):
|
|
from apps.notifications.management.commands.process_notifications import _send_bale
|
|
sent = []
|
|
class Provider:
|
|
def send_message(self, chat_id, text):
|
|
sent.append((chat_id, text))
|
|
return type("Result", (), {"success": True})()
|
|
monkeypatch.setattr("apps.notifications.management.commands.process_notifications.get_bale_provider", lambda: Provider())
|
|
from apps.customers.models import CustomerProfile
|
|
profile,_=CustomerProfile.objects.get_or_create(user=customer)
|
|
profile.first_name="سارا"; profile.last_name="احمدی"; profile.city="تهران"; profile.save()
|
|
ticket = create_ticket(customer=customer, category=category, subject="اعلان بله", body="پیام نخست")
|
|
created_job = NotificationJob.objects.get(ticket=ticket, event_type="NEW_CUSTOMER_TICKET")
|
|
_send_bale(created_job)
|
|
reply = reply_ticket(ticket_id=ticket.pk, user=customer, body="پاسخ مشتری")
|
|
reply_job = NotificationJob.objects.get(ticket=ticket, event_type="CUSTOMER_TICKET_REPLY")
|
|
_send_bale(reply_job)
|
|
assert [chat_id for chat_id, _ in sent] == ["domain-bale-group", "domain-bale-group"]
|
|
assert "تیکت جدید مشتری" in sent[0][1]
|
|
assert "پاسخ جدید مشتری" in sent[1][1]
|
|
assert reply.body in sent[1][1]
|
|
for text in (sent[0][1],sent[1][1]):
|
|
assert "👤 سارا احمدی" in text
|
|
assert f"📞 {customer.local_mobile}" in text
|
|
assert f"/staff/tickets/{ticket.public_id}/" in text
|
|
assert "🎫" in text and "📌 اعلان بله" in text and "📝" in text
|
|
assert "📍 تهران" in sent[0][1]
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_bale_group_notifications_are_isolated_between_units(category, customer, monkeypatch):
|
|
from apps.notifications.management.commands.process_notifications import _send_bale
|
|
|
|
second_unit = OrganizationalUnit.objects.create(
|
|
name="واحد دوم اعلان بله",
|
|
slug="second-bale-unit",
|
|
bale_chat_id="second-bale-group",
|
|
)
|
|
second_category = TicketCategory.objects.create(unit=second_unit, title="حسابداری", slug="second-bale-category")
|
|
sent = []
|
|
|
|
class Provider:
|
|
def send_message(self, chat_id, text):
|
|
sent.append((chat_id, text))
|
|
return type("Result", (), {"success": True})()
|
|
|
|
monkeypatch.setattr("apps.notifications.management.commands.process_notifications.get_bale_provider", lambda: Provider())
|
|
first_ticket = create_ticket(customer=customer, category=category, subject="درخواست فروش", body="پیام فروش")
|
|
second_ticket = create_ticket(customer=customer, category=second_category, subject="درخواست مالی", body="پیام مالی")
|
|
first_reply = reply_ticket(ticket_id=first_ticket.pk, user=customer, body="پاسخ فروش")
|
|
second_reply = reply_ticket(ticket_id=second_ticket.pk, user=customer, body="پاسخ مالی")
|
|
|
|
for ticket, event_type in (
|
|
(first_ticket, "NEW_CUSTOMER_TICKET"),
|
|
(second_ticket, "NEW_CUSTOMER_TICKET"),
|
|
(first_ticket, "CUSTOMER_TICKET_REPLY"),
|
|
(second_ticket, "CUSTOMER_TICKET_REPLY"),
|
|
):
|
|
_send_bale(NotificationJob.objects.get(ticket=ticket, event_type=event_type))
|
|
|
|
assert [chat_id for chat_id, _ in sent] == [
|
|
"domain-bale-group",
|
|
"second-bale-group",
|
|
"domain-bale-group",
|
|
"second-bale-group",
|
|
]
|
|
assert first_reply.body in sent[2][1]
|
|
assert second_reply.body in sent[3][1]
|
|
assert "درخواست مالی" not in sent[0][1] and "درخواست فروش" not in sent[1][1]
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_blank_unit_bale_chat_never_falls_back_to_global_group(category, customer, settings):
|
|
settings.BALE_GROUP_CHAT_ID = "legacy-global-group"
|
|
category.unit.bale_chat_id = ""
|
|
category.unit.save(update_fields=["bale_chat_id", "updated_at"])
|
|
|
|
ticket = create_ticket(customer=customer, category=category, subject="بدون گروه", body="پیام")
|
|
reply_ticket(ticket_id=ticket.pk, user=customer, body="پاسخ")
|
|
|
|
assert not NotificationJob.objects.filter(ticket=ticket, channel=NotificationJob.Channel.BALE).exists()
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_unit_notification_destinations_normalize_and_reject_duplicate_supervisors():
|
|
unit = OrganizationalUnit(
|
|
name="واحد مقصد اعلان",
|
|
slug="notification-destinations",
|
|
supervisor_mobile_1="0912 111 2233",
|
|
supervisor_mobile_2="0935 222 3344",
|
|
)
|
|
unit.full_clean()
|
|
unit.save()
|
|
assert unit.supervisor_mobiles == ("+989121112233", "+989352223344")
|
|
|
|
unit.supervisor_mobile_3 = "09121112233"
|
|
with pytest.raises(ValidationError, match="قبلاً"):
|
|
unit.full_clean()
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_new_ticket_sms_is_queued_for_each_unit_supervisor_only_on_creation(category, customer, monkeypatch):
|
|
from apps.notifications.management.commands.process_notifications import _send_sms
|
|
|
|
unit = category.unit
|
|
unit.supervisor_mobile_1 = "09121112233"
|
|
unit.supervisor_mobile_2 = "09352223344"
|
|
unit.supervisor_mobile_3 = "09903334455"
|
|
unit.save()
|
|
ticket = create_ticket(customer=customer, category=category, subject=" پیگیری بارگیری ", body="متن")
|
|
jobs = list(NotificationJob.objects.filter(ticket=ticket, event_type="UNIT_SUPERVISOR_NEW_TICKET").order_by("id"))
|
|
assert [job.payload["recipient_mobile"] for job in jobs] == list(unit.supervisor_mobiles)
|
|
|
|
sent = []
|
|
|
|
class Provider:
|
|
def send_message(self, mobile, message):
|
|
sent.append((mobile, message))
|
|
return type("Result", (), {"success": True})()
|
|
|
|
monkeypatch.setattr("apps.notifications.management.commands.process_notifications.get_sms_provider", lambda: Provider())
|
|
for job in jobs:
|
|
result, issued_link = _send_sms(job)
|
|
assert result.success and issued_link is None
|
|
|
|
assert [mobile for mobile, _ in sent] == list(unit.supervisor_mobiles)
|
|
for _, message in sent:
|
|
assert customer.local_mobile in message
|
|
assert "موضوع: پیگیری بارگیری" in message
|
|
assert f"دسته: {category.title}" in message
|
|
|
|
supervisor_job_count = NotificationJob.objects.filter(event_type="UNIT_SUPERVISOR_NEW_TICKET").count()
|
|
reply_ticket(ticket_id=ticket.pk, user=customer, body="پاسخ جدید")
|
|
assert NotificationJob.objects.filter(event_type="UNIT_SUPERVISOR_NEW_TICKET").count() == supervisor_job_count
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_staff_created_ticket_also_notifies_owning_unit_supervisors(category, customer):
|
|
category.unit.supervisor_mobile_1 = "09124445566"
|
|
category.unit.save()
|
|
staff = make_staff("a")
|
|
|
|
ticket = create_ticket(
|
|
customer=customer,
|
|
created_by=staff,
|
|
unit=category.unit,
|
|
category=category,
|
|
subject="ثبت توسط کارشناس",
|
|
body="متن",
|
|
)
|
|
|
|
job = NotificationJob.objects.get(ticket=ticket, event_type="UNIT_SUPERVISOR_NEW_TICKET")
|
|
assert job.payload == {"recipient_mobile": "+989124445566"}
|
|
|
|
@pytest.mark.django_db
|
|
def test_customer_bale_replies_are_debounced_by_database_policy(category,customer):
|
|
from apps.notifications.models import NotificationPolicy
|
|
policy=NotificationPolicy.objects.get(key="CUSTOMER_REPLY_BALE")
|
|
policy.quiet_period_seconds=720; policy.save()
|
|
ticket=create_ticket(customer=customer,category=category,subject="چت مشتری",body="شروع")
|
|
first=reply_ticket(ticket_id=ticket.pk,user=customer,body="پیام اول")
|
|
job=NotificationJob.objects.get(ticket=ticket,event_type="CUSTOMER_TICKET_REPLY")
|
|
first_due=job.next_attempt_at
|
|
second=reply_ticket(ticket_id=ticket.pk,user=customer,body="پیام دوم")
|
|
job.refresh_from_db()
|
|
assert NotificationJob.objects.filter(ticket=ticket,event_type="CUSTOMER_TICKET_REPLY").count()==1
|
|
assert job.payload=={"message_id":second.pk,"message_count":2}
|
|
assert job.next_attempt_at>=first_due and job.next_attempt_at>=timezone.now()+timedelta(seconds=715)
|
|
|
|
@pytest.mark.django_db
|
|
def test_reopened_ticket_bale_notification_is_immediate(category,customer):
|
|
ticket=create_ticket(customer=customer,category=category,subject="بازگشایی",body="شروع")
|
|
Ticket.objects.filter(pk=ticket.pk).update(status=Ticket.Status.CLOSED)
|
|
message=reply_ticket(ticket_id=ticket.pk,user=customer,body="لطفاً دوباره بررسی شود")
|
|
job=NotificationJob.objects.get(ticket=ticket,event_type="CUSTOMER_TICKET_REOPENED")
|
|
assert job.payload=={"message_id":message.pk,"message_count":1}
|
|
assert job.next_attempt_at<=timezone.now()
|
|
|
|
@pytest.mark.django_db
|
|
def test_bale_worker_defers_active_customer_chat(category,customer):
|
|
from apps.notifications.management.commands.process_notifications import _prepare_customer_reply_bale
|
|
ticket=create_ticket(customer=customer,category=category,subject="گفتگوی فعال",body="شروع")
|
|
reply_ticket(ticket_id=ticket.pk,user=customer,body="هنوز در حال گفتگو")
|
|
job=NotificationJob.objects.get(ticket=ticket,event_type="CUSTOMER_TICKET_REPLY")
|
|
NotificationJob.objects.filter(pk=job.pk).update(status="PROCESSING",next_attempt_at=timezone.now())
|
|
assert _prepare_customer_reply_bale(job) is False
|
|
job.refresh_from_db()
|
|
assert job.status=="PENDING" and job.last_error=="customer_chat_active"
|
|
|
|
@pytest.mark.django_db
|
|
def test_bale_ticket_excerpt_is_compact_and_whitespace_normalized(category,customer,settings):
|
|
from apps.notifications.management.commands.process_notifications import _bale_ticket_text
|
|
settings.PUBLIC_BASE_URL="https://support.example.com"
|
|
long_body=("متن طولانی با فاصله\n"*30)
|
|
ticket=create_ticket(customer=customer,category=category,subject="خلاصه کوتاه",body=long_body)
|
|
text=_bale_ticket_text(ticket)
|
|
excerpt=text.split("📝 ",1)[1].split("\n",1)[0]
|
|
assert len(excerpt)<=280 and excerpt.endswith("…")
|
|
assert " " not in excerpt and "https://support.example.com/staff/tickets/" in text
|
|
|
|
@pytest.mark.django_db
|
|
def test_ticket_full_delete_requires_manager_permission_and_denies_sales_expert(category, customer):
|
|
ticket = create_ticket(customer=customer, category=category, subject="حذف پرونده", body="متن اولیه")
|
|
expert = User.objects.create_user(username="delete-expert", mobile="09120000003", account_type="STAFF")
|
|
StaffProfile.objects.create(user=expert, primary_unit=category.unit, display_name="کارشناس", role=StaffProfile.Role.SALES_EXPERT)
|
|
expert.user_permissions.add(Permission.objects.get(codename="delete_ticket", content_type__app_label="tickets"))
|
|
with pytest.raises(PermissionDenied):
|
|
delete_ticket(ticket_id=ticket.pk, user=expert)
|
|
manager = User.objects.create_user(username="delete-manager", mobile="09120000004", account_type="STAFF")
|
|
StaffProfile.objects.create(user=manager, primary_unit=category.unit, display_name="مدیر", role=StaffProfile.Role.SALES_MANAGER)
|
|
for codename in ("delete_ticket", "view_all_tickets"):
|
|
manager.user_permissions.add(Permission.objects.get(codename=codename, content_type__app_label="tickets"))
|
|
manager.unit_memberships.get(unit=category.unit).permissions.add(
|
|
Permission.objects.get(codename="delete_ticket", content_type__app_label="tickets")
|
|
)
|
|
delete_ticket(ticket_id=ticket.pk, user=manager)
|
|
assert not Ticket.objects.filter(pk=ticket.pk).exists()
|
|
assert not NotificationJob.objects.filter(ticket_id=ticket.pk).exists()
|
|
|
|
@pytest.mark.django_db
|
|
@override_settings(DEBUG=True, DEV_OTP_MOBILE="+989202202129", DEV_OTP_CODE="12345")
|
|
def test_fixed_otp_is_available_for_development_only():
|
|
from apps.accounts.models import OTPChallenge
|
|
challenge, code = OTPChallenge.issue("09202202129")
|
|
assert code == "12345"
|
|
assert challenge.verify("12345")
|
|
|
|
@pytest.mark.django_db
|
|
@override_settings(DEBUG=False, DEV_OTP_MOBILE="+989202202129", DEV_OTP_CODE="12345")
|
|
def test_fixed_otp_is_disabled_outside_development():
|
|
from apps.accounts.models import OTPChallenge
|
|
_, code = OTPChallenge.issue("09202202129")
|
|
assert code != "12345"
|
|
|
|
@pytest.mark.django_db
|
|
def test_customer_cannot_view_others_ticket(category, customer):
|
|
other = User.objects.create_user(username="other", mobile="09121111111", account_type="CUSTOMER")
|
|
ticket = create_ticket(customer=customer, category=category, subject="موضوع معتبر", body="متن")
|
|
assert TicketPolicy.can_view(customer, ticket)
|
|
assert not TicketPolicy.can_view(other, ticket)
|
|
|
|
@pytest.mark.django_db
|
|
def test_selected_expert_is_private(category, customer):
|
|
a, b = make_staff("a"), make_staff("b")
|
|
ticket = create_ticket(customer=customer, category=category, subject="موضوع خصوصی", selected_expert=a, body="متن")
|
|
assert TicketPolicy.can_view(a, ticket)
|
|
assert not TicketPolicy.can_view(b, ticket)
|
|
|
|
@pytest.mark.django_db
|
|
def test_claim_locks_ticket(category, customer):
|
|
a, b = make_staff("a"), make_staff("b")
|
|
ticket = create_ticket(customer=customer, category=category, subject="موضوع عمومی", body="متن")
|
|
claim_ticket(ticket.pk, a)
|
|
ticket.refresh_from_db()
|
|
assert ticket.owner == a and ticket.status == Ticket.Status.OPEN
|
|
with pytest.raises(PermissionDenied):
|
|
claim_ticket(ticket.pk, b)
|
|
assert not TicketPolicy.can_view(b, ticket)
|
|
|
|
@pytest.mark.django_db
|
|
def test_closed_customer_reply_reopens(category, customer):
|
|
ticket = create_ticket(customer=customer, category=category, subject="موضوع بسته", body="متن")
|
|
ticket.status = Ticket.Status.CLOSED
|
|
ticket.save()
|
|
reply_ticket(ticket_id=ticket.pk, user=customer, body="دوباره نیاز دارم")
|
|
ticket.refresh_from_db()
|
|
assert ticket.status == Ticket.Status.REOPENED
|
|
|
|
@pytest.mark.django_db
|
|
def test_staff_reply_waits_customer(category, customer):
|
|
a = make_staff("a")
|
|
ticket = create_ticket(customer=customer, category=category, subject="پاسخ کارشناس", body="متن")
|
|
claim_ticket(ticket.pk, a)
|
|
reply_ticket(ticket_id=ticket.pk, user=a, body="پاسخ")
|
|
ticket.refresh_from_db()
|
|
assert ticket.status == Ticket.Status.WAITING_CUSTOMER
|
|
assert ticket.first_response_at
|
|
|
|
@pytest.mark.django_db
|
|
def test_attachment_uuid_path_fits_storage_field(tmp_path, category, customer):
|
|
AttachmentPolicy.objects.create(attachment_type="IMAGE", max_file_size=10 * 1024 * 1024, allowed_extensions=[".png"])
|
|
png=base64.b64decode("iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mNk+A8AAQUBAScY42YAAAAASUVORK5CYII=")
|
|
upload = SimpleUploadedFile("customer-image.png", png, content_type="image/png")
|
|
with override_settings(MEDIA_ROOT=tmp_path):
|
|
ticket = create_ticket(customer=customer, category=category, subject="درخواست تصویری", body="تصویر پیوست شد", files=[upload])
|
|
attachment = ticket.ticket_messages.get().attachments.get()
|
|
assert attachment.file.name.startswith("tickets/")
|
|
assert len(attachment.file.name) <= 500
|
|
assert attachment.file.storage.exists(attachment.file.name)
|
|
|
|
@pytest.mark.django_db
|
|
def test_spoofed_image_is_rejected(category, customer):
|
|
from django.core.exceptions import ValidationError
|
|
AttachmentPolicy.objects.create(attachment_type="IMAGE",max_file_size=1024,allowed_extensions=[".png"])
|
|
upload=SimpleUploadedFile("fake.png",b"not an image",content_type="image/png")
|
|
with pytest.raises(ValidationError): create_ticket(customer=customer,category=category,subject="تصویر نامعتبر",body="فایل",files=[upload])
|
|
|
|
@pytest.mark.django_db
|
|
def test_selected_expert_can_transfer_and_old_access_is_removed(category,customer):
|
|
from apps.tickets.services import transfer_ticket
|
|
a,b=make_staff("a"),make_staff("b")
|
|
for user in (a,b): user.user_permissions.add(Permission.objects.get(codename="transfer_ticket",content_type__app_label="tickets"))
|
|
ticket=create_ticket(customer=customer,category=category,subject="انتقال انتخابی",selected_expert=a,body="متن")
|
|
transfer_ticket(ticket.pk,a,b); ticket.refresh_from_db()
|
|
assert ticket.owner==b and ticket.selected_expert is None
|
|
assert not TicketPolicy.can_view(a,ticket) and TicketPolicy.can_view(b,ticket)
|
|
|
|
@pytest.mark.django_db
|
|
def test_internal_note_is_hidden_and_does_not_enqueue_sms(category,customer):
|
|
from apps.notifications.models import NotificationJob
|
|
a=make_staff("a"); a.user_permissions.add(Permission.objects.get(codename="add_internal_note",content_type__app_label="messaging"))
|
|
ticket=create_ticket(customer=customer,category=category,subject="یادداشت داخلی",body="متن"); claim_ticket(ticket.pk,a)
|
|
sms_count=NotificationJob.objects.filter(channel="SMS").count()
|
|
msg=reply_ticket(ticket_id=ticket.pk,user=a,body="محرمانه",internal=True)
|
|
assert msg.is_internal and NotificationJob.objects.filter(channel="SMS").count()==sms_count
|
|
|
|
@pytest.mark.django_db
|
|
def test_customer_can_soft_delete_own_message_but_not_another_authors(category,customer):
|
|
from apps.messaging.services import soft_delete_message
|
|
staff=make_staff("a")
|
|
ticket=create_ticket(customer=customer,category=category,subject="حذف پیام مشتری",body="پیام خود مشتری")
|
|
own=ticket.ticket_messages.get()
|
|
deleted=soft_delete_message(own.pk,customer)
|
|
assert deleted.deleted_at is not None and deleted.deleted_by==customer
|
|
staff_message=reply_ticket(ticket_id=ticket.pk,user=staff,body="پاسخ کارشناس")
|
|
with pytest.raises(PermissionDenied): soft_delete_message(staff_message.pk,customer)
|
|
|
|
@pytest.mark.django_db
|
|
def test_staff_reply_sms_is_debounced_using_database_policy(category,customer):
|
|
from apps.notifications.models import NotificationJob,NotificationPolicy
|
|
from django.utils import timezone
|
|
policy=NotificationPolicy.objects.get(key="STAFF_REPLY_SMS")
|
|
policy.quiet_period_seconds=420; policy.save()
|
|
a=make_staff("a"); ticket=create_ticket(customer=customer,category=category,subject="کنترل پیامک پاسخ",selected_expert=a,body="متن")
|
|
first=reply_ticket(ticket_id=ticket.pk,user=a,body="پاسخ اول")
|
|
job=NotificationJob.objects.get(ticket=ticket,event_type="STAFF_REPLY")
|
|
first_due=job.next_attempt_at
|
|
second=reply_ticket(ticket_id=ticket.pk,user=a,body="پاسخ دوم")
|
|
job.refresh_from_db()
|
|
assert NotificationJob.objects.filter(ticket=ticket,event_type="STAFF_REPLY").count()==1
|
|
assert job.payload=={"latest_message_id":second.pk} and job.next_attempt_at>=first_due
|
|
assert job.next_attempt_at>=timezone.now()+timedelta(seconds=415)
|
|
|
|
@pytest.mark.django_db
|
|
def test_customer_reply_cancels_pending_staff_sms(category,customer):
|
|
a=make_staff("a"); ticket=create_ticket(customer=customer,category=category,subject="گفت‌وگوی فعال",selected_expert=a,body="متن")
|
|
reply_ticket(ticket_id=ticket.pk,user=a,body="پاسخ کارشناس")
|
|
reply_ticket(ticket_id=ticket.pk,user=customer,body="پاسخ سریع مشتری")
|
|
job=NotificationJob.objects.get(ticket=ticket,event_type="STAFF_REPLY")
|
|
assert job.status==NotificationJob.Status.CANCELLED and job.last_error=="customer_replied"
|
|
|
|
@pytest.mark.django_db
|
|
def test_due_unread_staff_reply_is_sent_once(category,customer,monkeypatch):
|
|
from django.core.management import call_command
|
|
from apps.integrations.sms.base import SMSResult
|
|
from apps.notifications.management.commands import process_notifications
|
|
a=make_staff("a"); ticket=create_ticket(customer=customer,category=category,subject="پاسخ خوانده نشده",selected_expert=a,body="متن")
|
|
reply_ticket(ticket_id=ticket.pk,user=a,body="پاسخ کارشناس")
|
|
job=NotificationJob.objects.get(ticket=ticket,event_type="STAFF_REPLY")
|
|
NotificationJob.objects.filter(pk=job.pk).update(next_attempt_at=timezone.now()-timedelta(seconds=1))
|
|
ticket.last_activity_at=timezone.now()-timedelta(hours=1); ticket.save(update_fields=["last_activity_at"])
|
|
sent=[]
|
|
class Provider:
|
|
def send_message(self,mobile,message): sent.append((mobile,message)); return SMSResult(True,"ok")
|
|
monkeypatch.setattr(process_notifications,"get_sms_provider",lambda:Provider())
|
|
call_command("process_notifications",once=True,job_id=job.pk)
|
|
job.refresh_from_db()
|
|
assert job.status==NotificationJob.Status.SENT and len(sent)==1
|
|
call_command("process_notifications",once=True,job_id=job.pk)
|
|
assert len(sent)==1
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_every_failed_sms_gets_exactly_three_retries_three_minutes_apart(category,customer,monkeypatch,settings):
|
|
from django.core.management import call_command
|
|
from apps.integrations.sms.base import SMSResult
|
|
from apps.notifications.management.commands import process_notifications
|
|
settings.SMS_MAX_RETRIES=3
|
|
settings.SMS_RETRY_DELAY_SECONDS=180
|
|
ticket=create_ticket(customer=customer,category=category,subject="آزمون retry پیامک",body="متن")
|
|
job=NotificationJob.objects.get(ticket=ticket,event_type="CUSTOMER_TICKET_CREATED")
|
|
calls=[]
|
|
class Provider:
|
|
def send_message(self,mobile,message):
|
|
calls.append((mobile,message))
|
|
return SMSResult(False,error="0098SMS [کد -19]: شارژ پنل برای ارسال کافی نیست.",provider_code="-19")
|
|
monkeypatch.setattr(process_notifications,"get_sms_provider",lambda:Provider())
|
|
|
|
for attempt in range(1,5):
|
|
before=timezone.now()
|
|
call_command("process_notifications",once=True,job_id=job.pk)
|
|
job.refresh_from_db()
|
|
assert job.attempts==attempt
|
|
assert job.provider_status_code=="-19"
|
|
assert "کد -19" in job.last_error
|
|
if attempt<4:
|
|
assert job.status==NotificationJob.Status.RETRYING
|
|
assert timedelta(seconds=179) <= job.next_attempt_at-before <= timedelta(seconds=181)
|
|
NotificationJob.objects.filter(pk=job.pk).update(next_attempt_at=timezone.now()-timedelta(seconds=1))
|
|
else:
|
|
assert job.status==NotificationJob.Status.FAILED
|
|
|
|
call_command("process_notifications",once=True,job_id=job.pk)
|
|
assert len(calls)==4
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_worker_persists_0098_delivery_result(category,customer,monkeypatch,settings):
|
|
from django.core.management import call_command
|
|
from apps.integrations.sms.base import SMSDeliveryResult
|
|
from apps.notifications.management.commands import process_notifications
|
|
ticket=create_ticket(customer=customer,category=category,subject="استعلام دلیوری",body="متن")
|
|
job=NotificationJob.objects.get(ticket=ticket,event_type="CUSTOMER_TICKET_CREATED")
|
|
NotificationJob.objects.filter(pk=job.pk).update(
|
|
status=NotificationJob.Status.SENT,
|
|
provider_message_id="123456789",
|
|
delivery_status=NotificationJob.DeliveryStatus.PENDING,
|
|
delivery_next_check_at=timezone.now()-timedelta(seconds=1),
|
|
)
|
|
class Provider:
|
|
def get_delivery_status(self,provider_id):
|
|
assert provider_id=="123456789"
|
|
return SMSDeliveryResult("DELIVERED","5","0098SMS [کد 5]: پیامک به گوشی گیرنده رسیده است.")
|
|
monkeypatch.setattr(process_notifications,"get_sms_provider",lambda:Provider())
|
|
|
|
call_command("process_notifications",once=True,job_id=job.pk)
|
|
job.refresh_from_db()
|
|
assert job.delivery_status==NotificationJob.DeliveryStatus.DELIVERED
|
|
assert job.delivery_status_code=="5"
|
|
assert job.delivery_next_check_at is None
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_worker_rechecks_provisional_delivery_until_definitive_result(category,customer,monkeypatch,settings):
|
|
from django.core.management import call_command
|
|
from apps.integrations.sms.base import SMSDeliveryResult
|
|
from apps.notifications.management.commands import process_notifications
|
|
settings.SMS_DELIVERY_CHECK_DELAY_SECONDS=180
|
|
settings.SMS_DELIVERY_MAX_CHECKS=15
|
|
ticket=create_ticket(customer=customer,category=category,subject="پیگیری طولانی دلیوری",body="متن")
|
|
job=NotificationJob.objects.get(ticket=ticket,event_type="CUSTOMER_TICKET_CREATED")
|
|
NotificationJob.objects.filter(pk=job.pk).update(
|
|
status=NotificationJob.Status.SENT, sent_at=timezone.now(), provider_message_id="123456789",
|
|
delivery_status=NotificationJob.DeliveryStatus.PENDING,
|
|
delivery_next_check_at=timezone.now()-timedelta(seconds=1),
|
|
)
|
|
results=[
|
|
SMSDeliveryResult("UNKNOWN","8","وضعیت هنوز مشخص نشده است.",is_final=False),
|
|
SMSDeliveryResult("UNDELIVERED","6","تحویل هنوز تأیید نشده است.",is_final=False),
|
|
SMSDeliveryResult("DELIVERED","5","پیامک به گوشی رسیده است."),
|
|
]
|
|
class Provider:
|
|
def get_delivery_status(self,provider_id): return results.pop(0)
|
|
provider=Provider()
|
|
monkeypatch.setattr(process_notifications,"get_sms_provider",lambda:provider)
|
|
|
|
for expected_status,expected_attempts in ((NotificationJob.DeliveryStatus.UNKNOWN,1),(NotificationJob.DeliveryStatus.UNDELIVERED,2)):
|
|
before=timezone.now()
|
|
call_command("process_notifications",once=True,job_id=job.pk)
|
|
job.refresh_from_db()
|
|
assert job.delivery_status==expected_status
|
|
assert job.delivery_check_attempts==expected_attempts
|
|
assert timedelta(seconds=179) <= job.delivery_next_check_at-before <= timedelta(seconds=181)
|
|
NotificationJob.objects.filter(pk=job.pk).update(delivery_next_check_at=timezone.now()-timedelta(seconds=1))
|
|
|
|
call_command("process_notifications",once=True,job_id=job.pk)
|
|
job.refresh_from_db()
|
|
assert job.delivery_status==NotificationJob.DeliveryStatus.DELIVERED
|
|
assert job.delivery_check_attempts==3 and job.delivery_next_check_at is None
|
|
|
|
@pytest.mark.django_db
|
|
def test_customer_ticket_created_sms_has_seven_day_magic_link(category,customer,settings,monkeypatch):
|
|
from datetime import timedelta
|
|
from django.utils import timezone
|
|
from apps.accounts.models import TicketMagicLink
|
|
from apps.integrations.sms.base import SMSResult
|
|
from apps.notifications.management.commands import process_notifications
|
|
from apps.notifications.models import NotificationJob
|
|
settings.PUBLIC_BASE_URL="https://support.example.com"
|
|
settings.MAGIC_LINK_TTL_DAYS=7
|
|
ticket=create_ticket(customer=customer,category=category,subject="پیام تشکر",body="متن")
|
|
job=NotificationJob.objects.get(ticket=ticket,event_type="CUSTOMER_TICKET_CREATED")
|
|
captured={}
|
|
class Provider:
|
|
def send_message(self,mobile,message): captured.update(mobile=mobile,message=message); return SMSResult(True,"ok")
|
|
monkeypatch.setattr(process_notifications,"get_sms_provider",lambda:Provider())
|
|
before=timezone.now()
|
|
result,link=process_notifications._send_sms(job)
|
|
assert result.success and link.created_reason=="customer_created_ticket"
|
|
assert before+timedelta(days=7)-timedelta(seconds=1) <= link.expires_at <= before+timedelta(days=7)+timedelta(seconds=1)
|
|
assert "از ثبت درخواست شما متشکریم" in captured["message"]
|
|
assert "پس از بررسی و پاسخ کارشناس، از طریق پیامک" in captured["message"]
|
|
assert "https://support.example.com/t/" in captured["message"]
|
|
|
|
@pytest.mark.django_db
|
|
def test_sms_templates_are_database_managed_and_distinguish_ticket_creator():
|
|
from apps.notifications.models import SMSTemplate
|
|
customer_text=SMSTemplate.objects.get(event_type="CUSTOMER_TICKET_CREATED").body
|
|
staff_text=SMSTemplate.objects.get(event_type="STAFF_TICKET_CREATED").body
|
|
assert customer_text!=staff_text
|
|
assert "از ثبت درخواست شما متشکریم" in customer_text
|
|
assert "توسط واحد {unit_name} برای شما ثبت شد" in staff_text
|
|
assert "{magic_link}" in customer_text and "{first_name}" in customer_text
|
|
|
|
@pytest.mark.django_db(transaction=True)
|
|
def test_magic_link_is_ten_character_base36_and_retries_collision_without_select(category,customer,monkeypatch):
|
|
import re
|
|
from django.test.utils import CaptureQueriesContext
|
|
from django.db import connection
|
|
from django.utils import timezone
|
|
from apps.accounts.models import TicketMagicLink
|
|
ticket=create_ticket(customer=customer,category=category,subject="توکن کوتاه",body="متن")
|
|
TicketMagicLink.objects.create(ticket=ticket,customer=customer,token_hash=TicketMagicLink.digest("aaaaaaaaaa"),expires_at=timezone.now()+__import__('datetime').timedelta(days=7))
|
|
values=iter(["aaaaaaaaaa","b2c3d4e5f6"])
|
|
monkeypatch.setattr(TicketMagicLink,"generate_secure_token",staticmethod(lambda length=10:next(values)))
|
|
with CaptureQueriesContext(connection) as queries: token=TicketMagicLink.issue(ticket)
|
|
assert token=="b2c3d4e5f6" and re.fullmatch(r"[a-z0-9]{10}",token)
|
|
assert not any(query["sql"].lstrip().upper().startswith("SELECT") for query in queries)
|
|
|
|
@pytest.mark.django_db
|
|
def test_audio_only_webm_is_saved_as_voice(category,customer,monkeypatch,tmp_path,settings):
|
|
from apps.attachments.services import save_attachments
|
|
settings.MEDIA_ROOT=tmp_path
|
|
AttachmentPolicy.objects.update_or_create(attachment_type="VOICE",defaults={"max_file_size":1024*1024,"allowed_extensions":[".webm"],"allowed_mime_types":[]})
|
|
ticket=create_ticket(customer=customer,category=category,subject="ضبط صدا",body="متن")
|
|
message=ticket.ticket_messages.first()
|
|
monkeypatch.setattr("apps.attachments.services.magic.from_buffer",lambda *args,**kwargs:"video/webm")
|
|
upload=SimpleUploadedFile("voice.webm",b"webm-audio-content",content_type="audio/webm")
|
|
attachment=save_attachments(message,[upload],customer)[0]
|
|
assert attachment.attachment_type=="VOICE" and attachment.extension==".webm"
|
|
|
|
@pytest.mark.django_db
|
|
def test_webotp_sms_format_is_origin_bound(settings):
|
|
from apps.integrations.sms.providers import webotp_message
|
|
settings.WEBOTP_HOST="example.com"
|
|
assert webotp_message("123456")=="کد شما: 123456\n@example.com #123456"
|
|
|
|
@override_settings(SMS_PROVIDER_CLASS="apps.integrations.sms.sms0098.ZeroZeroNinetyEightSMSProvider")
|
|
def test_sms_provider_contract_and_default(settings):
|
|
from apps.integrations.sms.base import SMSProvider
|
|
from apps.integrations.sms.sms0098 import ZeroZeroNinetyEightSMSProvider
|
|
from apps.integrations.sms.factory import get_sms_provider,reset_sms_provider
|
|
assert issubclass(ZeroZeroNinetyEightSMSProvider,SMSProvider)
|
|
reset_sms_provider()
|
|
assert isinstance(get_sms_provider(),ZeroZeroNinetyEightSMSProvider)
|
|
reset_sms_provider()
|
|
|
|
def test_behin_provider_maps_send_array_contract(settings):
|
|
from apps.integrations.sms.behin import BehinSMSProvider
|
|
settings.BEHIN_SMS_USERNAME="user"; settings.BEHIN_SMS_PASSWORD="pass"; settings.BEHIN_SMS_SPECIAL_NUMBER="3000"
|
|
calls={}
|
|
class Service:
|
|
def SendArray(self,**kwargs): calls.update(kwargs); return [987654]
|
|
class Client: service=Service()
|
|
provider=BehinSMSProvider(); provider._client=Client()
|
|
result=provider.send_message("+989202202129","پیام آزمایشی")
|
|
assert result.success and "987654" in result.provider_id
|
|
assert calls["RecipientNumber"]=={"string":["+989202202129"]}
|
|
assert calls["CheckingMessageID"]=={"long":[0]}
|
|
assert calls["SpecialNumber"]=="3000" and calls["IsFlashMessage"] is False
|
|
|
|
def test_0098_provider_maps_soap_contract(settings):
|
|
from apps.integrations.sms.sms0098 import ZeroZeroNinetyEightSMSProvider
|
|
settings.SMS0098_USERNAME="user"; settings.SMS0098_PASSWORD="pass"; settings.SMS0098_FROM="3000"
|
|
calls={}
|
|
class Service:
|
|
def SendSMSWithID(self,**parameters): calls.update(parameters); return "123456789"
|
|
class Client: service=Service()
|
|
provider=ZeroZeroNinetyEightSMSProvider(); provider._client=Client()
|
|
result=provider.send_message("09202202129","پیام آزمایشی")
|
|
assert result.success and result.provider_id=="123456789"
|
|
assert calls["mobileno"]=="09202202129"
|
|
assert calls["text"]=="پیام آزمایشی"
|
|
assert calls["pnlno"]=="3000"
|
|
assert set(calls)=={"username","password","text","mobileno","pnlno"}
|
|
|
|
|
|
@pytest.mark.parametrize("code,expected", [
|
|
("-3", "تطابق ندارند"), ("10", "اشتباه است"), ("11", "کاراکتر غیرمجاز"),
|
|
("-17", "خالی است"), ("-18", "خطای شارژ"), ("-19", "شارژ پنل"),
|
|
("-22", "شماره موبایل"), ("66", "تطابق ندارند"), ("1111", "کاراکتر غیرمجاز"),
|
|
("Hang", "مسدود است"), ("No Doc", "مرحله دوم"), ("-23", "مجوز ارسال"),
|
|
("-24", "شماره پنل"), ("-25", "اعتبار شماره پنل"), ("-26", "تنظیمات کد OTP"),
|
|
("-27", "فرمت کد OTP"), ("-28", "تأیید نشده"), ("Wrong timestamp", "timestamp"),
|
|
("Wrong time", "خارج از بازه"), ("-29", "فرمت کد OTP"), ("-31", "یافت نشد"),
|
|
("-32", "مطابقت ندارد"), ("13", "IP غیرمجاز"),
|
|
])
|
|
def test_0098_provider_maps_every_documented_send_error(settings,code,expected):
|
|
from apps.integrations.sms.sms0098 import ZeroZeroNinetyEightSMSProvider
|
|
settings.SMS0098_USERNAME="user"; settings.SMS0098_PASSWORD="pass"; settings.SMS0098_FROM="3000"
|
|
provider=ZeroZeroNinetyEightSMSProvider()
|
|
result=provider._parse_send_result(code)
|
|
assert not result.success
|
|
assert result.provider_code==code
|
|
assert expected in result.error
|
|
|
|
|
|
@pytest.mark.parametrize("code,status,is_final", [("4","BLOCKED",True),("5","DELIVERED",True),("6","UNDELIVERED",False),("7","EXPIRED",True),("8","UNKNOWN",False),("9","ERROR",True),("12","ERROR",True),("13","ERROR",True)])
|
|
def test_0098_provider_maps_documented_delivery_states(settings,code,status,is_final):
|
|
from apps.integrations.sms.sms0098 import ZeroZeroNinetyEightSMSProvider
|
|
settings.SMS0098_USERNAME="user"; settings.SMS0098_PASSWORD="pass"; settings.SMS0098_FROM="3000"
|
|
class Service:
|
|
def smsdeliveryState(self,**parameters): return code
|
|
class Client: service=Service()
|
|
provider=ZeroZeroNinetyEightSMSProvider(); provider._client=Client()
|
|
result=provider.get_delivery_status("123456789")
|
|
assert result.status==status and result.provider_code==code and result.is_final is is_final
|
|
assert f"کد {code}" in result.description
|