1819 lines
81 KiB
Python
1819 lines
81 KiB
Python
"""تست‌های قابلیت پیامک: نرمال‌سازی، idempotency، opt-out، worker و تریگرها."""
|
|
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 integrations.sms.base import SMSProvider, SMSResult
|
|
from integrations.sms.normalize import normalize_mobile, to_local
|
|
|
|
from .factories import login_superuser, make_customer, make_user
|
|
|
|
|
|
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_custom_type_always_enabled(self):
|
|
customer = make_customer()
|
|
self.assertTrue(is_sms_enabled(customer, SMSType.CUSTOM))
|
|
|
|
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_due_scan_is_idempotent(self):
|
|
today = timezone.localdate()
|
|
Sarresid.objects.create(
|
|
customer=self.customer, price=5000, payStatus='false', type='check',
|
|
sarresidDT=timezone.now() + timedelta(days=2),
|
|
)
|
|
first = run_all(today)
|
|
second = run_all(today)
|
|
self.assertEqual(first['cheque_due'], 1)
|
|
self.assertEqual(second['cheque_due'], 0)
|
|
self.assertEqual(SMSQueue.objects.filter(sms_type=SMSType.CHEQUE_DUE).count(), 1)
|
|
|
|
def test_cheque_registration_rearms_per_cycle(self):
|
|
Sarresid.objects.create(
|
|
customer=self.customer, price=5000, payStatus='false', type='check',
|
|
sarresidDT=timezone.now() + timedelta(days=30),
|
|
)
|
|
# createdDT = today ⇒ days=0 روی چرخه؛ برای روز چرخهٔ بعد آن را جابه‌جا می‌کنیم.
|
|
sarresid = Sarresid.objects.get()
|
|
Sarresid.objects.filter(pk=sarresid.pk).update(
|
|
createdDT=timezone.now() - timedelta(days=3),
|
|
)
|
|
today = timezone.localdate()
|
|
self.assertEqual(run_all(today)['cheque_registration'], 1)
|
|
self.assertEqual(run_all(today)['cheque_registration'], 0)
|
|
Sarresid.objects.filter(pk=sarresid.pk).update(
|
|
createdDT=timezone.now() - timedelta(days=6),
|
|
)
|
|
self.assertEqual(run_all(today)['cheque_registration'], 1)
|
|
self.assertEqual(
|
|
SMSQueue.objects.filter(sms_type=SMSType.CHEQUE_REGISTRATION).count(), 2,
|
|
)
|
|
|
|
def test_debt_due_requires_positive_balance(self):
|
|
today = timezone.localdate()
|
|
Sarresid.objects.create(
|
|
customer=self.customer, price=5000, payStatus='false', type='nesiye',
|
|
sarresidDT=timezone.now(),
|
|
)
|
|
self.assertEqual(run_all(today)['debt_due'], 0)
|
|
Article.objects.create(customer=self.customer, debit=10000, credit=0)
|
|
run_all(today)
|
|
self.assertEqual(
|
|
SMSQueue.objects.filter(sms_type=SMSType.DEBT_DUE).count(), 1,
|
|
)
|
|
|
|
def test_cheque_registration_stops_once_registered(self):
|
|
"""با ثبت شمارهٔ صیادی، یادآوری «ثبت اطلاعات چک» متوقف می‌شود."""
|
|
sarresid = Sarresid.objects.create(
|
|
customer=self.customer, price=5000, payStatus='false', type='check',
|
|
sarresidDT=timezone.now() + timedelta(days=30),
|
|
)
|
|
Sarresid.objects.filter(pk=sarresid.pk).update(
|
|
createdDT=timezone.now() - timedelta(days=3),
|
|
)
|
|
today = timezone.localdate()
|
|
self.assertEqual(run_all(today)['cheque_registration'], 1)
|
|
|
|
Sarresid.objects.filter(pk=sarresid.pk).update(check_sayyad_id='1234567890123456')
|
|
Sarresid.objects.filter(pk=sarresid.pk).update(
|
|
createdDT=timezone.now() - timedelta(days=6),
|
|
)
|
|
self.assertEqual(run_all(today)['cheque_registration'], 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=2),
|
|
)
|
|
SMSTemplate.objects.filter(sms_type=SMSType.CHEQUE_DUE).update(is_enabled=False)
|
|
self.assertEqual(run_all(today)['cheque_due'], 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_due_cheque(self):
|
|
return Sarresid.objects.create(
|
|
customer=self.customer, price=5000, payStatus='false', type='check',
|
|
sarresidDT=timezone.now() + timedelta(days=1),
|
|
)
|
|
|
|
def test_first_run_sends_nothing_for_existing_records(self):
|
|
# مورد پیش از فعال‌سازی وجود داشته است (خط شروع = همین حالا).
|
|
self._make_due_cheque()
|
|
activate_sms(since=timezone.now())
|
|
self.assertEqual(run_all(self.today)['cheque_due'], 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_due'], 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_due'], 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_due'], 2)
|
|
# اجرای بعدی سهم بعدی را می‌فرستد و هرگز از سقف عبور نمی‌کند.
|
|
self.assertEqual(run_all(self.today)['cheque_due'], 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_due'], 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.CUSTOM,
|
|
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.CUSTOM,
|
|
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.CUSTOM,
|
|
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_DUE)
|
|
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.DEBT_DUE, SMSType.CUSTOM):
|
|
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.SURVEY)
|
|
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.SURVEY)
|
|
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_DUE)
|
|
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_due'))
|
|
|
|
|
|
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}),
|
|
)
|
|
|
|
|
|
class SMSPermissionTests(TestCase):
|
|
"""دسترسی به بخش پیامک فقط با پرمیشن ``admin_panel.ManageSMS``.
|
|
|
|
واگذاری این پرمیشن از Django Admin انجام می‌شود (کاربر → User permissions یا
|
|
از طریق Group). کاربران بدون این پرمیشن — حتی کاربران عادی پنل — نباید به هیچ‌یک
|
|
از مسیرهای پیامک دسترسی داشته باشند.
|
|
"""
|
|
|
|
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.CUSTOM,
|
|
phone_number='+989120000000', message='سلام', status=SMSQueue.Status.PENDING,
|
|
)
|
|
|
|
def _urls(self):
|
|
return [
|
|
('sms_settings', {}, 'get'),
|
|
('sms_templates', {}, 'get'),
|
|
('sms_queue', {}, 'get'),
|
|
('sms_template_edit', {'t_id': self.template.id}, 'post'),
|
|
('sms_customer_toggle', {'c_id': self.phone.customer_id}, 'post'),
|
|
('sms_primary_toggle', {'c_id': self.phone.customer_id}, 'post'),
|
|
('sms_send_custom', {'c_id': self.phone.customer_id}, 'post'),
|
|
('sms_queue_retry', {'q_id': self.queue_item.id}, 'post'),
|
|
('sms_queue_approve', {'q_id': self.queue_item.id}, 'post'),
|
|
('sms_queue_bulk', {}, 'post'),
|
|
('sms_queue_empty', {}, 'post'),
|
|
('sms_queue_delete', {'q_id': self.queue_item.id}, 'post'),
|
|
]
|
|
|
|
def _call(self, name, kwargs, method):
|
|
url = reverse(f'admin_panel:{name}', kwargs=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_')
|
|
}
|
|
self.assertEqual(sms_names, {name for name, _, _ in self._urls()})
|
|
|
|
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_permission_granted_in_django_admin_grants_access(self):
|
|
# همان کاری که Django Admin انجام می‌دهد: افزودن Permission به کاربر.
|
|
self.client.force_login(make_user('smsAllowed', perms=['ManageSMS']))
|
|
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_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'))
|
|
user = make_user('smsGrouped')
|
|
user.groups.add(group)
|
|
self.client.force_login(user)
|
|
self.assertNotEqual(
|
|
self._call('sms_settings', {}, 'get').status_code, 403,
|
|
)
|
|
|
|
def test_navbar_link_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.client.force_login(make_user('smsNavPerm', perms=['ManageSMS']))
|
|
shown = self.client.get(reverse('admin_panel:index'))
|
|
self.assertContains(shown, reverse('admin_panel:sms_settings'))
|
|
|
|
def test_permission_is_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')
|
|
|
|
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}))
|
|
|
|
|
|
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_DUE,
|
|
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_DUE,
|
|
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_due:{customer.id}'
|
|
self.assertEqual(
|
|
enqueue_sms(customer=customer, sms_type=SMSType.CHEQUE_DUE,
|
|
message='یادآوری', dedup_key=key),
|
|
1,
|
|
)
|
|
SMSQueue.objects.filter(dedup_key=key).delete()
|
|
# اجرای دوبارهٔ همان اسکنر: چون کلید در دفتر ثبت شده، صفر ساخته می‌شود.
|
|
self.assertEqual(
|
|
enqueue_sms(customer=customer, sms_type=SMSType.CHEQUE_DUE,
|
|
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=2),
|
|
)
|
|
activate_sms()
|
|
template = SMSTemplate.objects.get(sms_type=SMSType.CHEQUE_DUE)
|
|
apply_timing(template, 2)
|
|
template.save()
|
|
self.assertEqual(run_all(timezone.localdate())['cheque_due'], 1)
|
|
|
|
item = SMSQueue.objects.get(sms_type=SMSType.CHEQUE_DUE)
|
|
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_due'], 0)
|
|
self.assertEqual(SMSQueue.objects.filter(sms_type=SMSType.CHEQUE_DUE).count(), 0)
|
|
|
|
|
|
class QueueStatsTests(PanelTestCase):
|
|
"""صفحهٔ صف، آمار کل و شمارش هر وضعیت را نشان می‌دهد."""
|
|
|
|
username = 'smsStats'
|
|
|
|
def _item(self, status):
|
|
return SMSQueue.objects.create(
|
|
customer=make_customer('مشتری آمار'), sms_type=SMSType.CUSTOM,
|
|
phone_number='+989120000000', message='x', status=status,
|
|
)
|
|
|
|
def test_page_shows_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'], 3)
|
|
self.assertEqual(response.context['pending_count'], 2)
|
|
stats = dict((code, n) for code, _label, n in response.context['status_stats'])
|
|
self.assertEqual(stats['pending'], 2)
|
|
self.assertEqual(stats['sent'], 1)
|
|
self.assertEqual(stats['failed'], 0)
|
|
|
|
def test_filtered_count_reflects_active_filter(self):
|
|
self._item(SMSQueue.Status.PENDING)
|
|
self._item(SMSQueue.Status.SENT)
|
|
response = self.client.get(reverse('admin_panel:sms_queue'), {'status': 'sent'})
|
|
self.assertEqual(response.context['filtered_count'], 1)
|
|
self.assertEqual(response.context['total_count'], 2)
|
|
# آمار هر وضعیت روی کل صف است، نه فیلترشده.
|
|
self.assertEqual(response.context['pending_count'], 1)
|
|
|
|
def test_list_is_paginated(self):
|
|
for _ in range(55):
|
|
self._item(SMSQueue.Status.SENT)
|
|
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.CUSTOM,
|
|
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.SENT)
|
|
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_every_row_regardless_of_status(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.CUSTOM,
|
|
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(), 0)
|
|
|
|
def test_emptied_keys_are_ledgered_so_nothing_returns(self):
|
|
SMSQueue.objects.create(
|
|
customer=make_customer('مشتری'), sms_type=SMSType.CUSTOM,
|
|
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.CUSTOM, 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.CUSTOM,
|
|
phone_number='+989120000000', message='x',
|
|
)
|
|
self.client.get(reverse('admin_panel:sms_queue_empty'))
|
|
self.assertEqual(SMSQueue.objects.count(), 1)
|
|
|
|
|
|
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) # اجرای دوباره تکرار نمی‌کند
|