2569 lines
116 KiB
Python
2569 lines
116 KiB
Python
"""تست‌های قابلیت پیامک: نرمال‌سازی، idempotency، opt-out، worker و تریگرها."""
|
||
import json
|
||
import re
|
||
from datetime import timedelta
|
||
from unittest import mock
|
||
|
||
from django.core.management import call_command
|
||
from django.test import TestCase
|
||
from django.urls import reverse
|
||
from django.utils import timezone
|
||
|
||
from admin_panel.models import (Article, Customer, CustomerSMSSetting,
|
||
PhoneNumber, Sarresid, SMSDedup, SMSLog,
|
||
SMSProviderConfig, SMSQueue, SMSTemplate, SMSType)
|
||
from admin_panel.sms.enqueue import enqueue_sms
|
||
from admin_panel.sms.phones import customer_phone_entries
|
||
from admin_panel.sms.recipients import resolve_recipients
|
||
from admin_panel.sms.rules import is_sms_enabled
|
||
from admin_panel.sms.scanner import run_all
|
||
from admin_panel.sms.timing import apply_timing, current_timing_value, timing_for
|
||
from admin_panel.sms.triggers import notify_welcome
|
||
from integrations.sms.base import SMSProvider, SMSResult
|
||
from integrations.sms.normalize import normalize_mobile, to_local
|
||
|
||
from .factories import (login_superuser, make_customer, make_product, make_store,
|
||
make_user, stock_in)
|
||
|
||
|
||
def mark_primary(customer, *numbers):
|
||
"""تعیین شماره‌های اصلی از دید کاربر پنل (بر مبنای مقدارِ خودِ شماره)."""
|
||
from admin_panel.sms.phones import normalize_primary_values
|
||
|
||
customer.sms_primary_phones = normalize_primary_values(numbers)
|
||
customer.save(update_fields=['sms_primary_phones'])
|
||
|
||
|
||
def activate_sms(*, since=None, allow_backlog=False, backlog_max=0, max_per_run=200,
|
||
interval=0):
|
||
"""تنظیمات جریان را طوری می‌چیند که گویا قابلیت از مدتی قبل فعال بوده است."""
|
||
config = SMSProviderConfig.load()
|
||
config.scan_since = since if since is not None else (timezone.now() - timedelta(days=365))
|
||
config.allow_backlog = allow_backlog
|
||
config.backlog_max = backlog_max
|
||
config.max_per_run = max_per_run
|
||
config.send_interval_seconds = interval
|
||
config.provider = config.provider or 'console'
|
||
config.save()
|
||
return config
|
||
|
||
|
||
class NormalizeTests(TestCase):
|
||
def test_variants_normalize_to_canonical(self):
|
||
for raw in ('09120000000', '9120000000', '989120000000',
|
||
'+989120000000', '00989120000000', '۰۹۱۲۰۰۰۰۰۰۰'):
|
||
self.assertEqual(normalize_mobile(raw), '+989120000000', msg=raw)
|
||
|
||
def test_invalid_returns_empty(self):
|
||
for raw in ('', None, '123', 'abcd', '0811234567'):
|
||
self.assertEqual(normalize_mobile(raw), '', msg=repr(raw))
|
||
|
||
def test_to_local(self):
|
||
self.assertEqual(to_local('+989120000000'), '09120000000')
|
||
|
||
|
||
class RecipientTests(TestCase):
|
||
"""یافتن شماره‌ها از همهٔ روش‌های ذخیره‌سازی + انتخاب شمارهٔ اصلی."""
|
||
|
||
def test_primary_phones_are_used(self):
|
||
customer = make_customer()
|
||
PhoneNumber.objects.create(customer=customer, phone_number='09120000001')
|
||
PhoneNumber.objects.create(customer=customer, phone_number='09120000002')
|
||
mark_primary(customer, '09120000002')
|
||
self.assertEqual(resolve_recipients(customer), ['+989120000002'])
|
||
|
||
def test_fallback_to_first_number_when_no_primary(self):
|
||
customer = make_customer()
|
||
PhoneNumber.objects.create(customer=customer, phone_number='09120000001')
|
||
PhoneNumber.objects.create(customer=customer, phone_number='09120000002')
|
||
self.assertEqual(resolve_recipients(customer), ['+989120000001'])
|
||
|
||
def test_multiple_primary_numbers(self):
|
||
customer = make_customer()
|
||
PhoneNumber.objects.create(customer=customer, phone_number='09120000001')
|
||
PhoneNumber.objects.create(customer=customer, phone_number='09120000002')
|
||
mark_primary(customer, '09120000001', '09120000002')
|
||
self.assertEqual(
|
||
sorted(resolve_recipients(customer)),
|
||
['+989120000001', '+989120000002'],
|
||
)
|
||
|
||
def test_legacy_phone_fields_are_discovered(self):
|
||
"""شماره‌های ذخیره‌شده در فیلدهای قدیمی phone1/2/3 هم پیدا می‌شوند."""
|
||
customer = make_customer()
|
||
customer.phone1 = '9133722815' # بدون صفر ابتدایی
|
||
customer.phone2 = '09121112222'
|
||
customer.save(update_fields=['phone1', 'phone2'])
|
||
entries = customer_phone_entries(customer)
|
||
self.assertEqual(
|
||
[e.normalized for e in entries],
|
||
['+989133722815', '+989121112222'],
|
||
)
|
||
self.assertEqual(resolve_recipients(customer), ['+989133722815'])
|
||
|
||
def test_phone_number_rows_and_legacy_fields_are_merged_without_duplicates(self):
|
||
customer = make_customer()
|
||
customer.phone1 = '9133722815'
|
||
customer.save(update_fields=['phone1'])
|
||
PhoneNumber.objects.create(customer=customer, phone_number='09133722815')
|
||
entries = customer_phone_entries(customer)
|
||
self.assertEqual(len(entries), 1)
|
||
self.assertEqual(entries[0].normalized, '+989133722815')
|
||
|
||
def test_meaningless_values_are_discarded(self):
|
||
"""مقدارهای بی‌معنی مثل '0' یا شمارهٔ ثابت نباید گیرنده شوند."""
|
||
customer = make_customer()
|
||
customer.phone1 = '0'
|
||
customer.phone2 = '27'
|
||
customer.phone3 = '57387608'
|
||
customer.save(update_fields=['phone1', 'phone2', 'phone3'])
|
||
self.assertEqual(resolve_recipients(customer), [])
|
||
self.assertTrue(all(not e.is_mobile for e in customer_phone_entries(customer)))
|
||
|
||
def test_display_adds_leading_zero(self):
|
||
customer = make_customer()
|
||
customer.phone1 = '9133722815'
|
||
customer.save(update_fields=['phone1'])
|
||
self.assertEqual(customer_phone_entries(customer)[0].display, '09133722815')
|
||
|
||
def test_primary_selection_self_heals_when_number_changes(self):
|
||
"""اگر شمارهٔ اصلی ویرایش/حذف شود، سیستم خودکار به شمارهٔ اول برمی‌گردد."""
|
||
customer = make_customer()
|
||
PhoneNumber.objects.create(customer=customer, phone_number='09120000001')
|
||
mark_primary(customer, '09120000003')
|
||
PhoneNumber.objects.create(customer=customer, phone_number='09120000002')
|
||
# شمارهٔ اصلی دیگر روی مشتری نیست ⇒ اولین شمارهٔ معتبر گیرنده می‌شود.
|
||
self.assertEqual(resolve_recipients(customer), ['+989120000001'])
|
||
|
||
|
||
class RuleTests(TestCase):
|
||
def test_opt_out_model(self):
|
||
customer = make_customer()
|
||
self.assertTrue(is_sms_enabled(customer, SMSType.WELCOME))
|
||
CustomerSMSSetting.objects.create(
|
||
customer=customer, sms_type=SMSType.WELCOME, is_enabled=False,
|
||
)
|
||
self.assertFalse(is_sms_enabled(customer, SMSType.WELCOME))
|
||
|
||
def test_enqueue_respects_opt_out(self):
|
||
customer = make_customer()
|
||
PhoneNumber.objects.create(customer=customer, phone_number='09120000001')
|
||
CustomerSMSSetting.objects.create(
|
||
customer=customer, sms_type=SMSType.WELCOME, is_enabled=False,
|
||
)
|
||
created = enqueue_sms(
|
||
customer=customer, sms_type=SMSType.WELCOME,
|
||
context={'name': customer.name_family}, dedup_key='welcome:test',
|
||
)
|
||
self.assertEqual(created, 0)
|
||
self.assertEqual(SMSQueue.objects.count(), 0)
|
||
|
||
|
||
class EnqueueIdempotencyTests(TestCase):
|
||
def setUp(self):
|
||
self.customer = make_customer()
|
||
PhoneNumber.objects.create(
|
||
customer=self.customer, phone_number='09120000001',
|
||
)
|
||
|
||
def test_same_dedup_key_creates_one_row(self):
|
||
for _ in range(2):
|
||
enqueue_sms(
|
||
customer=self.customer, sms_type=SMSType.WELCOME,
|
||
context={'name': self.customer.name_family}, dedup_key='welcome:1',
|
||
)
|
||
self.assertEqual(SMSQueue.objects.count(), 1)
|
||
|
||
def test_multi_recipient_dedup_per_number(self):
|
||
PhoneNumber.objects.create(
|
||
customer=self.customer, phone_number='09120000002',
|
||
)
|
||
mark_primary(self.customer, '09120000001', '09120000002')
|
||
enqueue_sms(
|
||
customer=self.customer, sms_type=SMSType.WELCOME,
|
||
context={'name': self.customer.name_family}, dedup_key='welcome:1',
|
||
)
|
||
self.assertEqual(SMSQueue.objects.count(), 2)
|
||
|
||
|
||
class ScannerTests(TestCase):
|
||
def setUp(self):
|
||
activate_sms()
|
||
self.customer = make_customer()
|
||
PhoneNumber.objects.create(
|
||
customer=self.customer, phone_number='09120000001',
|
||
)
|
||
|
||
def test_cheque_funding_fires_before_due_and_is_idempotent(self):
|
||
"""یادآوری تأمین موجودی، پیش از سررسید و فقط یک‌بار."""
|
||
today = timezone.localdate()
|
||
# سررسید فردا، offset پیش‌فرض = ۱ ⇒ همین امروز واجد شرط است.
|
||
Sarresid.objects.create(
|
||
customer=self.customer, price=5000, payStatus='false', type='check',
|
||
sarresidDT=timezone.now() + timedelta(days=1),
|
||
)
|
||
first = run_all(today)
|
||
second = run_all(today)
|
||
self.assertEqual(first['cheque_registration'], 1)
|
||
self.assertEqual(second['cheque_registration'], 0)
|
||
self.assertEqual(
|
||
SMSQueue.objects.filter(sms_type=SMSType.CHEQUE_REGISTRATION).count(), 1,
|
||
)
|
||
|
||
def test_cheque_funding_waits_when_due_is_far_ahead(self):
|
||
"""سررسیدِ دورتر از پنجرهٔ یادآوری هیچ پیامکی نمی‌سازد."""
|
||
today = timezone.localdate()
|
||
Sarresid.objects.create(
|
||
customer=self.customer, price=5000, payStatus='false', type='check',
|
||
sarresidDT=timezone.now() + timedelta(days=5),
|
||
)
|
||
self.assertEqual(run_all(today)['cheque_registration'], 0)
|
||
|
||
def test_cheque_funding_stops_when_check_is_paid(self):
|
||
"""چکِ وصول‌شده یادآوری تأمین موجودی نمی‌گیرد."""
|
||
today = timezone.localdate()
|
||
Sarresid.objects.create(
|
||
customer=self.customer, price=5000, payStatus='true', type='check',
|
||
sarresidDT=timezone.now() + timedelta(days=1),
|
||
)
|
||
self.assertEqual(run_all(today)['cheque_registration'], 0)
|
||
|
||
def test_cheque_funding_offset_is_configurable(self):
|
||
"""تعداد روز پیش از سررسید از قالب (پنل) قابل تنظیم است."""
|
||
template = SMSTemplate.objects.get(sms_type=SMSType.CHEQUE_REGISTRATION)
|
||
apply_timing(template, 10)
|
||
template.save()
|
||
today = timezone.localdate()
|
||
# سررسید فردا ولی offset=۱۰ ⇒ داخل پنجره ⇒ پیامک می‌رود.
|
||
Sarresid.objects.create(
|
||
customer=self.customer, price=5000, payStatus='false', type='check',
|
||
sarresidDT=timezone.now() + timedelta(days=1),
|
||
)
|
||
self.assertEqual(run_all(today)['cheque_registration'], 1)
|
||
# با offset کوچک‌تر (۱)، سررسید ۱۲ روز آینده بیرون پنجره است ⇒ ۰.
|
||
Sarresid.objects.all().delete()
|
||
SMSDedup.objects.all().delete()
|
||
SMSQueue.objects.all().delete()
|
||
apply_timing(template, 1)
|
||
template.save()
|
||
Sarresid.objects.create(
|
||
customer=self.customer, price=5000, payStatus='false', type='check',
|
||
sarresidDT=timezone.now() + timedelta(days=12),
|
||
)
|
||
self.assertEqual(run_all(today)['cheque_registration'], 0)
|
||
# همان چک با offset=۱۲ داخل پنجره می‌شود ⇒ ۱.
|
||
apply_timing(template, 12)
|
||
template.save()
|
||
self.assertEqual(run_all(today)['cheque_registration'], 1)
|
||
|
||
def test_cheque_funding_days_is_days_remaining(self):
|
||
"""«{days}» تعداد روز باقی‌مانده تا سررسید است (مثبت = پیش از سررسید)."""
|
||
today = timezone.localdate()
|
||
Sarresid.objects.create(
|
||
customer=self.customer, price=5000, payStatus='false', type='check',
|
||
sarresidDT=timezone.now() + timedelta(days=1),
|
||
)
|
||
# قالبِ دیتابیس ممکن است {days} نداشته باشد؛ متن را برای این تست می‌گذاریم
|
||
# تا معنای روزِ باقی‌مانده سنجیده شود (متنِ خودِ رکورد DB دست‌نخورده می‌ماند).
|
||
SMSTemplate.objects.filter(
|
||
sms_type=SMSType.CHEQUE_REGISTRATION,
|
||
).update(text='{days}')
|
||
self.assertEqual(run_all(today)['cheque_registration'], 1)
|
||
item = SMSQueue.objects.get(sms_type=SMSType.CHEQUE_REGISTRATION)
|
||
# سررسید فردا ⇒ روز باقی‌مانده = ۱
|
||
self.assertEqual(item.message, "1")
|
||
|
||
def test_cheque_funding_exposes_sayyad_id(self):
|
||
"""«{sayyad_id}» شناسهٔ صیاد همان چک را نشان می‌دهد."""
|
||
today = timezone.localdate()
|
||
Sarresid.objects.create(
|
||
customer=self.customer, price=5000, payStatus='false', type='check',
|
||
sarresidDT=timezone.now() + timedelta(days=1),
|
||
check_row_number='4321', check_sayyad_id='1234567890123456',
|
||
)
|
||
# متن با هر دو متغیر چک، برای سنجش مقدار واقعی (DB دست‌نخورده می‌ماند).
|
||
SMSTemplate.objects.filter(
|
||
sms_type=SMSType.CHEQUE_REGISTRATION,
|
||
).update(text='{check_row}|{sayyad_id}')
|
||
self.assertEqual(run_all(today)['cheque_registration'], 1)
|
||
item = SMSQueue.objects.get(sms_type=SMSType.CHEQUE_REGISTRATION)
|
||
self.assertEqual(item.message, "4321|1234567890123456")
|
||
|
||
def test_cheque_funding_sayyad_id_empty_when_not_set(self):
|
||
"""اگر شناسهٔ صیاد ثبت نشده باشد، متغیر خالی رندر می‌شود (نه خطا)."""
|
||
today = timezone.localdate()
|
||
Sarresid.objects.create(
|
||
customer=self.customer, price=5000, payStatus='false', type='check',
|
||
sarresidDT=timezone.now() + timedelta(days=1),
|
||
check_row_number='9999',
|
||
)
|
||
SMSTemplate.objects.filter(
|
||
sms_type=SMSType.CHEQUE_REGISTRATION,
|
||
).update(text='[{sayyad_id}]')
|
||
self.assertEqual(run_all(today)['cheque_registration'], 1)
|
||
item = SMSQueue.objects.get(sms_type=SMSType.CHEQUE_REGISTRATION)
|
||
self.assertEqual(item.message, "[]")
|
||
|
||
def test_balance_is_available_in_every_template(self):
|
||
"""«{balance}» در همهٔ قالب‌ها (نه فقط چک) قابل استفاده است."""
|
||
from admin_panel.models import Article, SMSType as T
|
||
|
||
Article.objects.create(customer=self.customer, debit=500000, credit=0)
|
||
Article.objects.create(customer=self.customer, debit=0, credit=125000)
|
||
# ۵۰۰٬۰۰۰ − ۱۲۵٬۰۰۰ = ۳۷۵٬۰۰۰
|
||
for sms_type in T.values:
|
||
SMSTemplate.objects.filter(sms_type=sms_type).update(
|
||
text='{balance}|{balance_sep}',
|
||
)
|
||
# رویدادی: خوش‌آمدگویی
|
||
notify_welcome(self.customer)
|
||
item = SMSQueue.objects.get(sms_type=T.WELCOME)
|
||
self.assertEqual(item.message, "375000|375,000")
|
||
|
||
def test_balance_reflects_real_articles(self):
|
||
"""مانده حساب = مجموع بدهکار − مجموع بستانکار اسناد مشتری."""
|
||
from admin_panel.models import Article
|
||
from admin_panel.sms.balance import customer_balance
|
||
|
||
self.assertEqual(customer_balance(self.customer), 0)
|
||
Article.objects.create(customer=self.customer, debit=900000, credit=0)
|
||
Article.objects.create(customer=self.customer, debit=0, credit=400000)
|
||
self.assertEqual(customer_balance(self.customer), 500000)
|
||
# مانده منفی هم باید درست برگردد (بستانکار بیشتر از بدهکار).
|
||
Article.objects.create(customer=self.customer, debit=0, credit=750000)
|
||
self.assertEqual(customer_balance(self.customer), -250000)
|
||
|
||
def test_balance_zero_when_customer_is_none(self):
|
||
"""بدون مشتری، مانده صفر است و خطایی رخ نمی‌دهد."""
|
||
from admin_panel.sms.balance import customer_balance
|
||
|
||
self.assertEqual(customer_balance(None), 0)
|
||
|
||
def test_balance_does_not_touch_due_amount(self):
|
||
"""{balance} (مانده کل حساب) با {amount} (مبلغ چک) قاطی نمی‌شود."""
|
||
from admin_panel.models import Article
|
||
|
||
today = timezone.localdate()
|
||
Article.objects.create(customer=self.customer, debit=8000000, credit=0)
|
||
Sarresid.objects.create(
|
||
customer=self.customer, price=5000, payStatus='false', type='check',
|
||
sarresidDT=timezone.now() + timedelta(days=1),
|
||
)
|
||
SMSTemplate.objects.filter(
|
||
sms_type=SMSType.CHEQUE_REGISTRATION,
|
||
).update(text='amount={amount_sep} balance={balance_sep}')
|
||
self.assertEqual(run_all(today)['cheque_registration'], 1)
|
||
item = SMSQueue.objects.get(sms_type=SMSType.CHEQUE_REGISTRATION)
|
||
# مبلغ چک = ۵٬۰۰۰ (سررسیدشده) ولی مانده حساب = ۸٬۰۰۰٬۰۰۰
|
||
self.assertEqual(item.message, "amount=5,000 balance=8,000,000")
|
||
|
||
def test_nesiye_due_fires_on_due_date_and_ignores_checks(self):
|
||
"""یادآوری سررسید فقط برای نسیه است، نه چک."""
|
||
today = timezone.localdate()
|
||
Sarresid.objects.create(
|
||
customer=self.customer, price=5000, payStatus='false', type='nesiye',
|
||
sarresidDT=timezone.now(),
|
||
)
|
||
Sarresid.objects.create(
|
||
customer=self.customer, price=7000, payStatus='false', type='check',
|
||
sarresidDT=timezone.now(),
|
||
)
|
||
result = run_all(today)
|
||
self.assertEqual(result['nesiye_due'], 1)
|
||
self.assertEqual(
|
||
SMSQueue.objects.filter(sms_type=SMSType.NESIYE_DUE).count(), 1,
|
||
)
|
||
|
||
def test_nesiye_due_is_idempotent(self):
|
||
today = timezone.localdate()
|
||
Sarresid.objects.create(
|
||
customer=self.customer, price=5000, payStatus='false', type='nesiye',
|
||
sarresidDT=timezone.now(),
|
||
)
|
||
self.assertEqual(run_all(today)['nesiye_due'], 1)
|
||
self.assertEqual(run_all(today)['nesiye_due'], 0)
|
||
|
||
def test_visit_period_comes_from_template(self):
|
||
"""دورهٔ بازدید از پنل قابل تنظیم است (پیش‌فرض ۹ ماه)."""
|
||
customer = make_customer('مشتری بازدید')
|
||
PhoneNumber.objects.create(customer=customer, phone_number='09120000005')
|
||
# نصب ۱۰ ماه پیش، بدون بازدید ⇒ عقب‌افتاده از دورهٔ ۹ ماه.
|
||
Customer.objects.filter(pk=customer.pk).update(
|
||
installDate=timezone.localdate() - timedelta(days=300),
|
||
)
|
||
customer.refresh_from_db()
|
||
config = activate_sms()
|
||
config.save()
|
||
template = SMSTemplate.objects.get(sms_type=SMSType.VISIT)
|
||
apply_timing(template, 9)
|
||
template.save()
|
||
self.assertEqual(run_all(timezone.localdate())['visit'], 1)
|
||
|
||
# با دورهٔ ۱۲ ماه، همان مشتری دیگر عقب‌افتاده نیست.
|
||
SMSQueue.objects.all().delete()
|
||
apply_timing(template, 12)
|
||
template.save()
|
||
self.assertEqual(run_all(timezone.localdate())['visit'], 0)
|
||
|
||
def test_period_start_filters_out_very_old_customers(self):
|
||
"""«تاریخ شروع دوره» مشتریان خیلی قدیمی را از صف بازدید بیرون می‌گذارد."""
|
||
template = SMSTemplate.objects.get(sms_type=SMSType.VISIT)
|
||
apply_timing(template, 9)
|
||
template.save()
|
||
old = make_customer('مشتری خیلی قدیمی')
|
||
PhoneNumber.objects.create(customer=old, phone_number='09120000021')
|
||
Customer.objects.filter(pk=old.pk).update(
|
||
installDate=timezone.localdate() - timedelta(days=1500),
|
||
)
|
||
fresh = make_customer('مشتری تازه عقب‌افتاده')
|
||
PhoneNumber.objects.create(customer=fresh, phone_number='09120000022')
|
||
Customer.objects.filter(pk=fresh.pk).update(
|
||
installDate=timezone.localdate() - timedelta(days=290),
|
||
)
|
||
# شروع دوره از ۳۰ روز پیش: پنجرهٔ مشتری قدیمی خیلی قبل‌تر باز شده ⇒ فیلتر می‌شود.
|
||
activate_sms(since=timezone.now() - timedelta(days=30))
|
||
self.assertEqual(run_all(timezone.localdate())['visit'], 1)
|
||
self.assertEqual(SMSQueue.objects.filter(customer=fresh).count(), 1)
|
||
self.assertEqual(SMSQueue.objects.filter(customer=old).count(), 0)
|
||
|
||
def test_deleted_visit_reminder_rearms_next_period_only(self):
|
||
"""حذفِ یادآوری بازدید فقط همان دوره را کنسل می‌کند؛ دورهٔ بعد دوباره ساخته می‌شود."""
|
||
customer = make_customer('مشتری بازدید دوره‌ای')
|
||
PhoneNumber.objects.create(customer=customer, phone_number='09120000007')
|
||
Customer.objects.filter(pk=customer.pk).update(
|
||
installDate=timezone.localdate() - timedelta(days=300),
|
||
)
|
||
template = SMSTemplate.objects.get(sms_type=SMSType.VISIT)
|
||
apply_timing(template, 9)
|
||
template.save()
|
||
period = timedelta(weeks=int(9 * 4.3))
|
||
days_to_next = period.days - (300 % period.days) # تا شروع دورهٔ بعد
|
||
today = timezone.localdate()
|
||
|
||
def keys():
|
||
return list(
|
||
SMSQueue.objects.filter(customer=customer).values_list('dedup_key', flat=True)
|
||
)
|
||
|
||
run_all(today)
|
||
first = keys()
|
||
self.assertEqual(len(first), 1)
|
||
|
||
# مدیر مورد را حذف می‌کند ⇒ در همان دوره دوباره ساخته نمی‌شود.
|
||
SMSQueue.objects.filter(customer=customer).delete()
|
||
run_all(today)
|
||
self.assertEqual(keys(), [])
|
||
# حتی نزدیک پایان دوره هم برنمی‌گردد.
|
||
run_all(today + timedelta(days=days_to_next - 1))
|
||
self.assertEqual(keys(), [])
|
||
|
||
# در دورهٔ بعد کلیدِ جدید ساخته می‌شود (دوباره ارسال می‌شود، نه خاموشِ دائمی).
|
||
run_all(today + timedelta(days=days_to_next))
|
||
self.assertEqual(len(keys()), 1)
|
||
self.assertNotEqual(keys(), first)
|
||
|
||
def test_disabled_template_is_skipped(self):
|
||
today = timezone.localdate()
|
||
Sarresid.objects.create(
|
||
customer=self.customer, price=5000, payStatus='false', type='check',
|
||
sarresidDT=timezone.now() + timedelta(days=1),
|
||
)
|
||
SMSTemplate.objects.filter(
|
||
sms_type=SMSType.CHEQUE_REGISTRATION,
|
||
).update(is_enabled=False)
|
||
self.assertEqual(run_all(today)['cheque_registration'], 0)
|
||
|
||
|
||
class ColdStartTests(TestCase):
|
||
"""اجرای نخست پس از نصب نباید موارد قدیمی موجود را یک‌باره ارسال کند."""
|
||
|
||
def setUp(self):
|
||
self.customer = make_customer()
|
||
PhoneNumber.objects.create(
|
||
customer=self.customer, phone_number='09120000001',
|
||
)
|
||
self.today = timezone.localdate()
|
||
|
||
def _make_cheque(self, days_from_today):
|
||
"""چکِ پرداخت‌نشده با سررسیدِ ``days_from_today`` نسبت به امروز."""
|
||
return Sarresid.objects.create(
|
||
customer=self.customer, price=5000, payStatus='false', type='check',
|
||
sarresidDT=timezone.now() + timedelta(days=days_from_today),
|
||
)
|
||
|
||
def _make_due_cheque(self):
|
||
# سررسید فردا تا یادآوری «تأمین موجودی» (offset=۱) واجد شرط شود.
|
||
return self._make_cheque(1)
|
||
|
||
def test_first_run_sends_nothing_for_existing_records(self):
|
||
# مورد قدیمی: لحظهٔ واجد‌شرط‌شدنش (سررسید − offset) پیش از «خط شروع» است.
|
||
self._make_cheque(-3)
|
||
activate_sms(since=timezone.now())
|
||
self.assertEqual(run_all(self.today)['cheque_registration'], 0)
|
||
self.assertEqual(SMSQueue.objects.count(), 0)
|
||
|
||
def test_scan_since_defaults_to_now_on_first_load(self):
|
||
SMSProviderConfig.load()
|
||
# حالت رکورد مهاجرت‌نشده: مقدار NULL در دیتابیس.
|
||
SMSProviderConfig.objects.filter(pk=1).update(scan_since=None)
|
||
self.assertIsNotNone(SMSProviderConfig.load().scan_since)
|
||
|
||
def test_new_record_after_scan_since_is_sent(self):
|
||
activate_sms(since=timezone.now() - timedelta(days=3))
|
||
self._make_due_cheque()
|
||
self.assertEqual(run_all(self.today)['cheque_registration'], 1)
|
||
|
||
def test_backlog_blocked_until_admin_allows_it(self):
|
||
for _ in range(3):
|
||
self._make_due_cheque()
|
||
activate_sms(since=timezone.now() + timedelta(days=1), allow_backlog=False)
|
||
self.assertEqual(run_all(self.today)['cheque_registration'], 0)
|
||
|
||
def test_backlog_respects_admin_cap(self):
|
||
for _ in range(5):
|
||
self._make_due_cheque()
|
||
activate_sms(since=timezone.now() + timedelta(days=1),
|
||
allow_backlog=True, backlog_max=2)
|
||
self.assertEqual(run_all(self.today)['cheque_registration'], 2)
|
||
# اجرای بعدی سهم بعدی را می‌فرستد و هرگز از سقف عبور نمی‌کند.
|
||
self.assertEqual(run_all(self.today)['cheque_registration'], 2)
|
||
self.assertEqual(SMSQueue.objects.count(), 4)
|
||
|
||
def test_max_per_run_caps_each_scanner(self):
|
||
for _ in range(5):
|
||
self._make_due_cheque()
|
||
activate_sms(since=timezone.now() - timedelta(days=3), max_per_run=2)
|
||
self.assertEqual(run_all(self.today)['cheque_registration'], 2)
|
||
|
||
|
||
class _FailingProvider(SMSProvider):
|
||
def send_otp(self, mobile, code):
|
||
return SMSResult(False, error='otp unsupported')
|
||
|
||
def send_message(self, mobile, message):
|
||
return SMSResult(False, error='boom')
|
||
|
||
|
||
class WorkerTests(TestCase):
|
||
def setUp(self):
|
||
self.customer = make_customer()
|
||
self.queue_item = SMSQueue.objects.create(
|
||
customer=self.customer, sms_type=SMSType.WELCOME,
|
||
phone_number='+989120000000', message='سلام', status=SMSQueue.Status.APPROVED,
|
||
)
|
||
|
||
def _config(self, provider, *, interval=0, batch=200):
|
||
config = activate_sms(interval=interval, max_per_run=batch)
|
||
config.provider = provider
|
||
config.is_enabled = True
|
||
config.save()
|
||
return config
|
||
|
||
def test_console_provider_marks_sent(self):
|
||
self._config('console')
|
||
call_command('sms_worker', '--once')
|
||
self.queue_item.refresh_from_db()
|
||
self.assertEqual(self.queue_item.status, SMSQueue.Status.SENT)
|
||
self.assertTrue(SMSLog.objects.filter(queue=self.queue_item, success=True).exists())
|
||
|
||
def test_failing_provider_backs_off_then_fails(self):
|
||
self._config('console')
|
||
from unittest import mock
|
||
with mock.patch(
|
||
'admin_panel.management.commands.sms_worker.get_sms_provider',
|
||
return_value=_FailingProvider(),
|
||
):
|
||
for _ in range(5):
|
||
# شبیه‌سازی گذر زمان: پاک‌کردن backoff تا تلاش بعدی مجاز شود.
|
||
SMSQueue.objects.filter(pk=self.queue_item.pk).update(next_retry_at=None)
|
||
call_command('sms_worker', '--once')
|
||
|
||
self.queue_item.refresh_from_db()
|
||
self.assertEqual(self.queue_item.status, SMSQueue.Status.FAILED)
|
||
self.assertEqual(self.queue_item.attempts, 5)
|
||
self.assertEqual(SMSLog.objects.filter(queue=self.queue_item, success=False).count(), 5)
|
||
|
||
def test_disabled_config_leaves_queue_untouched(self):
|
||
config = self._config('console')
|
||
config.is_enabled = False
|
||
config.save()
|
||
call_command('sms_worker', '--once')
|
||
self.queue_item.refresh_from_db()
|
||
self.assertEqual(self.queue_item.status, SMSQueue.Status.APPROVED)
|
||
|
||
def test_pending_item_is_not_sent_until_approved(self):
|
||
"""بدون تأیید مدیر، worker نباید پیامکِ «در انتظار تأیید» را بفرستد."""
|
||
self._config('console')
|
||
self.queue_item.status = SMSQueue.Status.PENDING
|
||
self.queue_item.save(update_fields=['status'])
|
||
call_command('sms_worker', '--once')
|
||
self.queue_item.refresh_from_db()
|
||
self.assertEqual(self.queue_item.status, SMSQueue.Status.PENDING)
|
||
self.assertFalse(SMSLog.objects.filter(queue=self.queue_item).exists())
|
||
|
||
def test_send_interval_is_respected_between_messages(self):
|
||
for i in range(2):
|
||
SMSQueue.objects.create(
|
||
customer=self.customer, sms_type=SMSType.WELCOME,
|
||
phone_number=f'+98912000001{i}', message='سلام',
|
||
status=SMSQueue.Status.APPROVED,
|
||
)
|
||
self._config('console', interval=1)
|
||
from unittest import mock
|
||
with mock.patch('time.sleep') as sleeper:
|
||
call_command('sms_worker', '--once')
|
||
# سه پیام ⇒ دقیقاً دو فاصله (نه پس از پیام آخر).
|
||
total = sum(call.args[0] for call in sleeper.call_args_list)
|
||
self.assertAlmostEqual(total, 2.0, places=2)
|
||
self.assertEqual(
|
||
SMSQueue.objects.filter(status=SMSQueue.Status.SENT).count(), 3,
|
||
)
|
||
|
||
|
||
class ConsoleEncodingTests(TestCase):
|
||
"""رگرسیون خطای 'charmap': کنسول محدود ویندوز نباید ارسال را «ناموفق» کند."""
|
||
|
||
def setUp(self):
|
||
self.customer = make_customer()
|
||
self.queue_item = SMSQueue.objects.create(
|
||
customer=self.customer, sms_type=SMSType.WELCOME,
|
||
phone_number='+989120000000', message='سلام مشتری عزیز',
|
||
status=SMSQueue.Status.APPROVED,
|
||
)
|
||
|
||
@staticmethod
|
||
def _cp1252_stdout():
|
||
"""استریم stdout با کدپیج ویندوز؛ نوشتن فارسی روی آن خطا می‌دهد."""
|
||
import io
|
||
return io.TextIOWrapper(io.BytesIO(), encoding='cp1252')
|
||
|
||
def test_console_provider_never_raises_on_unencodable_text(self):
|
||
import sys
|
||
from integrations.sms.console import ConsoleSMSProvider
|
||
|
||
original = sys.stdout
|
||
sys.stdout = self._cp1252_stdout()
|
||
try:
|
||
result = ConsoleSMSProvider(SMSProviderConfig.load()).send_message(
|
||
'+989120000000', 'سلام مشتری عزیز',
|
||
)
|
||
finally:
|
||
sys.stdout = original
|
||
|
||
self.assertTrue(result.success)
|
||
|
||
def test_worker_marks_sent_when_console_cannot_encode(self):
|
||
import sys
|
||
config = activate_sms(interval=0)
|
||
config.provider = 'console'
|
||
config.is_enabled = True
|
||
config.save()
|
||
|
||
original = sys.stdout
|
||
sys.stdout = self._cp1252_stdout()
|
||
try:
|
||
call_command('sms_worker', '--once')
|
||
finally:
|
||
sys.stdout = original
|
||
|
||
self.queue_item.refresh_from_db()
|
||
self.assertEqual(self.queue_item.status, SMSQueue.Status.SENT)
|
||
self.assertEqual(self.queue_item.last_error, '')
|
||
|
||
def test_worker_survives_a_failing_drain_batch(self):
|
||
"""یک دور خراب نباید worker را از پا دربیاورد (وگرنه صف راکد می‌ماند)."""
|
||
from unittest import mock
|
||
config = activate_sms(interval=0)
|
||
config.provider = 'console'
|
||
config.is_enabled = True
|
||
config.save()
|
||
|
||
with mock.patch(
|
||
'admin_panel.management.commands.sms_worker.Command._drain_batch',
|
||
side_effect=RuntimeError('db hiccup'),
|
||
):
|
||
# نباید استثنا به بیرون درز کند.
|
||
call_command('sms_worker', '--once')
|
||
|
||
self.queue_item.refresh_from_db()
|
||
self.assertEqual(self.queue_item.status, SMSQueue.Status.APPROVED)
|
||
|
||
def test_std_streams_helper_is_safe_on_non_reconfigurable_streams(self):
|
||
import sys
|
||
from admin_panel.apps import make_std_streams_unicode_safe
|
||
|
||
original = sys.stdout
|
||
sys.stdout = object() # جریانی که reconfigure ندارد
|
||
try:
|
||
make_std_streams_unicode_safe() # نباید استثنا بدهد
|
||
finally:
|
||
sys.stdout = original
|
||
|
||
|
||
class HttpProviderRobustnessTests(TestCase):
|
||
"""provider عمومی HTTP باید پاسخ غیر-JSON را به خطای خوانا تبدیل کند، نه استثنا."""
|
||
|
||
def _provider(self, status=200, body=b'<html>oops</html>', content_type='text/html'):
|
||
from unittest import mock
|
||
from integrations.sms.http import HTTPSMSProvider
|
||
|
||
config = SMSProviderConfig.load()
|
||
config.api_url = 'https://sms.example.test/send'
|
||
provider = HTTPSMSProvider(config)
|
||
|
||
response = mock.Mock()
|
||
response.status_code = status
|
||
response.content = body
|
||
response.headers = {'Content-Type': content_type}
|
||
response.json.side_effect = ValueError('Expecting value: line 1 column 1')
|
||
response.raise_for_status.return_value = None
|
||
return provider, mock.patch('integrations.sms.http.requests.post', return_value=response)
|
||
|
||
def test_non_json_response_becomes_a_clean_error(self):
|
||
provider, patched = self._provider()
|
||
with patched:
|
||
result = provider.send_message('+989120000000', 'سلام')
|
||
self.assertFalse(result.success)
|
||
self.assertIn('JSON', result.error)
|
||
self.assertEqual(result.provider_code, 'BAD_RESPONSE')
|
||
|
||
def test_invalid_recipient_is_rejected_without_any_request(self):
|
||
"""شمارهٔ نامعتبر نباید به سرویس‌دهنده برود (هزینه/خطای گنگ)."""
|
||
provider, patched = self._provider(body=b'{}', content_type='application/json')
|
||
with patched as post:
|
||
result = provider.send_message('000', 'سلام')
|
||
self.assertFalse(result.success)
|
||
self.assertEqual(result.provider_code, 'INVALID_NUMBER')
|
||
post.assert_not_called()
|
||
|
||
|
||
class ZeroZeroNinetyEightLinkTests(TestCase):
|
||
"""سرویس لینکی 0098SMS: GET با پارامترهای vendor، شمارهٔ محلی، و تفکیک پاسخ."""
|
||
|
||
def _provider(self, **overrides):
|
||
from integrations.sms.sms0098 import ZeroZeroNinetyEightSMSProvider
|
||
|
||
config = SMSProviderConfig.load()
|
||
config.username = 'vsms7937'
|
||
config.password = '335577775533'
|
||
config.sender = '30005367'
|
||
config.api_url = 'http://www.0098sms.com/sendsmslink.aspx'
|
||
config.timeout_seconds = 10
|
||
for field, value in overrides.items():
|
||
setattr(config, field, value)
|
||
return ZeroZeroNinetyEightSMSProvider(config)
|
||
|
||
def _mock_get(self, text='0', status=200):
|
||
response = mock.Mock()
|
||
response.status_code = status
|
||
response.text = text
|
||
response.raise_for_status.return_value = None
|
||
return mock.patch('integrations.sms.sms0098.requests.get', return_value=response)
|
||
|
||
def test_request_matches_vendor_contract_and_uses_local_number(self):
|
||
"""پارامترها باید همان نام‌های vendor و شماره با صفر ابتدایی باشند."""
|
||
provider = self._provider()
|
||
with self._mock_get('0') as get:
|
||
result = provider.send_message('+989384999655', 'سلام مشتری')
|
||
self.assertTrue(result.success)
|
||
self.assertEqual(result.provider_code, '0')
|
||
args, kwargs = get.call_args
|
||
self.assertEqual(args[0], 'http://www.0098sms.com/sendsmslink.aspx')
|
||
self.assertEqual(kwargs['params'], {
|
||
'USERNAME': 'vsms7937',
|
||
'PASSWORD': '335577775533',
|
||
'FROM': '30005367',
|
||
'TO': '09384999655',
|
||
'TEXT': 'سلام مشتری',
|
||
})
|
||
self.assertEqual(kwargs['timeout'], 10)
|
||
|
||
def test_credentials_come_from_the_database_config(self):
|
||
"""هیچ مقدار ثابتی در کد نیست؛ همه از SMSProviderConfig خوانده می‌شود."""
|
||
provider = self._provider(username='other', password='pw2', sender='30009999')
|
||
with self._mock_get('0') as get:
|
||
provider.send_message('09202202129', 'متن')
|
||
self.assertEqual(get.call_args.kwargs['params']['USERNAME'], 'other')
|
||
self.assertEqual(get.call_args.kwargs['params']['PASSWORD'], 'pw2')
|
||
self.assertEqual(get.call_args.kwargs['params']['FROM'], '30009999')
|
||
|
||
def test_explicit_error_body_is_a_clean_failure(self):
|
||
provider = self._provider()
|
||
with self._mock_get('error'):
|
||
result = provider.send_message('09384999655', 'سلام')
|
||
self.assertFalse(result.success)
|
||
self.assertEqual(result.provider_code, 'REJECTED')
|
||
|
||
def test_timeout_is_unconfirmed_not_silent_success(self):
|
||
"""مهلت تمام‌شده «موفق» ثبت نمی‌شود؛ اما دور انداخته هم نمی‌شود (retry)."""
|
||
import requests
|
||
provider = self._provider()
|
||
with mock.patch('integrations.sms.sms0098.requests.get',
|
||
side_effect=requests.Timeout('timed out')):
|
||
result = provider.send_message('09384999655', 'سلام')
|
||
self.assertFalse(result.success)
|
||
self.assertEqual(result.provider_code, 'UNCONFIRMED')
|
||
|
||
def test_invalid_number_never_reaches_the_provider(self):
|
||
provider = self._provider()
|
||
with self._mock_get('0') as get:
|
||
result = provider.send_message('000', 'سلام')
|
||
self.assertFalse(result.success)
|
||
self.assertEqual(result.provider_code, 'INVALID_NUMBER')
|
||
get.assert_not_called()
|
||
|
||
def test_missing_credentials_is_a_config_error(self):
|
||
provider = self._provider(username='')
|
||
with self._mock_get('0') as get:
|
||
result = provider.send_message('09384999655', 'سلام')
|
||
self.assertFalse(result.success)
|
||
self.assertEqual(result.provider_code, 'CONFIG')
|
||
get.assert_not_called()
|
||
|
||
|
||
class ZeroZeroNinetyEightWebServiceTests(TestCase):
|
||
"""وب‌سرویس SOAP 0098SMS: ارسال با smsid + استعلام دلیوری (zeep ماک‌شده)."""
|
||
|
||
def _provider(self, **overrides):
|
||
from integrations.sms.sms0098ws import ZeroZeroNinetyEightWebServiceProvider
|
||
|
||
config = SMSProviderConfig.load()
|
||
config.username = 'vsms7937'
|
||
config.password = '335577775533'
|
||
config.panel_no = '30005367'
|
||
config.webservice_url = 'https://webservice.0098sms.com/service.asmx'
|
||
config.timeout_seconds = 10
|
||
for field, value in overrides.items():
|
||
setattr(config, field, value)
|
||
return ZeroZeroNinetyEightWebServiceProvider(config)
|
||
|
||
def _patched_client(self):
|
||
"""zeep.Client را با یک client جعلی جایگزین می‌کند و service آن را برمی‌گرداند."""
|
||
fake = mock.Mock()
|
||
return fake, mock.patch(
|
||
'integrations.sms.sms0098ws.zeep.Client', return_value=fake,
|
||
)
|
||
|
||
def test_send_returns_smsid_and_uses_vendor_params(self):
|
||
provider = self._provider()
|
||
fake, patched = self._patched_client()
|
||
fake.service.SendSMSWithID.return_value = '123456789'
|
||
with patched:
|
||
result = provider.send_message('+989384999655', 'سلام مشتری')
|
||
|
||
self.assertTrue(result.success)
|
||
self.assertEqual(result.provider_id, '123456789')
|
||
# نام پارامترها باید دقیقاً مطابق WSDL (تماماً کوچک) باشد.
|
||
kwargs = fake.service.SendSMSWithID.call_args.kwargs
|
||
self.assertEqual(kwargs['username'], 'vsms7937')
|
||
self.assertEqual(kwargs['password'], '335577775533')
|
||
self.assertEqual(kwargs['text'], 'سلام مشتری')
|
||
self.assertEqual(kwargs['mobileno'], '09384999655') # شمارهٔ محلی
|
||
self.assertEqual(kwargs['pnlno'], '30005367')
|
||
|
||
def test_send_uses_exact_wsdl_kwarg_names(self):
|
||
"""رگرسیون: راهنمای vendor نام‌ها را با حرف بزرگ می‌نویسد، ولی WSDL
|
||
سرویس تماماً کوچک است؛ نام بزرگ ⇒ zeep «unexpected keyword argument»."""
|
||
provider = self._provider()
|
||
fake, patched = self._patched_client()
|
||
fake.service.SendSMSWithID.return_value = '123456789'
|
||
with patched:
|
||
provider.send_message('09384999655', 'متن')
|
||
kwargs = fake.service.SendSMSWithID.call_args.kwargs
|
||
self.assertEqual(
|
||
set(kwargs), {'username', 'password', 'text', 'mobileno', 'pnlno'},
|
||
)
|
||
for wrong in ('Username', 'Password', 'MobileNo', 'PnlNo'):
|
||
self.assertNotIn(wrong, kwargs)
|
||
|
||
def test_send_zero_is_accepted_without_id(self):
|
||
provider = self._provider()
|
||
fake, patched = self._patched_client()
|
||
fake.service.SendSMSWithID.return_value = 0
|
||
with patched:
|
||
result = provider.send_message('09384999655', 'متن')
|
||
self.assertTrue(result.success)
|
||
self.assertEqual(result.provider_id, '')
|
||
self.assertEqual(result.provider_code, '0')
|
||
|
||
def test_error_code_is_translated_to_persian(self):
|
||
provider = self._provider()
|
||
fake, patched = self._patched_client()
|
||
fake.service.SendSMSWithID.return_value = '66'
|
||
with patched:
|
||
result = provider.send_message('09384999655', 'متن')
|
||
self.assertFalse(result.success)
|
||
self.assertEqual(result.provider_code, '66')
|
||
self.assertIn('شارژ', result.error)
|
||
|
||
def test_invalid_number_never_reaches_the_service(self):
|
||
provider = self._provider()
|
||
fake, patched = self._patched_client()
|
||
with patched:
|
||
result = provider.send_message('000', 'متن')
|
||
self.assertFalse(result.success)
|
||
self.assertEqual(result.provider_code, 'INVALID_NUMBER')
|
||
fake.service.SendSMSWithID.assert_not_called()
|
||
|
||
def test_missing_panel_uses_sender_as_fallback(self):
|
||
provider = self._provider(panel_no='', sender='30009999')
|
||
fake, patched = self._patched_client()
|
||
fake.service.SendSMSWithID.return_value = '123456789'
|
||
with patched:
|
||
provider.send_message('09384999655', 'متن')
|
||
self.assertEqual(fake.service.SendSMSWithID.call_args.kwargs['pnlno'], '30009999')
|
||
|
||
def test_missing_credentials_is_a_config_error(self):
|
||
provider = self._provider(username='')
|
||
fake, patched = self._patched_client()
|
||
with patched:
|
||
result = provider.send_message('09384999655', 'متن')
|
||
self.assertFalse(result.success)
|
||
self.assertEqual(result.provider_code, 'CONFIG')
|
||
fake.service.SendSMSWithID.assert_not_called()
|
||
|
||
def test_delivery_state_is_classified(self):
|
||
provider = self._provider()
|
||
fake, patched = self._patched_client()
|
||
for raw, status, is_final in (
|
||
('5', 'delivered', True),
|
||
('6', 'not_delivered', True),
|
||
('7', 'expired', True),
|
||
('8', 'unknown', False),
|
||
):
|
||
fake.service.smsdeliveryState.return_value = raw
|
||
with patched:
|
||
result = provider.get_delivery_status('123456789')
|
||
self.assertEqual(result.status, status)
|
||
self.assertEqual(result.is_final, is_final)
|
||
self.assertEqual(
|
||
fake.service.smsdeliveryState.call_args.kwargs['smsid'], '123456789',
|
||
)
|
||
self.assertEqual(
|
||
fake.service.smsdeliveryState.call_args.kwargs['username'], 'vsms7937',
|
||
)
|
||
|
||
def test_delivery_query_error_returns_none(self):
|
||
provider = self._provider()
|
||
fake, patched = self._patched_client()
|
||
fake.service.smsdeliveryState.side_effect = RuntimeError('network')
|
||
with patched:
|
||
self.assertIsNone(provider.get_delivery_status('123456789'))
|
||
|
||
def test_provider_is_registered_in_factory(self):
|
||
from integrations.sms.factory import PROVIDER_CLASSES
|
||
self.assertIn('sms0098ws', PROVIDER_CLASSES)
|
||
|
||
|
||
class SMSDeliveryCommandTests(TestCase):
|
||
"""دستور sms_delivery: استعلام smsid و به‌روزرسانی وضعیت تحویل."""
|
||
|
||
def setUp(self):
|
||
from admin_panel.models import SMSQueue
|
||
self.SMSQueue = SMSQueue
|
||
config = SMSProviderConfig.load()
|
||
config.provider = 'sms0098ws'
|
||
config.username = 'u'
|
||
config.password = 'p'
|
||
config.panel_no = '3000'
|
||
config.webservice_url = 'https://webservice.0098sms.com/service.asmx'
|
||
config.save()
|
||
self.old = timezone.now() - timedelta(minutes=10)
|
||
|
||
def _row(self, provider_id='123456789', delivery_status='pending'):
|
||
return self.SMSQueue.objects.create(
|
||
phone_number='09120000000', message='متن', status=self.SMSQueue.Status.SENT,
|
||
provider_id=provider_id, delivery_status=delivery_status, sent_at=self.old,
|
||
)
|
||
|
||
def _run(self, result):
|
||
fake = mock.Mock()
|
||
fake.get_delivery_status.return_value = result
|
||
with mock.patch(
|
||
'admin_panel.management.commands.sms_delivery.get_sms_provider',
|
||
return_value=fake,
|
||
):
|
||
call_command('sms_delivery', '--once')
|
||
return fake
|
||
|
||
def test_delivered_row_is_updated_and_becomes_final(self):
|
||
from integrations.sms.base import SMSDeliveryResult
|
||
row = self._row()
|
||
self._run(SMSDeliveryResult('delivered', '5', 'رسیده به گوشی.', True))
|
||
row.refresh_from_db()
|
||
self.assertEqual(row.delivery_status, 'delivered')
|
||
self.assertIsNotNone(row.delivery_checked_at)
|
||
|
||
def test_final_row_is_not_queried_again(self):
|
||
from integrations.sms.base import SMSDeliveryResult
|
||
row = self._row(delivery_status='delivered')
|
||
row.delivery_checked_at = self.old
|
||
row.save()
|
||
fake = self._run(SMSDeliveryResult('delivered', '5', 'رسیده به گوشی.', True))
|
||
fake.get_delivery_status.assert_not_called()
|
||
|
||
def test_row_without_provider_id_is_skipped(self):
|
||
from integrations.sms.base import SMSDeliveryResult
|
||
self._row(provider_id='')
|
||
fake = self._run(SMSDeliveryResult('delivered', '5', 'x', True))
|
||
fake.get_delivery_status.assert_not_called()
|
||
|
||
def test_none_result_leaves_row_untouched_for_next_round(self):
|
||
row = self._row()
|
||
self._run(None)
|
||
row.refresh_from_db()
|
||
self.assertEqual(row.delivery_status, 'pending')
|
||
self.assertIsNone(row.delivery_checked_at)
|
||
|
||
def test_non_webservice_provider_is_a_noop(self):
|
||
from integrations.sms.base import SMSDeliveryResult
|
||
SMSProviderConfig.objects.filter(pk=1).update(provider='sms0098')
|
||
row = self._row()
|
||
fake = self._run(SMSDeliveryResult('delivered', '5', 'x', True))
|
||
fake.get_delivery_status.assert_not_called()
|
||
row.refresh_from_db()
|
||
self.assertEqual(row.delivery_status, 'pending')
|
||
|
||
|
||
class PanelTestCase(TestCase):
|
||
"""پایهٔ تست‌های پنل پیامک: یک کاربر سوپریوزر وارد‌شده می‌سازد."""
|
||
|
||
username = 'smsPanel'
|
||
|
||
def setUp(self):
|
||
self.user = login_superuser(self.client, self.username)
|
||
|
||
|
||
class TimingsTests(PanelTestCase):
|
||
"""هر نوع پیامک زمان‌بندی خودش را دارد؛ فقط یک فیلد زمانی معنا دارد."""
|
||
|
||
username = 'smsTiming'
|
||
|
||
def test_every_sms_type_has_timing_metadata(self):
|
||
from admin_panel.sms.timing import TIMING
|
||
for code, _label in SMSType.choices:
|
||
self.assertIn(code, TIMING, msg=code)
|
||
|
||
def test_apply_timing_pins_only_the_relevant_field(self):
|
||
template = SMSTemplate.objects.get(sms_type=SMSType.CHEQUE_REGISTRATION)
|
||
apply_timing(template, 5)
|
||
self.assertEqual(template.offset_days, 5)
|
||
self.assertIsNone(template.repeat_interval_days)
|
||
self.assertIsNone(template.repeat_interval_months)
|
||
|
||
visit = SMSTemplate.objects.get(sms_type=SMSType.VISIT)
|
||
apply_timing(visit, 6)
|
||
self.assertEqual(visit.repeat_interval_months, 6)
|
||
self.assertEqual(visit.offset_days, 0)
|
||
self.assertIsNone(visit.repeat_interval_days)
|
||
|
||
def test_types_without_timing_have_no_fields(self):
|
||
for sms_type in (SMSType.WELCOME, SMSType.SURVEY,
|
||
SMSType.VISIT_CREATED, SMSType.PAYMENT_THANKS):
|
||
self.assertIsNone(timing_for(sms_type)['field'])
|
||
|
||
def test_visit_default_is_nine_months(self):
|
||
template = SMSTemplate.objects.get(sms_type=SMSType.VISIT)
|
||
SMSTemplate.objects.filter(pk=template.pk).update(repeat_interval_months=None)
|
||
template.refresh_from_db()
|
||
self.assertEqual(current_timing_value(template), 9)
|
||
|
||
def test_edit_view_saves_the_relevant_timing_field(self):
|
||
template = SMSTemplate.objects.get(sms_type=SMSType.CHEQUE_REGISTRATION)
|
||
self.client.post(
|
||
reverse('admin_panel:sms_template_edit', kwargs={'t_id': template.id}),
|
||
{'title': template.title, 'text': template.text,
|
||
'is_enabled': 'on', 'timing_value': '4'},
|
||
HTTP_REFERER='/panel/',
|
||
)
|
||
template.refresh_from_db()
|
||
self.assertEqual(template.offset_days, 4)
|
||
self.assertIsNone(template.repeat_interval_days)
|
||
|
||
def test_edit_view_rejects_non_numeric_timing(self):
|
||
template = SMSTemplate.objects.get(sms_type=SMSType.CHEQUE_REGISTRATION)
|
||
before = template.offset_days
|
||
self.client.post(
|
||
reverse('admin_panel:sms_template_edit', kwargs={'t_id': template.id}),
|
||
{'title': template.title, 'text': template.text,
|
||
'is_enabled': 'on', 'timing_value': 'چند روز'},
|
||
HTTP_REFERER='/panel/',
|
||
)
|
||
template.refresh_from_db()
|
||
self.assertEqual(template.offset_days, before)
|
||
|
||
|
||
class TemplatesPageUITests(PanelTestCase):
|
||
"""قرارداد صفحهٔ قالب‌ها: کارت هر نوع، کادر متن، پیش‌نمایش و بازگشت به همان کارت."""
|
||
|
||
username = 'smsUI'
|
||
|
||
def _get(self):
|
||
return self.client.get(reverse('admin_panel:sms_templates'))
|
||
|
||
def test_page_renders_a_card_per_sms_type(self):
|
||
html = self._get().content.decode('utf-8')
|
||
for code, _label in SMSType.choices:
|
||
self.assertIn(f'id="type-{code}"', html, msg=code)
|
||
|
||
def test_editor_uses_a_dedicated_text_class_per_type(self):
|
||
html = self._get().content.decode('utf-8')
|
||
self.assertIn('smsx-textarea', html)
|
||
for code, _label in SMSType.choices:
|
||
self.assertIn(f'id="text-{code}"', html, msg=code)
|
||
|
||
def test_editor_config_is_valid_json(self):
|
||
import json as _json
|
||
import re
|
||
html = self._get().content.decode('utf-8')
|
||
match = re.search(
|
||
r'<script id="smsx-config" type="application/json">(.*?)</script>', html, re.S,
|
||
)
|
||
self.assertIsNotNone(match)
|
||
config = _json.loads(match.group(1))
|
||
self.assertIn('{name}', config['tokens'])
|
||
self.assertEqual(config['sample']['name'], 'مشتری نمونه')
|
||
|
||
def test_preview_substitutes_sample_values(self):
|
||
html = self._get().content.decode('utf-8')
|
||
# متن خوش‌آمدگویی شامل {name} است و پیش‌نمایش باید نام نمونه را نشان دهد.
|
||
self.assertIn('مشتری نمونه', html)
|
||
|
||
def test_only_one_card_is_open_initially(self):
|
||
import re
|
||
html = self._get().content.decode('utf-8')
|
||
opened = re.findall(r'<details class="smsx-card" id="type-\w+" open', html)
|
||
self.assertEqual(len(opened), 1)
|
||
|
||
def test_layout_css_is_actually_rendered_in_the_page(self):
|
||
"""رگرسیون: اگر بلوک CSS صفحه در HTML نیاید، کارت‌ها و کادرها به‌هم می‌ریزند."""
|
||
html = self._get().content.decode('utf-8')
|
||
styles = re.findall(r'<style[^>]*>(.*?)</style>', html, re.S)
|
||
css = '\n'.join(styles)
|
||
# box-sizing سراسری داخل صفحه، گرید فرم، و قاعدهٔ خود textarea
|
||
self.assertIn('.smsx, .smsx *', css)
|
||
self.assertIn('box-sizing: border-box', css)
|
||
self.assertIn('.smsx-grid', css)
|
||
self.assertIn('textarea.smsx-textarea', css)
|
||
|
||
def test_every_field_cell_starts_with_its_own_label(self):
|
||
"""رگرسیون: هر سلول فرم باید برچسب مستقل داشته باشد تا برچسب/کنترل روی هم نیفتد."""
|
||
html = self._get().content.decode('utf-8')
|
||
cells = re.findall(r'<div class="smsx-cell[^"]*">', html)
|
||
labelled = re.findall(
|
||
r'<div class="smsx-cell[^"]*">\s*<(?:label|span) class="smsx-label"', html,
|
||
)
|
||
self.assertTrue(cells)
|
||
self.assertEqual(len(cells), len(labelled))
|
||
|
||
def test_save_redirects_back_to_the_same_card(self):
|
||
template = SMSTemplate.objects.get(sms_type=SMSType.VISIT)
|
||
response = self.client.post(
|
||
reverse('admin_panel:sms_template_edit', kwargs={'t_id': template.id}),
|
||
{'title': template.title, 'text': template.text,
|
||
'is_enabled': 'on', 'timing_value': '9'},
|
||
HTTP_REFERER='/panel/',
|
||
)
|
||
self.assertEqual(response.status_code, 302)
|
||
self.assertTrue(response['Location'].endswith('#type-visit'))
|
||
|
||
def test_invalid_timing_also_returns_to_the_same_card(self):
|
||
template = SMSTemplate.objects.get(sms_type=SMSType.CHEQUE_REGISTRATION)
|
||
response = self.client.post(
|
||
reverse('admin_panel:sms_template_edit', kwargs={'t_id': template.id}),
|
||
{'title': template.title, 'text': template.text,
|
||
'is_enabled': 'on', 'timing_value': 'abc'},
|
||
HTTP_REFERER='/panel/',
|
||
)
|
||
self.assertEqual(response.status_code, 302)
|
||
self.assertTrue(response['Location'].endswith('#type-cheque_registration'))
|
||
|
||
|
||
class AutoSendTests(PanelTestCase):
|
||
"""«ارسال خودکار» هر نوع پیامک: پیش‌فرض خاموش (نیازمند تأیید مدیر)."""
|
||
|
||
username = 'smsAuto'
|
||
|
||
def _overdue_visit_customer(self):
|
||
customer = make_customer('مشتری بازدید خودکار')
|
||
PhoneNumber.objects.create(customer=customer, phone_number='09120000031')
|
||
Customer.objects.filter(pk=customer.pk).update(
|
||
installDate=timezone.localdate() - timedelta(days=300),
|
||
)
|
||
template = SMSTemplate.objects.get(sms_type=SMSType.VISIT)
|
||
apply_timing(template, 9)
|
||
template.save()
|
||
return customer, template
|
||
|
||
def test_default_is_pending_so_manager_must_approve(self):
|
||
customer, template = self._overdue_visit_customer()
|
||
self.assertFalse(template.auto_send) # پیش‌فرض خاموش
|
||
activate_sms()
|
||
self.assertEqual(run_all(timezone.localdate())['visit'], 1)
|
||
item = SMSQueue.objects.get(customer=customer, sms_type=SMSType.VISIT)
|
||
self.assertEqual(item.status, SMSQueue.Status.PENDING)
|
||
|
||
def test_auto_send_enqueues_approved_without_manager(self):
|
||
customer, template = self._overdue_visit_customer()
|
||
template.auto_send = True
|
||
template.save(update_fields=['auto_send'])
|
||
activate_sms()
|
||
self.assertEqual(run_all(timezone.localdate())['visit'], 1)
|
||
item = SMSQueue.objects.get(customer=customer, sms_type=SMSType.VISIT)
|
||
self.assertEqual(item.status, SMSQueue.Status.APPROVED)
|
||
|
||
def test_auto_send_does_not_bypass_opt_out_or_disabled(self):
|
||
customer, template = self._overdue_visit_customer()
|
||
template.auto_send = True
|
||
template.is_enabled = False
|
||
template.save(update_fields=['auto_send', 'is_enabled'])
|
||
activate_sms()
|
||
self.assertEqual(run_all(timezone.localdate())['visit'], 0)
|
||
self.assertFalse(SMSQueue.objects.filter(customer=customer).exists())
|
||
|
||
def test_welcome_auto_send_skips_approval_on_customer_creation(self):
|
||
template = SMSTemplate.objects.get(sms_type=SMSType.WELCOME)
|
||
template.auto_send = True
|
||
template.save(update_fields=['auto_send'])
|
||
activate_sms()
|
||
self.client.post(
|
||
reverse('admin_panel:customersAdd'),
|
||
{
|
||
'an_name': 'مشتری خوش‌آمد خودکار', 'an_address': 'تهران',
|
||
'installDate': '1400/01/01', 'idnCode': '888',
|
||
'phone_titles': ['موبایل'], 'phones': ['09120000032'],
|
||
'phone_primary': ['0'],
|
||
},
|
||
HTTP_REFERER='/panel/',
|
||
)
|
||
item = SMSQueue.objects.get(sms_type=SMSType.WELCOME,
|
||
customer__name_family='مشتری خوش‌آمد خودکار')
|
||
self.assertEqual(item.status, SMSQueue.Status.APPROVED)
|
||
|
||
def test_template_edit_persists_auto_send(self):
|
||
template = SMSTemplate.objects.get(sms_type=SMSType.NESIYE_DUE)
|
||
url = reverse('admin_panel:sms_template_edit', kwargs={'t_id': template.id})
|
||
self.client.post(
|
||
url,
|
||
{'title': template.title, 'text': template.text,
|
||
'is_enabled': 'on', 'auto_send': 'on'},
|
||
HTTP_REFERER='/panel/',
|
||
)
|
||
template.refresh_from_db()
|
||
self.assertTrue(template.auto_send)
|
||
# ارسال دوبارهٔ فرم بدون تیک ⇒ خاموش می‌شود (چک‌باکس عدم‌حضور = خاموش).
|
||
self.client.post(
|
||
url,
|
||
{'title': template.title, 'text': template.text, 'is_enabled': 'on'},
|
||
HTTP_REFERER='/panel/',
|
||
)
|
||
template.refresh_from_db()
|
||
self.assertFalse(template.auto_send)
|
||
|
||
def test_templates_page_shows_auto_send_switch(self):
|
||
html = self.client.get(reverse('admin_panel:sms_templates')).content.decode('utf-8')
|
||
self.assertIn('name="auto_send"', html)
|
||
|
||
|
||
class PrimaryNumberPanelTests(PanelTestCase):
|
||
"""تعیین شمارهٔ اصلی از صفحهٔ مشتری، مستقل از اینکه شماره کجا ذخیره شده است."""
|
||
|
||
username = 'smsPrimary'
|
||
|
||
def test_toggle_marks_legacy_number_as_primary(self):
|
||
customer = make_customer('مشتری شماره قدیمی')
|
||
customer.phone1 = '9133722815'
|
||
customer.phone2 = '09121112222'
|
||
customer.save(update_fields=['phone1', 'phone2'])
|
||
self.client.post(
|
||
reverse('admin_panel:sms_primary_toggle', kwargs={'c_id': customer.id}),
|
||
{'phone': '09121112222', 'is_primary': 'true'},
|
||
HTTP_REFERER='/panel/',
|
||
)
|
||
customer.refresh_from_db()
|
||
self.assertEqual(customer.sms_primary_phones, '+989121112222')
|
||
self.assertEqual(resolve_recipients(customer), ['+989121112222'])
|
||
|
||
def test_toggle_accepts_number_without_leading_zero(self):
|
||
customer = make_customer('مشتری بدون صفر')
|
||
customer.phone1 = '9133722815'
|
||
customer.save(update_fields=['phone1'])
|
||
self.client.post(
|
||
reverse('admin_panel:sms_primary_toggle', kwargs={'c_id': customer.id}),
|
||
{'phone': '9133722815', 'is_primary': 'true'},
|
||
HTTP_REFERER='/panel/',
|
||
)
|
||
customer.refresh_from_db()
|
||
self.assertEqual(customer.sms_primary_phones, '+989133722815')
|
||
|
||
def test_toggle_rejects_unknown_number(self):
|
||
customer = make_customer('مشتری ناشناس')
|
||
self.client.post(
|
||
reverse('admin_panel:sms_primary_toggle', kwargs={'c_id': customer.id}),
|
||
{'phone': '09129999999', 'is_primary': 'true'},
|
||
HTTP_REFERER='/panel/',
|
||
)
|
||
customer.refresh_from_db()
|
||
self.assertEqual(customer.sms_primary_phones, '')
|
||
|
||
def test_customer_page_lists_legacy_numbers_and_toggles(self):
|
||
customer = make_customer('مشتری صفحه')
|
||
customer.phone1 = '9133722815'
|
||
customer.save(update_fields=['phone1'])
|
||
response = self.client.get(
|
||
reverse('admin_panel:customer', kwargs={'c_id': customer.id}),
|
||
)
|
||
self.assertEqual(response.status_code, 200)
|
||
self.assertContains(response, '09133722815')
|
||
self.assertContains(
|
||
response, reverse('admin_panel:sms_primary_toggle', kwargs={'c_id': customer.id}),
|
||
)
|
||
|
||
def test_primary_toggle_button_renders_next_to_phone_in_info_column(self):
|
||
"""دکمهٔ تعیین «اصلی» باید کنار شماره در ستون اطلاعات مشتری باشد (نه پایین صفحه)."""
|
||
customer = make_customer('مشتری ستون راست')
|
||
PhoneNumber.objects.create(customer=customer, phone_number='09120000000')
|
||
html = self.client.get(
|
||
reverse('admin_panel:customer', kwargs={'c_id': customer.id}),
|
||
).content.decode('utf-8')
|
||
self.assertIn('sms-primary-toggle', html)
|
||
self.assertIn('data-phone="09120000000"', html)
|
||
# URL به‌صورت data-url در ستون اطلاعات می‌آید (نه به‌شکل فرم قدیمی پایین صفحه).
|
||
self.assertIn(
|
||
'data-url="%s"' % reverse('admin_panel:sms_primary_toggle', kwargs={'c_id': customer.id}),
|
||
html,
|
||
)
|
||
|
||
def test_primary_marker_reflects_current_selection(self):
|
||
customer = make_customer('مشتری انتخاب‌شده')
|
||
PhoneNumber.objects.create(customer=customer, phone_number='09120000001')
|
||
PhoneNumber.objects.create(customer=customer, phone_number='09120000002')
|
||
mark_primary(customer, '09120000002')
|
||
html = self.client.get(
|
||
reverse('admin_panel:customer', kwargs={'c_id': customer.id}),
|
||
).content.decode('utf-8')
|
||
# شمارهٔ اصلی با ستارهٔ پرشده (★) و data-primary="true" علامت خورده است.
|
||
self.assertIn('data-phone="09120000002"', html)
|
||
self.assertIn('data-primary="true"', html)
|
||
self.assertIn('★', html)
|
||
|
||
def test_sms_box_no_longer_contains_primary_phone_list(self):
|
||
"""بخش حذف‌شده نباید دیگر در صفحه باشد."""
|
||
customer = make_customer('مشتری باکس')
|
||
PhoneNumber.objects.create(customer=customer, phone_number='09120000000')
|
||
html = self.client.get(
|
||
reverse('admin_panel:customer', kwargs={'c_id': customer.id}),
|
||
).content.decode('utf-8')
|
||
self.assertNotIn('شماره‌های اصلی (گیرنده پیامک)', html)
|
||
self.assertNotIn('گیرندگان فعلی', html)
|
||
|
||
|
||
|
||
class SMSPermissionTests(TestCase):
|
||
"""دسترسی به بخش پیامک دو سطح دارد:
|
||
|
||
* ``ManageSMS`` — مسیرهای عملیاتی (صف، گزارش، پیامک مشتری).
|
||
* ``ManageSMSSettings`` — صفحات حساس «تنظیمات سامانه» و «قالب‌ها».
|
||
|
||
واگذاری از Django Admin انجام می‌شود. کاربر بدون هیچ‌کدام، به هیچ مسیری دسترسی
|
||
ندارد؛ و کاربرِ فقط دارای ``ManageSMS`` نباید به تنظیمات/قالب‌ها دسترسی داشته باشد.
|
||
"""
|
||
|
||
# مسیرهای عملیاتی: کافی است ManageSMS باشد.
|
||
OPERATIONAL = [
|
||
('sms_queue', {}, 'get'),
|
||
('sms_report', {}, 'get'),
|
||
('sms_customer_toggle', {'c_id': None}, 'post'),
|
||
('sms_primary_toggle', {'c_id': None}, 'post'),
|
||
('sms_send_custom', {'c_id': None}, 'post'),
|
||
('sms_queue_retry', {'q_id': None}, 'post'),
|
||
('sms_queue_approve', {'q_id': None}, 'post'),
|
||
('sms_queue_bulk', {}, 'post'),
|
||
('sms_queue_empty', {}, 'post'),
|
||
('sms_queue_delete', {'q_id': None}, 'post'),
|
||
('customerSmsHistory', {'c_id': None}, 'get'),
|
||
]
|
||
# مسیرهای تنظیمات/قالب: نیاز به ManageSMSSettings.
|
||
SETTINGS = [
|
||
('sms_settings', {}, 'get'),
|
||
('sms_templates', {}, 'get'),
|
||
('sms_template_edit', {'t_id': None}, 'post'),
|
||
]
|
||
|
||
def setUp(self):
|
||
customer = make_customer('مشتری دسترسی')
|
||
self.phone = PhoneNumber.objects.create(
|
||
customer=customer, phone_number='09120000000',
|
||
)
|
||
self.template = SMSTemplate.objects.get(sms_type=SMSType.WELCOME)
|
||
self.queue_item = SMSQueue.objects.create(
|
||
customer=customer, sms_type=SMSType.WELCOME,
|
||
phone_number='+989120000000', message='سلام', status=SMSQueue.Status.PENDING,
|
||
)
|
||
self.fills = {
|
||
'c_id': self.phone.customer_id,
|
||
'q_id': self.queue_item.id,
|
||
't_id': self.template.id,
|
||
}
|
||
|
||
def _urls(self):
|
||
return self.OPERATIONAL + self.SETTINGS
|
||
|
||
def _resolve(self, kwargs):
|
||
return {k: (self.fills.get(k) if v is None else v) for k, v in kwargs.items()}
|
||
|
||
def _call(self, name, kwargs, method):
|
||
url = reverse(f'admin_panel:{name}', kwargs=self._resolve(kwargs))
|
||
headers = {'HTTP_REFERER': '/panel/'}
|
||
if method == 'get':
|
||
return self.client.get(url, **headers)
|
||
return self.client.post(url, {}, **headers)
|
||
|
||
def test_every_sms_url_is_covered_by_this_test(self):
|
||
"""اگر مسیر پیامکی اضافه شد، این تست یادآوری می‌کند که پوشش داده شود."""
|
||
from admin_panel.urls import urlpatterns
|
||
sms_names = {
|
||
p.name for p in urlpatterns if p.name and p.name.startswith('sms_')
|
||
}
|
||
covered = {name for name, _, _ in self._urls() if name.startswith('sms_')}
|
||
self.assertEqual(sms_names, covered)
|
||
|
||
def test_anonymous_user_is_redirected_to_login(self):
|
||
for name, kwargs, method in self._urls():
|
||
with self.subTest(url=name):
|
||
response = self._call(name, kwargs, method)
|
||
self.assertEqual(response.status_code, 302)
|
||
self.assertIn('/panel/p_login/', response['Location'])
|
||
|
||
def test_panel_user_without_permission_gets_403(self):
|
||
self.client.force_login(make_user('smsNoPerm'))
|
||
for name, kwargs, method in self._urls():
|
||
with self.subTest(url=name):
|
||
response = self._call(name, kwargs, method)
|
||
self.assertEqual(response.status_code, 403)
|
||
|
||
def test_superuser_has_access(self):
|
||
self.client.force_login(make_user('smsSuper', superuser=True))
|
||
for name, kwargs, method in self._urls():
|
||
with self.subTest(url=name):
|
||
self.assertNotIn(self._call(name, kwargs, method).status_code, (403, 405))
|
||
|
||
def test_full_permissions_grant_access(self):
|
||
# همان کاری که Django Admin انجام می‌دهد: افزودن Permission به کاربر.
|
||
self.client.force_login(
|
||
make_user('smsAllowed', perms=['ManageSMS', 'ManageSMSSettings']))
|
||
for name, kwargs, method in self._urls():
|
||
with self.subTest(url=name):
|
||
response = self._call(name, kwargs, method)
|
||
self.assertNotIn(response.status_code, (403, 405))
|
||
|
||
def test_manage_sms_alone_cannot_reach_settings_pages(self):
|
||
"""کاربر فقط با ManageSMS نباید تنظیمات/قالب‌ها را ببیند یا ویرایش کند."""
|
||
self.client.force_login(make_user('smsOpsOnly', perms=['ManageSMS']))
|
||
for name, kwargs, method in self.SETTINGS:
|
||
with self.subTest(url=name):
|
||
self.assertEqual(self._call(name, kwargs, method).status_code, 403)
|
||
# ولی مسیرهای عملیاتی برایش باز است.
|
||
self.assertNotEqual(self._call('sms_queue', {}, 'get').status_code, 403)
|
||
|
||
def test_manage_sms_settings_alone_can_open_settings_pages(self):
|
||
"""کاربر فقط با ManageSMSSettings باید تنظیمات/قالب‌ها را ببیند."""
|
||
self.client.force_login(make_user('smsCfgOnly', perms=['ManageSMSSettings']))
|
||
for name, kwargs, method in self.SETTINGS:
|
||
with self.subTest(url=name):
|
||
self.assertNotEqual(self._call(name, kwargs, method).status_code, 403)
|
||
# ولی به مسیرهای عملیاتی دسترسی ندارد.
|
||
self.assertEqual(self._call('sms_queue', {}, 'get').status_code, 403)
|
||
|
||
def test_permission_granted_via_group_grants_access(self):
|
||
from django.contrib.auth.models import Group, Permission
|
||
group = Group.objects.create(name='اپراتور پیامک')
|
||
group.permissions.add(Permission.objects.get(codename='ManageSMS'))
|
||
group.permissions.add(Permission.objects.get(codename='ManageSMSSettings'))
|
||
user = make_user('smsGrouped')
|
||
user.groups.add(group)
|
||
self.client.force_login(user)
|
||
self.assertNotEqual(
|
||
self._call('sms_settings', {}, 'get').status_code, 403,
|
||
)
|
||
self.assertNotEqual(
|
||
self._call('sms_queue', {}, 'get').status_code, 403,
|
||
)
|
||
|
||
def test_navbar_links_visible_only_with_permission(self):
|
||
self.client.force_login(make_user('smsNavNoPerm'))
|
||
hidden = self.client.get(reverse('admin_panel:index'))
|
||
self.assertNotContains(hidden, reverse('admin_panel:sms_settings'))
|
||
self.assertNotContains(hidden, reverse('admin_panel:sms_queue'))
|
||
|
||
# با ManageSMS: فقط صف/گزارش دیده می‌شود، تنظیمات نه.
|
||
self.client.force_login(make_user('smsNavOps', perms=['ManageSMS']))
|
||
ops = self.client.get(reverse('admin_panel:index'))
|
||
self.assertContains(ops, reverse('admin_panel:sms_queue'))
|
||
self.assertNotContains(ops, reverse('admin_panel:sms_settings'))
|
||
|
||
# با ManageSMSSettings: تنظیمات دیده می‌شود، صف نه.
|
||
self.client.force_login(make_user('smsNavCfg', perms=['ManageSMSSettings']))
|
||
cfg = self.client.get(reverse('admin_panel:index'))
|
||
self.assertContains(cfg, reverse('admin_panel:sms_settings'))
|
||
self.assertNotContains(cfg, reverse('admin_panel:sms_queue'))
|
||
|
||
def test_permissions_are_assignable_from_django_admin_user_page(self):
|
||
"""مسیر واگذاری: Django Admin → کاربر → User permissions."""
|
||
admin_user = make_user('smsSiteAdmin', superuser=True)
|
||
target = make_user('smsTarget')
|
||
self.client.force_login(admin_user)
|
||
response = self.client.get(
|
||
reverse('admin:auth_user_change', args=[target.id]),
|
||
)
|
||
self.assertEqual(response.status_code, 200)
|
||
self.assertContains(response, 'ManageSMS')
|
||
self.assertContains(response, 'ManageSMSSettings')
|
||
|
||
def test_customer_page_hides_sms_box_without_permission(self):
|
||
self.client.force_login(make_user('smsCustNoPerm'))
|
||
response = self.client.get(
|
||
reverse('admin_panel:customer2', kwargs={'c_id': self.phone.customer_id}),
|
||
)
|
||
self.assertNotContains(response, reverse('admin_panel:sms_send_custom',
|
||
kwargs={'c_id': self.phone.customer_id}))
|
||
# دکمهٔ تعیین شمارهٔ اصلی (منتقل‌شده به ستون اطلاعات) هم نباید دیده شود.
|
||
self.assertNotContains(response, reverse('admin_panel:sms_primary_toggle',
|
||
kwargs={'c_id': self.phone.customer_id}))
|
||
self.assertNotContains(response, 'sms-primary-toggle')
|
||
|
||
|
||
class SettingsPanelTests(PanelTestCase):
|
||
username = 'smsSettings'
|
||
|
||
def test_flow_controls_are_saved_from_panel(self):
|
||
response = self.client.post(
|
||
reverse('admin_panel:sms_settings'),
|
||
{
|
||
'provider': 'console', 'is_enabled': 'on',
|
||
'scan_since': '2026-01-01 08:30:00',
|
||
'allow_backlog': 'on', 'backlog_max': '7',
|
||
'max_per_run': '25', 'send_interval_seconds': '3',
|
||
},
|
||
HTTP_REFERER='/panel/',
|
||
)
|
||
self.assertEqual(response.status_code, 302)
|
||
config = SMSProviderConfig.load()
|
||
self.assertTrue(config.allow_backlog)
|
||
self.assertEqual(config.backlog_max, 7)
|
||
self.assertEqual(config.max_per_run, 25)
|
||
self.assertEqual(config.send_interval_seconds, 3)
|
||
self.assertEqual(
|
||
timezone.localtime(config.scan_since).strftime('%Y-%m-%d %H:%M:%S'),
|
||
'2026-01-01 08:30:00',
|
||
)
|
||
|
||
def test_jalali_period_start_saved_as_start_of_day(self):
|
||
"""ورودی شمسی (YYYY/MM/DD) روی «آغاز همان روز» در تایم‌زون پروژه ذخیره می‌شود."""
|
||
from persiantools.jdatetime import JalaliDate
|
||
|
||
self.client.post(
|
||
reverse('admin_panel:sms_settings'),
|
||
{'provider': 'console', 'scan_since': '1403/07/18'},
|
||
HTTP_REFERER='/panel/',
|
||
)
|
||
local = timezone.localtime(SMSProviderConfig.load().scan_since)
|
||
self.assertEqual(local.date(), JalaliDate(1403, 7, 18).to_gregorian())
|
||
self.assertEqual((local.hour, local.minute, local.second), (0, 0, 0))
|
||
|
||
def test_invalid_scan_since_keeps_previous_value(self):
|
||
config = SMSProviderConfig.load()
|
||
previous = config.scan_since
|
||
self.client.post(
|
||
reverse('admin_panel:sms_settings'),
|
||
{'provider': 'console', 'scan_since': 'not-a-date'},
|
||
HTTP_REFERER='/panel/',
|
||
)
|
||
self.assertEqual(SMSProviderConfig.load().scan_since, previous)
|
||
|
||
|
||
class WelcomeTriggerTests(PanelTestCase):
|
||
username = 'smsAdmin'
|
||
|
||
def test_customer_creation_enqueues_welcome(self):
|
||
response = self.client.post(
|
||
reverse('admin_panel:customersAdd'),
|
||
{
|
||
'an_name': 'مشتری پیامکی', 'an_address': 'تهران',
|
||
'installDate': '1400/01/01', 'idnCode': '999',
|
||
'phone_titles': ['موبایل'], 'phones': ['09120000000'],
|
||
'phone_primary': ['0'],
|
||
},
|
||
HTTP_REFERER='/panel/',
|
||
)
|
||
self.assertIn(response.status_code, (200, 302))
|
||
customer = Customer.objects.get(name_family='مشتری پیامکی')
|
||
# شماره‌ای که «اصلی» علامت خورده در فهرست شماره‌های اصلی ذخیره می‌شود.
|
||
self.assertEqual(customer.sms_primary_phones, '+989120000000')
|
||
queue = SMSQueue.objects.filter(customer=customer, sms_type=SMSType.WELCOME)
|
||
self.assertEqual(queue.count(), 1)
|
||
self.assertEqual(queue.first().phone_number, '+989120000000')
|
||
|
||
def test_customer_edit_preserves_primary_selection(self):
|
||
customer = make_customer('مشتری ویرایشی')
|
||
PhoneNumber.objects.create(
|
||
customer=customer, title='موبایل', phone_number='09120000000',
|
||
)
|
||
mark_primary(customer, '09120000000')
|
||
response = self.client.post(
|
||
reverse('admin_panel:customerEdit', kwargs={'c_id': customer.id}),
|
||
{
|
||
'an_name': 'مشتری ویرایشی', 'an_address': 'تهران',
|
||
'installDate': '1400/01/01', 'idnCode': '777',
|
||
'phone_titles': ['موبایل'], 'phones': ['09120000000'],
|
||
},
|
||
HTTP_REFERER='/panel/',
|
||
)
|
||
self.assertIn(response.status_code, (200, 302))
|
||
customer.refresh_from_db()
|
||
# انتخابِ اصلی به رکورد وابسته نیست؛ با بازسازی رکوردها از بین نمی‌رود.
|
||
self.assertEqual(customer.sms_primary_phones, '+989120000000')
|
||
self.assertEqual(resolve_recipients(customer), ['+989120000000'])
|
||
|
||
|
||
class QueueReviewTests(PanelTestCase):
|
||
"""دروازهٔ بازبینی مدیر: ارسال فقط پس از تأیید صریح، به‌همراه حذف از صف."""
|
||
|
||
username = 'smsReview'
|
||
|
||
def _item(self, **kwargs):
|
||
defaults = dict(
|
||
customer=make_customer('مشتری صف'), sms_type=SMSType.CHEQUE_REGISTRATION,
|
||
phone_number='+989120000000', message='یادآوری',
|
||
)
|
||
defaults.update(kwargs)
|
||
return SMSQueue.objects.create(**defaults)
|
||
|
||
def test_scan_enqueue_lands_in_pending_review(self):
|
||
item = self._item()
|
||
self.assertEqual(item.status, SMSQueue.Status.PENDING)
|
||
|
||
def test_approve_view_moves_item_to_approved(self):
|
||
item = self._item()
|
||
self.client.post(reverse('admin_panel:sms_queue_approve', kwargs={'q_id': item.id}))
|
||
item.refresh_from_db()
|
||
self.assertEqual(item.status, SMSQueue.Status.APPROVED)
|
||
|
||
def test_approve_view_ignores_non_pending_items(self):
|
||
item = self._item(status=SMSQueue.Status.SENT)
|
||
self.client.post(reverse('admin_panel:sms_queue_approve', kwargs={'q_id': item.id}))
|
||
item.refresh_from_db()
|
||
self.assertEqual(item.status, SMSQueue.Status.SENT)
|
||
|
||
def test_delete_view_removes_item_from_queue(self):
|
||
item = self._item()
|
||
self.client.post(reverse('admin_panel:sms_queue_delete', kwargs={'q_id': item.id}))
|
||
self.assertFalse(SMSQueue.objects.filter(id=item.id).exists())
|
||
|
||
def test_bulk_approve_all_approves_only_pending(self):
|
||
pending = self._item()
|
||
sent = self._item(status=SMSQueue.Status.SENT)
|
||
self.client.post(reverse('admin_panel:sms_queue_bulk'), {'action': 'approve_all'})
|
||
pending.refresh_from_db()
|
||
sent.refresh_from_db()
|
||
self.assertEqual(pending.status, SMSQueue.Status.APPROVED)
|
||
self.assertEqual(sent.status, SMSQueue.Status.SENT)
|
||
|
||
def test_bulk_delete_selected_removes_only_chosen(self):
|
||
chosen = self._item()
|
||
kept = self._item()
|
||
self.client.post(
|
||
reverse('admin_panel:sms_queue_bulk'),
|
||
{'action': 'delete_selected', 'ids': [chosen.id]},
|
||
)
|
||
self.assertFalse(SMSQueue.objects.filter(id=chosen.id).exists())
|
||
self.assertTrue(SMSQueue.objects.filter(id=kept.id).exists())
|
||
|
||
def test_custom_manual_send_is_auto_approved(self):
|
||
customer = make_customer('مشتری دستی')
|
||
PhoneNumber.objects.create(customer=customer, phone_number='09120000000')
|
||
mark_primary(customer, '09120000000')
|
||
self.client.post(
|
||
reverse('admin_panel:sms_send_custom', kwargs={'c_id': customer.id}),
|
||
{'text': 'سلام دستی'},
|
||
)
|
||
item = SMSQueue.objects.get(customer=customer)
|
||
self.assertEqual(item.status, SMSQueue.Status.APPROVED)
|
||
|
||
|
||
class NoticeMechanismTests(PanelTestCase):
|
||
"""پیام نتیجهٔ عملیات از طریق ``?notice=<code>`` رد و بدل می‌شود، نه flash message.
|
||
|
||
پیام‌های جنگو در session می‌مانند و اگر قالب مقصد بلوک پیام نداشته باشد، در
|
||
صفحهٔ بعدیِ دارای بلوک «سرگردان» ظاهر می‌شوند؛ این‌جا کد فقط روی همان URL وجود
|
||
دارد پس پیام هم فقط همان‌جا دیده می‌شود.
|
||
"""
|
||
|
||
username = 'smsNotice'
|
||
|
||
def _item(self, **kwargs):
|
||
defaults = dict(
|
||
customer=make_customer('مشتری اعلان'), sms_type=SMSType.CHEQUE_REGISTRATION,
|
||
phone_number='+989120000000', message='یادآوری',
|
||
)
|
||
defaults.update(kwargs)
|
||
return SMSQueue.objects.create(**defaults)
|
||
|
||
def test_resolve_returns_text_and_kind_for_known_code(self):
|
||
from admin_panel.sms.notices import resolve
|
||
|
||
request = mock.Mock(GET={'notice': 'approved'})
|
||
data = resolve(request)
|
||
self.assertEqual(data['notice_kind'], 'is-success')
|
||
self.assertIn('تأیید', data['notice_text'])
|
||
|
||
def test_resolve_ignores_unknown_or_missing_code(self):
|
||
from admin_panel.sms.notices import resolve
|
||
|
||
self.assertEqual(resolve(mock.Mock(GET={'notice': 'nope'})), {})
|
||
self.assertEqual(resolve(mock.Mock(GET={})), {})
|
||
|
||
def test_resolve_fills_placeholders_from_querystring(self):
|
||
from admin_panel.sms.notices import resolve
|
||
|
||
request = mock.Mock(GET={'notice': 'tpl_saved', 'title': 'خوش‌آمدگویی'})
|
||
self.assertIn('خوش‌آمدگویی', resolve(request)['notice_text'])
|
||
|
||
def test_approve_redirects_with_notice_code_not_session_message(self):
|
||
item = self._item()
|
||
response = self.client.post(
|
||
reverse('admin_panel:sms_queue_approve', kwargs={'q_id': item.id})
|
||
)
|
||
self.assertEqual(response.status_code, 302)
|
||
self.assertIn('notice=approved', response.url)
|
||
# هیچ پیام sessionی باقی نمی‌ماند.
|
||
self.assertEqual(list(self.client.session.get('_messages', [])), [])
|
||
|
||
def test_rendered_page_shows_the_notice(self):
|
||
url = reverse('admin_panel:sms_queue') + '?notice=deleted'
|
||
response = self.client.get(url)
|
||
self.assertContains(response, 'پیامک از صف حذف شد.')
|
||
|
||
def test_bulk_counts_are_passed_through_querystring(self):
|
||
self._item()
|
||
response = self.client.post(
|
||
reverse('admin_panel:sms_queue_bulk'), {'action': 'approve_all'}
|
||
)
|
||
self.assertIn('notice=bulk_approved_all', response.url)
|
||
self.assertIn('count=1', response.url)
|
||
|
||
|
||
class TestMessageSettingsTests(PanelTestCase):
|
||
"""متن پیامک آزمایشی از پنل قابل ویرایش است و در ارسال‌های بعدی می‌ماند."""
|
||
|
||
username = 'smsTestMsg'
|
||
|
||
def test_editing_test_message_persists_and_is_sent(self):
|
||
response = self.client.post(reverse('admin_panel:sms_settings'), {
|
||
'action': 'test', 'test_phone': '09120000000',
|
||
'test_message': 'متن آزمایشی دلخواه',
|
||
})
|
||
self.assertEqual(response.status_code, 302)
|
||
self.assertEqual(SMSProviderConfig.load().test_message, 'متن آزمایشی دلخواه')
|
||
self.assertTrue(
|
||
SMSQueue.objects.filter(message='متن آزمایشی دلخواه').exists()
|
||
)
|
||
|
||
def test_default_test_message_has_no_brand_name(self):
|
||
text = SMSProviderConfig.load().test_message
|
||
self.assertNotIn('کتیبه', text)
|
||
|
||
|
||
class DeletedStaysDeletedTests(PanelTestCase):
|
||
"""پیامکِ حذف‌شده از صف با اجرای مجدد برنامه بازنمی‌گردد.
|
||
|
||
کلید یکتاسازی در دفتر ``SMSDedup`` می‌ماند، پس اسکنر (که در هر اجرا همان
|
||
کلیدهای تکرارشونده را می‌سازد) دیگر آن را از نو در صف نمی‌گذارد.
|
||
"""
|
||
|
||
username = 'smsNoRevive'
|
||
|
||
def test_deleted_queue_item_is_not_re_enqueued(self):
|
||
customer = make_customer('مشتری حذف')
|
||
PhoneNumber.objects.create(customer=customer, phone_number='09120000000')
|
||
mark_primary(customer, '09120000000')
|
||
key = f'cheque_fund:{customer.id}'
|
||
self.assertEqual(
|
||
enqueue_sms(customer=customer, sms_type=SMSType.CHEQUE_REGISTRATION,
|
||
message='یادآوری', dedup_key=key),
|
||
1,
|
||
)
|
||
SMSQueue.objects.filter(dedup_key=key).delete()
|
||
# اجرای دوبارهٔ همان اسکنر: چون کلید در دفتر ثبت شده، صفر ساخته می‌شود.
|
||
self.assertEqual(
|
||
enqueue_sms(customer=customer, sms_type=SMSType.CHEQUE_REGISTRATION,
|
||
message='یادآوری', dedup_key=key),
|
||
0,
|
||
)
|
||
self.assertFalse(SMSQueue.objects.filter(dedup_key=key).exists())
|
||
|
||
def test_delete_view_then_rescan_creates_nothing(self):
|
||
from admin_panel.sms.scanner import run_all
|
||
|
||
customer = make_customer('مشتری اسکن')
|
||
PhoneNumber.objects.create(customer=customer, phone_number='09120000000')
|
||
mark_primary(customer, '09120000000')
|
||
Sarresid.objects.create(
|
||
customer=customer, price=5000, payStatus='false', type='check',
|
||
sarresidDT=timezone.now() + timedelta(days=1),
|
||
)
|
||
activate_sms()
|
||
template = SMSTemplate.objects.get(sms_type=SMSType.CHEQUE_REGISTRATION)
|
||
apply_timing(template, 1)
|
||
template.save()
|
||
self.assertEqual(run_all(timezone.localdate())['cheque_registration'], 1)
|
||
|
||
item = SMSQueue.objects.get(sms_type=SMSType.CHEQUE_REGISTRATION)
|
||
self.client.post(reverse('admin_panel:sms_queue_delete', kwargs={'q_id': item.id}))
|
||
self.assertFalse(SMSQueue.objects.filter(id=item.id).exists())
|
||
|
||
# اجرای مجدد اسکنر: صفر مورد جدید.
|
||
self.assertEqual(run_all(timezone.localdate())['cheque_registration'], 0)
|
||
self.assertEqual(
|
||
SMSQueue.objects.filter(sms_type=SMSType.CHEQUE_REGISTRATION).count(), 0,
|
||
)
|
||
|
||
|
||
class QueueStatsTests(PanelTestCase):
|
||
"""صفحهٔ صف، آمار صف را نشان می‌دهد؛ ارسال‌شده‌ها از صف بیرون می‌مانند."""
|
||
|
||
username = 'smsStats'
|
||
|
||
def _item(self, status):
|
||
return SMSQueue.objects.create(
|
||
customer=make_customer('مشتری آمار'), sms_type=SMSType.WELCOME,
|
||
phone_number='+989120000000', message='x', status=status,
|
||
)
|
||
|
||
def test_page_shows_queue_total_and_per_status_counts(self):
|
||
self._item(SMSQueue.Status.PENDING)
|
||
self._item(SMSQueue.Status.PENDING)
|
||
self._item(SMSQueue.Status.SENT)
|
||
response = self.client.get(reverse('admin_panel:sms_queue'))
|
||
# «کل صف» فقط موارد باقی‌مانده است (بدون ارسال‌شده).
|
||
self.assertEqual(response.context['total_count'], 2)
|
||
self.assertEqual(response.context['pending_count'], 2)
|
||
# ارسال‌شده‌ها جداگانه شمرده می‌شوند (کارت گزارش).
|
||
self.assertEqual(response.context['sent_count'], 1)
|
||
stats = dict((code, n) for code, _label, n in response.context['status_stats'])
|
||
self.assertEqual(stats['pending'], 2)
|
||
self.assertNotIn('sent', stats)
|
||
self.assertEqual(stats['failed'], 0)
|
||
|
||
def test_sent_rows_are_excluded_from_listing(self):
|
||
self._item(SMSQueue.Status.PENDING)
|
||
self._item(SMSQueue.Status.SENT)
|
||
response = self.client.get(reverse('admin_panel:sms_queue'))
|
||
ids = [item.id for item in response.context['queue']]
|
||
sent = SMSQueue.objects.get(status=SMSQueue.Status.SENT)
|
||
self.assertNotIn(sent.id, ids)
|
||
|
||
def test_status_filter_does_not_offer_sent(self):
|
||
response = self.client.get(reverse('admin_panel:sms_queue'))
|
||
codes = [code for code, _label in response.context['status_choices']]
|
||
self.assertNotIn('sent', codes)
|
||
self.assertIn('pending', codes)
|
||
|
||
def test_list_is_paginated(self):
|
||
for _ in range(55):
|
||
self._item(SMSQueue.Status.FAILED)
|
||
response = self.client.get(reverse('admin_panel:sms_queue'))
|
||
self.assertEqual(len(response.context['queue']), 50)
|
||
self.assertEqual(response.context['queue'].paginator.num_pages, 2)
|
||
|
||
def test_customer_name_links_to_customer_page(self):
|
||
customer = make_customer('مشتری لینک')
|
||
SMSQueue.objects.create(
|
||
customer=customer, sms_type=SMSType.WELCOME,
|
||
phone_number='+989120000000', message='x',
|
||
)
|
||
response = self.client.get(reverse('admin_panel:sms_queue'))
|
||
html = response.content.decode()
|
||
self.assertIn(
|
||
reverse('admin_panel:customer', kwargs={'c_id': customer.id}), html,
|
||
)
|
||
self.assertIn('مشتری لینک', html)
|
||
|
||
def test_dates_render_as_jalali_datetime(self):
|
||
import re
|
||
|
||
self._item(SMSQueue.Status.PENDING)
|
||
response = self.client.get(reverse('admin_panel:sms_queue'))
|
||
html = response.content.decode()
|
||
# تاریخ شمسی با ساعت، و هیچ ISO میلادی باقی نمانده باشد.
|
||
self.assertRegex(html, r'\d{4}/\d{2}/\d{2} \d{2}:\d{2}')
|
||
self.assertIsNone(re.search(r'\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}', html))
|
||
|
||
def test_jalali_datetime_filter_handles_none_and_localizes(self):
|
||
from admin_panel.templatetags.custom_tags import toJalaliDT
|
||
|
||
self.assertEqual(toJalaliDT(None), '')
|
||
from django.utils import timezone
|
||
|
||
dt = timezone.localtime(timezone.now()).replace(hour=13, minute=5)
|
||
self.assertTrue(toJalaliDT(dt).endswith('13:05'))
|
||
|
||
|
||
class EmptyQueueTests(PanelTestCase):
|
||
"""خالی کردن صف: موارد باقی‌مانده حذف و کلیدها در دفتر حفظ می‌شوند.
|
||
|
||
پیامک‌های ارسال‌شده جزو صف نیستند و حذف نمی‌شوند (تاریخچه در گزارش ارسال می‌ماند).
|
||
"""
|
||
|
||
username = 'smsEmpty'
|
||
|
||
def test_empty_removes_queue_rows_but_keeps_sent(self):
|
||
for status in (SMSQueue.Status.PENDING, SMSQueue.Status.APPROVED,
|
||
SMSQueue.Status.SENT, SMSQueue.Status.FAILED):
|
||
SMSQueue.objects.create(
|
||
customer=make_customer('مشتری'), sms_type=SMSType.WELCOME,
|
||
phone_number='+989120000000', message='x', status=status,
|
||
dedup_key=f'custom:{status}',
|
||
)
|
||
response = self.client.post(reverse('admin_panel:sms_queue_empty'))
|
||
self.assertEqual(response.status_code, 302)
|
||
self.assertIn('notice=queue_emptied', response.url)
|
||
# فقط ارسال‌شده‌ها می‌مانند.
|
||
self.assertEqual(SMSQueue.objects.count(), 1)
|
||
self.assertTrue(
|
||
SMSQueue.objects.filter(status=SMSQueue.Status.SENT).exists()
|
||
)
|
||
|
||
def test_emptied_keys_are_ledgered_so_nothing_returns(self):
|
||
SMSQueue.objects.create(
|
||
customer=make_customer('مشتری'), sms_type=SMSType.WELCOME,
|
||
phone_number='+989120000000', message='x', dedup_key='custom:keep',
|
||
)
|
||
self.client.post(reverse('admin_panel:sms_queue_empty'))
|
||
self.assertTrue(SMSDedup.objects.filter(key='custom:keep').exists())
|
||
self.assertEqual(
|
||
enqueue_sms(customer=None, sms_type=SMSType.WELCOME, message='x',
|
||
recipients=['+989120000000'], dedup_key='custom:keep'),
|
||
0,
|
||
)
|
||
self.assertEqual(SMSQueue.objects.count(), 0)
|
||
|
||
def test_get_does_not_empty(self):
|
||
SMSQueue.objects.create(
|
||
customer=make_customer('مشتری'), sms_type=SMSType.WELCOME,
|
||
phone_number='+989120000000', message='x',
|
||
)
|
||
self.client.get(reverse('admin_panel:sms_queue_empty'))
|
||
self.assertEqual(SMSQueue.objects.count(), 1)
|
||
|
||
|
||
class SentQueueProtectionTests(PanelTestCase):
|
||
"""پیامک ارسال‌شده از صف حذف نمی‌شود؛ سابقه‌اش در «گزارش ارسال» می‌ماند."""
|
||
|
||
username = 'smsSentProt'
|
||
|
||
def _sent(self):
|
||
return SMSQueue.objects.create(
|
||
customer=make_customer('مشتری ارسال'), sms_type=SMSType.WELCOME,
|
||
phone_number='+989120000000', message='x', status=SMSQueue.Status.SENT,
|
||
)
|
||
|
||
def test_delete_single_sent_is_blocked(self):
|
||
item = self._sent()
|
||
response = self.client.post(
|
||
reverse('admin_panel:sms_queue_delete', kwargs={'q_id': item.id}),
|
||
)
|
||
self.assertEqual(response.status_code, 302)
|
||
self.assertIn('notice=delete_done_blocked', response.url)
|
||
self.assertTrue(SMSQueue.objects.filter(id=item.id).exists())
|
||
|
||
def test_bulk_delete_skips_sent_rows(self):
|
||
sent = self._sent()
|
||
pending = SMSQueue.objects.create(
|
||
customer=make_customer('مشتری درانتظار'), sms_type=SMSType.WELCOME,
|
||
phone_number='+989120000001', message='y', status=SMSQueue.Status.PENDING,
|
||
)
|
||
self.client.post(
|
||
reverse('admin_panel:sms_queue_bulk'),
|
||
{'action': 'delete_selected', 'ids': [sent.id, pending.id]},
|
||
)
|
||
self.assertTrue(SMSQueue.objects.filter(id=sent.id).exists())
|
||
self.assertFalse(SMSQueue.objects.filter(id=pending.id).exists())
|
||
|
||
def test_queue_page_has_no_delete_button_for_sent(self):
|
||
item = self._sent()
|
||
html = self.client.get(reverse('admin_panel:sms_queue')).content.decode()
|
||
self.assertNotIn(
|
||
reverse('admin_panel:sms_queue_delete', kwargs={'q_id': item.id}), html,
|
||
)
|
||
|
||
|
||
class PhoneCleanupTests(TestCase):
|
||
"""قاعدهٔ پاک‌سازی فیلدهای تلفن قدیمی و idempotent بودنش."""
|
||
|
||
def test_mobile_without_leading_zero_becomes_canonical(self):
|
||
from admin_panel.sms.phone_cleanup import clean_legacy_cell
|
||
|
||
# مقدار استاندارد می‌شود و متن اصلی جداگانه برمی‌گردد تا در notes بماند.
|
||
self.assertEqual(clean_legacy_cell('9133722815'), ('09133722815', '9133722815'))
|
||
self.assertEqual(clean_legacy_cell('۰۹۱۳۳۷۲۲۸۱۵'), ('09133722815', '۰۹۱۳۳۷۲۲۸۱۵'))
|
||
self.assertEqual(clean_legacy_cell('0913-372-2815'), ('09133722815', '0913-372-2815'))
|
||
|
||
def test_canonical_cell_is_noop(self):
|
||
from admin_panel.sms.phone_cleanup import clean_legacy_cell
|
||
|
||
self.assertEqual(clean_legacy_cell('09133722815'), (None, None))
|
||
self.assertEqual(clean_legacy_cell(''), (None, None))
|
||
|
||
def test_embedded_mobile_is_extracted_and_text_noted(self):
|
||
from admin_panel.sms.phone_cleanup import clean_legacy_cell
|
||
|
||
value, note = clean_legacy_cell('موبایل: 09133722815')
|
||
self.assertEqual(value, '09133722815')
|
||
self.assertEqual(note, 'موبایل: 09133722815')
|
||
|
||
def test_eight_digit_landline_untouched(self):
|
||
from admin_panel.sms.phone_cleanup import clean_legacy_cell
|
||
|
||
self.assertEqual(clean_legacy_cell('57387608'), (None, None))
|
||
|
||
def test_name_only_cell_cleared_and_noted(self):
|
||
from admin_panel.sms.phone_cleanup import clean_legacy_cell
|
||
|
||
self.assertEqual(clean_legacy_cell('اعظمی'), ('', 'اعظمی'))
|
||
|
||
def test_pure_junk_fragment_cleared_but_preserved_in_note(self):
|
||
from admin_panel.sms.phone_cleanup import clean_legacy_cell
|
||
|
||
self.assertEqual(clean_legacy_cell('0'), ('', '0'))
|
||
self.assertEqual(clean_legacy_cell('573'), ('', '573'))
|
||
|
||
def test_embedded_mobile_keeps_dropped_landline_in_note(self):
|
||
from admin_panel.sms.phone_cleanup import clean_legacy_cell
|
||
|
||
value, note = clean_legacy_cell('57322182/9130261671')
|
||
self.assertEqual(value, '09130261671')
|
||
self.assertEqual(note, '57322182/9130261671')
|
||
|
||
def test_clean_customer_is_idempotent(self):
|
||
from admin_panel.sms.phone_cleanup import clean_customer
|
||
|
||
customer = make_customer('مشتری', phones=[])
|
||
customer.phone1, customer.phone2, customer.phone3 = '9133722815', 'اعظمی', '09120000000'
|
||
customer.save()
|
||
self.assertEqual(clean_customer(customer), 2)
|
||
customer.refresh_from_db()
|
||
self.assertEqual(customer.phone1, '09133722815')
|
||
self.assertEqual(customer.phone2, '')
|
||
self.assertEqual(customer.phone3, '09120000000')
|
||
self.assertIn('اعظمی', customer.notes)
|
||
self.assertEqual(clean_customer(customer), 0)
|
||
again = Customer.objects.get(id=customer.id)
|
||
self.assertEqual(again.notes.count('پاک‌سازی خودکار'), 1)
|
||
|
||
def test_clean_all_reports_stats(self):
|
||
from admin_panel.sms.phone_cleanup import clean_all
|
||
|
||
customer = make_customer('مشتری')
|
||
customer.phone1 = '9133722815'
|
||
customer.save()
|
||
make_customer('دست‌نخورده') # بدون شمارهٔ مشکل‌دار
|
||
stats = clean_all(dry_run=True)
|
||
self.assertGreaterEqual(stats['customers'], 1)
|
||
self.assertGreaterEqual(stats['fields'], 1)
|
||
|
||
def test_command_dry_run_is_noop(self):
|
||
from io import StringIO
|
||
|
||
from django.core.management import call_command
|
||
|
||
customer = make_customer('مشتری')
|
||
customer.phone1 = '9133722815'
|
||
customer.save()
|
||
out = StringIO()
|
||
call_command('fix_customer_phones', '--dry-run', stdout=out)
|
||
customer.refresh_from_db()
|
||
self.assertEqual(customer.phone1, '9133722815')
|
||
|
||
def test_command_applies_and_reruns_clean(self):
|
||
from io import StringIO
|
||
|
||
from django.core.management import call_command
|
||
|
||
customer = make_customer('مشتری')
|
||
customer.phone1 = '9133722815'
|
||
customer.save()
|
||
call_command('fix_customer_phones', stdout=StringIO())
|
||
customer.refresh_from_db()
|
||
self.assertEqual(customer.phone1, '09133722815')
|
||
call_command('fix_customer_phones', stdout=StringIO())
|
||
customer.refresh_from_db()
|
||
self.assertEqual(customer.phone1, '09133722815')
|
||
self.assertIn('9133722815', customer.notes) # مقدار اصلی حفظ شده
|
||
self.assertEqual(customer.notes.count('پاک‌سازی خودکار'), 1) # اجرای دوباره تکرار نمی‌کند
|
||
|
||
|
||
class EventTriggerTests(PanelTestCase):
|
||
"""تریگرهای رویدادی پیامک (بدون signal): نظرسنجی، ثبت بازدید و تشکر از پرداخت."""
|
||
|
||
username = 'smsEvents'
|
||
|
||
def _customer_with_phone(self, name='مشتری رویداد'):
|
||
customer = make_customer(name)
|
||
PhoneNumber.objects.create(customer=customer, phone_number='09120000000')
|
||
mark_primary(customer, '09120000000')
|
||
return customer
|
||
|
||
# ---------- قرارداد: دقیقاً ۷ نوع ----------
|
||
|
||
def test_exactly_seven_sms_types(self):
|
||
self.assertEqual(len(SMSType.choices), 7)
|
||
self.assertEqual(
|
||
{code for code, _label in SMSType.choices},
|
||
{'welcome', 'survey', 'cheque_registration', 'visit',
|
||
'visit_created', 'payment_thanks', 'nesiye_due'},
|
||
)
|
||
|
||
def test_removed_types_have_no_template(self):
|
||
self.assertFalse(
|
||
SMSTemplate.objects.filter(
|
||
sms_type__in=['cheque_due', 'debt_due', 'custom'],
|
||
).exists(),
|
||
)
|
||
|
||
def test_every_type_has_a_seeded_template(self):
|
||
seeded = set(SMSTemplate.objects.values_list('sms_type', flat=True))
|
||
self.assertEqual(seeded, {code for code, _label in SMSType.choices})
|
||
|
||
# ---------- نظرسنجی + تشکر از پرداخت نقدی روی فاکتور ----------
|
||
|
||
def _post_factor(self, customer, payments):
|
||
store = make_store('انبار رویداد', owner=self.user)
|
||
product = make_product('کالای رویداد', price=1000)
|
||
stock_in(store, product, 20)
|
||
url = reverse('admin_panel:newFactor2', kwargs={'c_id': customer.id})
|
||
payload = {
|
||
'prdsList': json.dumps([{
|
||
'prd_id': product.id, 'prd_count': 2, 'price': 1000,
|
||
'garanty_Esh': '', 'product_name': product.product_name,
|
||
}]),
|
||
'PardakhtHa': json.dumps(payments),
|
||
'FactorType': 'Nasb', 'FactorDesc': '', 'sarResidDate': '',
|
||
}
|
||
return self.client.post(url, payload)
|
||
|
||
def test_factor_creation_enqueues_survey_and_cash_thanks(self):
|
||
customer = self._customer_with_phone()
|
||
response = self._post_factor(customer, {'l_naghd': 2000})
|
||
self.assertIn(response.status_code, (200, 302))
|
||
self.assertTrue(
|
||
SMSQueue.objects.filter(customer=customer, sms_type=SMSType.SURVEY).exists(),
|
||
)
|
||
self.assertTrue(
|
||
SMSQueue.objects.filter(customer=customer,
|
||
sms_type=SMSType.PAYMENT_THANKS).exists(),
|
||
)
|
||
|
||
def test_factor_without_payment_has_survey_but_no_thanks(self):
|
||
customer = self._customer_with_phone()
|
||
self._post_factor(customer, {})
|
||
self.assertTrue(
|
||
SMSQueue.objects.filter(customer=customer, sms_type=SMSType.SURVEY).exists(),
|
||
)
|
||
self.assertFalse(
|
||
SMSQueue.objects.filter(customer=customer,
|
||
sms_type=SMSType.PAYMENT_THANKS).exists(),
|
||
)
|
||
|
||
def test_factor_discount_only_sends_survey_not_thanks(self):
|
||
"""تخفیف، پرداختِ واقعی نیست؛ نباید پیامک تشکر بفرستد (فقط نظرسنجی)."""
|
||
customer = self._customer_with_phone()
|
||
self._post_factor(customer, {'l_takhfif': 2000})
|
||
self.assertTrue(
|
||
SMSQueue.objects.filter(customer=customer, sms_type=SMSType.SURVEY).exists(),
|
||
)
|
||
self.assertFalse(
|
||
SMSQueue.objects.filter(customer=customer,
|
||
sms_type=SMSType.PAYMENT_THANKS).exists(),
|
||
)
|
||
|
||
|
||
# ---------- ثبت اقدام از نوع بازدید ----------
|
||
|
||
def _post_action(self, customer, action_type):
|
||
url = reverse('admin_panel:customerNewAction', kwargs={'c_id': customer.id})
|
||
return self.client.post(url, {
|
||
'actionType': action_type,
|
||
'dateTimeAnjam': '1700000000000',
|
||
'desc': 'توضیح', 'userID': self.user.id,
|
||
})
|
||
|
||
def test_bazdid_action_enqueues_visit_created(self):
|
||
customer = self._customer_with_phone()
|
||
response = self._post_action(customer, 'bazdid')
|
||
self.assertIn(response.status_code, (200, 302))
|
||
item = SMSQueue.objects.filter(customer=customer,
|
||
sms_type=SMSType.VISIT_CREATED)
|
||
self.assertTrue(item.exists())
|
||
# متن شامل تاریخ بازدید است.
|
||
self.assertIn('بازدید', item.first().message)
|
||
|
||
def test_non_bazdid_action_has_no_visit_created(self):
|
||
customer = self._customer_with_phone()
|
||
self._post_action(customer, 'sayer')
|
||
self.assertFalse(
|
||
SMSQueue.objects.filter(sms_type=SMSType.VISIT_CREATED).exists(),
|
||
)
|
||
|
||
# ---------- تشکر از پرداخت (وصول چک، نسیه، سند دستی) ----------
|
||
|
||
def test_pass_check_enqueues_payment_thanks(self):
|
||
customer = self._customer_with_phone()
|
||
sarresid = Sarresid.objects.create(
|
||
customer=customer, price=5000, payStatus='false', type='check',
|
||
)
|
||
url = reverse('admin_panel:passCheck', kwargs={'articl_id': sarresid.id})
|
||
response = self.client.post(url, {'message': 'ok'}, HTTP_REFERER='/panel/')
|
||
self.assertEqual(response.status_code, 200)
|
||
self.assertTrue(
|
||
SMSQueue.objects.filter(customer=customer,
|
||
sms_type=SMSType.PAYMENT_THANKS).exists(),
|
||
)
|
||
|
||
def test_pardakht_nesiye_enqueues_payment_thanks(self):
|
||
customer = self._customer_with_phone()
|
||
sarresid = Sarresid.objects.create(
|
||
customer=customer, price=5000, payStatus='false', type='nesiye',
|
||
)
|
||
url = reverse('admin_panel:pardakhtNesiye', kwargs={'cs_id': customer.id})
|
||
response = self.client.post(url, {
|
||
'next_sarresidDateTime': '1700000000000',
|
||
'next_sarresidPrice': '5000',
|
||
'sr_articlID': sarresid.id,
|
||
'nahvePardakht': 'naghd',
|
||
}, HTTP_REFERER='/panel/')
|
||
self.assertIn(response.status_code, (200, 302))
|
||
self.assertTrue(
|
||
SMSQueue.objects.filter(customer=customer,
|
||
sms_type=SMSType.PAYMENT_THANKS).exists(),
|
||
)
|
||
|
||
def test_pardakht_nesiye_discount_has_no_thanks(self):
|
||
"""پرداخت نسیه با نحوهٔ «تخفیف» پرداختِ واقعی نیست ⇒ بدون تشکر."""
|
||
customer = self._customer_with_phone()
|
||
sarresid = Sarresid.objects.create(
|
||
customer=customer, price=5000, payStatus='false', type='nesiye',
|
||
)
|
||
url = reverse('admin_panel:pardakhtNesiye', kwargs={'cs_id': customer.id})
|
||
response = self.client.post(url, {
|
||
'next_sarresidDateTime': '1700000000000',
|
||
'next_sarresidPrice': '5000',
|
||
'sr_articlID': sarresid.id,
|
||
'nahvePardakht': 'takhfif',
|
||
}, HTTP_REFERER='/panel/')
|
||
self.assertIn(response.status_code, (200, 302))
|
||
self.assertFalse(
|
||
SMSQueue.objects.filter(sms_type=SMSType.PAYMENT_THANKS).exists(),
|
||
)
|
||
|
||
def test_manual_credit_article_enqueues_payment_thanks(self):
|
||
customer = self._customer_with_phone()
|
||
url = reverse('admin_panel:customerHesabdariAdd', kwargs={'c_id': customer.id})
|
||
response = self.client.post(url, {
|
||
'price': '1500', 'type': 'credit', 'typePy': '', 'desc': 'پرداخت',
|
||
'nahvePardakhat': 'نقد',
|
||
}, HTTP_REFERER='/panel/')
|
||
self.assertIn(response.status_code, (200, 302))
|
||
self.assertTrue(
|
||
SMSQueue.objects.filter(customer=customer,
|
||
sms_type=SMSType.PAYMENT_THANKS).exists(),
|
||
)
|
||
|
||
def test_manual_credit_article_discount_has_no_thanks(self):
|
||
"""سند بستانکار با نحوهٔ «تخفیف» نباید پیامک تشکر بفرستد."""
|
||
customer = self._customer_with_phone()
|
||
url = reverse('admin_panel:customerHesabdariAdd', kwargs={'c_id': customer.id})
|
||
self.client.post(url, {
|
||
'price': '1500', 'type': 'credit', 'typePy': '', 'desc': 'تخفیف',
|
||
'nahvePardakhat': 'تخفیف',
|
||
}, HTTP_REFERER='/panel/')
|
||
self.assertFalse(
|
||
SMSQueue.objects.filter(sms_type=SMSType.PAYMENT_THANKS).exists(),
|
||
)
|
||
|
||
def test_manual_debit_article_has_no_thanks(self):
|
||
customer = self._customer_with_phone()
|
||
url = reverse('admin_panel:customerHesabdariAdd', kwargs={'c_id': customer.id})
|
||
self.client.post(url, {
|
||
'price': '1500', 'type': 'debit', 'typePy': '', 'desc': 'بدهی',
|
||
'nahvePardakhat': '',
|
||
}, HTTP_REFERER='/panel/')
|
||
self.assertFalse(
|
||
SMSQueue.objects.filter(sms_type=SMSType.PAYMENT_THANKS).exists(),
|
||
)
|
||
|
||
|
||
class CustomerSmsHistoryTests(PanelTestCase):
|
||
"""تب «سوابق پیامک» در صفحهٔ مشتری: متن دقیق، تاریخ شمسی، وضعیت تحویل، صفحه‌بندی."""
|
||
|
||
username = 'smsHistory'
|
||
|
||
def setUp(self):
|
||
super().setUp()
|
||
self.customer = make_customer('مشتری سابقه')
|
||
PhoneNumber.objects.create(customer=self.customer, phone_number='09120000000')
|
||
|
||
def _item(self, **kwargs):
|
||
defaults = dict(
|
||
customer=self.customer, sms_type=SMSType.WELCOME,
|
||
phone_number='+989120000000', message='متن پیامک دقیق',
|
||
status=SMSQueue.Status.SENT,
|
||
)
|
||
defaults.update(kwargs)
|
||
return SMSQueue.objects.create(**defaults)
|
||
|
||
def _url(self):
|
||
return reverse('admin_panel:customerSmsHistory', kwargs={'c_id': self.customer.id})
|
||
|
||
def test_page_shows_exact_message_text(self):
|
||
self._item(message='کد تخفیف ویژه شما: ABCD-1234')
|
||
html = self.client.get(self._url()).content.decode('utf-8')
|
||
self.assertIn('کد تخفیف ویژه شما: ABCD-1234', html)
|
||
|
||
def test_page_shows_jalali_date(self):
|
||
from admin_panel.templatetags.custom_tags import toJalaliDT
|
||
|
||
item = self._item()
|
||
html = self.client.get(self._url()).content.decode('utf-8')
|
||
self.assertIn(toJalaliDT(item.created_at), html)
|
||
|
||
def test_page_shows_delivery_status(self):
|
||
self._item(delivery_status=SMSQueue.DeliveryStatus.DELIVERED)
|
||
html = self.client.get(self._url()).content.decode('utf-8')
|
||
# برچسب خوانا برای «رسیده به گوشی».
|
||
self.assertIn('رسیده به گوشی', html)
|
||
|
||
def test_page_shows_no_tracking_label_when_delivery_empty(self):
|
||
self._item(delivery_status='')
|
||
html = self.client.get(self._url()).content.decode('utf-8')
|
||
self.assertIn('بدون پیگیری', html)
|
||
|
||
def test_only_this_customer_messages_are_shown(self):
|
||
self._item(message='پیام مشتری الف')
|
||
other = make_customer('مشتری دیگر')
|
||
SMSQueue.objects.create(
|
||
customer=other, sms_type=SMSType.WELCOME,
|
||
phone_number='+989120000009', message='پیام مشتری ب',
|
||
status=SMSQueue.Status.SENT,
|
||
)
|
||
html = self.client.get(self._url()).content.decode('utf-8')
|
||
self.assertIn('پیام مشتری الف', html)
|
||
self.assertNotIn('پیام مشتری ب', html)
|
||
|
||
def test_list_is_paginated(self):
|
||
for i in range(25):
|
||
self._item(message=f'پیام شمارهٔ {i}')
|
||
page1 = self.client.get(self._url()).content.decode('utf-8')
|
||
self.assertIn('صفحهٔ 1 از 2', page1)
|
||
page2 = self.client.get(self._url() + '?page=2').content.decode('utf-8')
|
||
self.assertIn('صفحهٔ 2 از 2', page2)
|
||
|
||
def test_delivery_filter_applies(self):
|
||
self._item(message='تحویل‌شده', delivery_status=SMSQueue.DeliveryStatus.DELIVERED)
|
||
self._item(message='درانتظار', delivery_status=SMSQueue.DeliveryStatus.PENDING)
|
||
html = self.client.get(
|
||
self._url() + '?delivery=' + SMSQueue.DeliveryStatus.DELIVERED,
|
||
).content.decode('utf-8')
|
||
self.assertIn('تحویل‌شده', html)
|
||
self.assertNotIn('درانتظار', html)
|
||
|
||
def test_customer_page_has_sms_history_tab(self):
|
||
html = self.client.get(
|
||
reverse('admin_panel:customer', kwargs={'c_id': self.customer.id}),
|
||
).content.decode('utf-8')
|
||
self.assertIn('سوابق پیامک', html)
|
||
self.assertIn(self._url(), html)
|
||
|
||
def test_tab_hidden_without_manage_sms_permission(self):
|
||
self.client.force_login(make_user('smsNoTab'))
|
||
html = self.client.get(
|
||
reverse('admin_panel:customer', kwargs={'c_id': self.customer.id}),
|
||
).content.decode('utf-8')
|
||
self.assertNotIn(self._url(), html)
|
||
|
||
def test_history_requires_manage_sms_permission(self):
|
||
self.client.force_login(make_user('smsNoHist'))
|
||
self.assertEqual(self.client.get(self._url()).status_code, 403)
|
||
|
||
def test_empty_history_shows_placeholder(self):
|
||
html = self.client.get(self._url()).content.decode('utf-8')
|
||
self.assertIn('پیامکی برای این مشتری ثبت نشده است', html)
|
||
|
||
|
||
class SmsPartsTests(TestCase):
|
||
"""محاسبهٔ «تعداد پیامکِ محاسبه‌شده» (parts) بر اساس کدگذاری GSM-7 / UCS-2."""
|
||
|
||
def test_english_text_is_gsm7(self):
|
||
from admin_panel.sms.parts import detect_encoding, parts_count
|
||
|
||
self.assertEqual(detect_encoding('Hello world 123'), 'gsm7')
|
||
self.assertEqual(parts_count('Hello world'), 1)
|
||
|
||
def test_persian_text_is_ucs2(self):
|
||
from admin_panel.sms.parts import detect_encoding, parts_count
|
||
|
||
self.assertEqual(detect_encoding('سلام مشتری'), 'ucs2')
|
||
# ۷۰ کاراکتر UCS-2 در یک پیامک جا می‌شود.
|
||
self.assertEqual(parts_count('ب' * 70), 1)
|
||
self.assertEqual(parts_count('ب' * 71), 2)
|
||
|
||
def test_mixed_text_is_ucs2(self):
|
||
from admin_panel.sms.parts import detect_encoding
|
||
|
||
# ترکیب فارسی + انگلیسی + عدد ⇒ کل پیامک UCS-2.
|
||
self.assertEqual(detect_encoding('مشتری AB 12'), 'ucs2')
|
||
|
||
def test_gsm7_boundaries(self):
|
||
from admin_panel.sms.parts import parts_count
|
||
|
||
self.assertEqual(parts_count('a' * 160), 1)
|
||
self.assertEqual(parts_count('a' * 161), 2)
|
||
self.assertEqual(parts_count('a' * 306), 2) # دو بخش ۱۵۳تایی
|
||
self.assertEqual(parts_count('a' * 307), 3)
|
||
|
||
def test_gsm7_extended_chars_count_double(self):
|
||
from admin_panel.sms.parts import encoded_length, parts_count
|
||
|
||
# `{` و `}` کاراکترهای الحاقی‌اند: هرکدام ۲ واحد.
|
||
self.assertEqual(encoded_length('{}'), 4)
|
||
self.assertEqual(parts_count('{}'), 1)
|
||
|
||
def test_emoji_counts_as_two_units(self):
|
||
from admin_panel.sms.parts import encoded_length, parts_count
|
||
|
||
self.assertEqual(encoded_length('😀'), 2)
|
||
self.assertEqual(parts_count('😀' * 35), 1)
|
||
self.assertEqual(parts_count('😀' * 36), 2)
|
||
|
||
def test_empty_message_has_no_parts(self):
|
||
from admin_panel.sms.parts import parts_count
|
||
|
||
self.assertEqual(parts_count(''), 0)
|
||
|
||
|
||
class SmsReportTests(PanelTestCase):
|
||
"""گزارش ارسال پیامک: آمار، فیلتر تاریخ شمسی، جستجو، تعداد پیامک، دسترسی."""
|
||
|
||
username = 'smsReport'
|
||
|
||
def setUp(self):
|
||
super().setUp()
|
||
self.customer = make_customer('مشتری گزارش')
|
||
PhoneNumber.objects.create(customer=self.customer, phone_number='09120000000')
|
||
|
||
def _item(self, **kwargs):
|
||
defaults = dict(
|
||
customer=self.customer, sms_type=SMSType.WELCOME,
|
||
phone_number='+989120000000', message='سلام مشتری',
|
||
status=SMSQueue.Status.SENT,
|
||
)
|
||
defaults.update(kwargs)
|
||
return SMSQueue.objects.create(**defaults)
|
||
|
||
def _url(self):
|
||
return reverse('admin_panel:sms_report')
|
||
|
||
def test_report_lists_rows_and_totals(self):
|
||
self._item(message='Hello') # ۱ بخش
|
||
self._item(message='ب' * 200) # UCS-2، چند بخش
|
||
html = self.client.get(self._url()).content.decode('utf-8')
|
||
self.assertIn('Hello', html)
|
||
self.assertIn('مشتری گزارش', html)
|
||
# مجموع پیامک محاسبه‌شده باید نمایش داده شود.
|
||
self.assertIn('مجموع پیامک محاسبه‌شده', html)
|
||
|
||
def test_report_computes_parts_per_message(self):
|
||
from admin_panel.sms.parts import parts_count
|
||
|
||
self._item(message='ب' * 71)
|
||
response = self.client.get(self._url())
|
||
# تعداد بخش محاسبه‌شده در context صفحه هست.
|
||
parts = [row['parts'] for row in response.context['rows']]
|
||
self.assertEqual(parts, [parts_count('ب' * 71)])
|
||
self.assertEqual(parts[0], 2)
|
||
|
||
def test_report_filters_by_jalali_date_range(self):
|
||
from persiantools.jdatetime import JalaliDate
|
||
|
||
self._item(message='قدیمی')
|
||
# بازه‌ای که قطعاً بعد از «الان» است ⇒ نتیجه خالی.
|
||
future = JalaliDate(timezone.localdate() + timedelta(days=5)).strftime('%Y/%m/%d')
|
||
html = self.client.get(self._url() + f'?date_from={future}').content.decode('utf-8')
|
||
self.assertIn('پیامکی با این شرایط یافت نشد', html)
|
||
|
||
def test_report_filters_by_status(self):
|
||
self._item(message='متن-یکتای-موفق', status=SMSQueue.Status.SENT)
|
||
self._item(message='متن-یکتای-ناموفق', status=SMSQueue.Status.FAILED)
|
||
html = self.client.get(self._url() + '?status=failed').content.decode('utf-8')
|
||
self.assertIn('متن-یکتای-ناموفق', html)
|
||
self.assertNotIn('متن-یکتای-موفق', html)
|
||
|
||
def test_report_filters_by_type(self):
|
||
self._item(message='متن-یکتای-خوشامد', sms_type=SMSType.WELCOME)
|
||
self._item(message='متن-یکتای-نظرسنجی', sms_type=SMSType.SURVEY)
|
||
html = self.client.get(self._url() + '?sms_type=survey').content.decode('utf-8')
|
||
self.assertIn('متن-یکتای-نظرسنجی', html)
|
||
self.assertNotIn('متن-یکتای-خوشامد', html)
|
||
|
||
def test_report_search_matches_message_text(self):
|
||
self._item(message='کد تخفیف ABCD')
|
||
html = self.client.get(self._url() + '?q=ABCD').content.decode('utf-8')
|
||
self.assertIn('ABCD', html)
|
||
|
||
def test_report_has_link_to_customer_sms_history(self):
|
||
item = self._item()
|
||
html = self.client.get(self._url()).content.decode('utf-8')
|
||
self.assertIn(
|
||
reverse('admin_panel:customerSmsHistory', kwargs={'c_id': item.customer_id}),
|
||
html,
|
||
)
|
||
|
||
def test_report_paginates(self):
|
||
for i in range(55):
|
||
self._item(message=f'پیامک {i}')
|
||
response = self.client.get(self._url())
|
||
self.assertEqual(response.context['page'].paginator.num_pages, 2)
|
||
|
||
def test_report_requires_manage_sms_permission(self):
|
||
self.client.force_login(make_user('smsRepNoPerm'))
|
||
self.assertEqual(self.client.get(self._url()).status_code, 403)
|
||
|
||
def test_report_excludes_pending_and_cancelled(self):
|
||
"""فقط پیامک‌های تأییدشده/ارسال‌شده باید در گزارش باشند."""
|
||
self._item(message='در-انتظار-تأیید', status=SMSQueue.Status.PENDING)
|
||
self._item(message='لغو-شده', status=SMSQueue.Status.CANCELLED)
|
||
self._item(message='تأیید-شده', status=SMSQueue.Status.APPROVED)
|
||
self._item(message='ارسال-شده', status=SMSQueue.Status.SENT)
|
||
html = self.client.get(self._url()).content.decode('utf-8')
|
||
self.assertNotIn('در-انتظار-تأیید', html)
|
||
self.assertNotIn('لغو-شده', html)
|
||
self.assertIn('تأیید-شده', html)
|
||
self.assertIn('ارسال-شده', html)
|
||
|
||
def test_report_totals_count_only_sent(self):
|
||
self._item(message='در-انتظار', status=SMSQueue.Status.PENDING)
|
||
self._item(message='ارسال-شده', status=SMSQueue.Status.SENT)
|
||
response = self.client.get(self._url())
|
||
self.assertEqual(response.context['total_records'], 1)
|
||
self.assertEqual(response.context['page'].paginator.count, 1)
|
||
|
||
def test_status_dropdown_hides_pending_and_cancelled(self):
|
||
response = self.client.get(self._url())
|
||
codes = [code for code, _label in response.context['status_choices']]
|
||
self.assertNotIn('pending', codes)
|
||
self.assertNotIn('cancelled', codes)
|
||
self.assertIn('sent', codes)
|
||
self.assertIn('approved', codes)
|
||
|
||
|