Files
khatibi/tests/test_sms.py
T
2026-10-27 11:49:09 +03:30

3050 lines
141 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""تست‌های قابلیت پیامک: نرمال‌سازی، idempotency، opt-out، worker و تریگرها."""
import datetime
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, scan_visit
from admin_panel.sms.timing import apply_timing, current_timing_value, timing_for
from admin_panel.sms.triggers import notify_welcome
from admin_panel.visit_policy import visit_period
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):
"""حذفِ یادآوری بازدید فقط همان دوره را کنسل می‌کند؛ دورهٔ بعد دوباره ساخته می‌شود.
نکته: اسکنر با ``now`` تزریقی اجرا می‌شود تا این تست به ساعت واقعی سیستم وابسته نباشد
(قبلاً با ``run_all`` تمام اسکن‌ها اجرا می‌شدند و ماتریس دوره به تاریخ واقعی گره می‌خورد).
"""
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()
self.assertEqual(visit_period().months, 9)
period = visit_period()
install = timezone.localdate() - timedelta(days=300)
today = timezone.localdate()
# روز ورود به چرخهٔ بعد = موعد تقویمی بعدی (install + 9m، +18m، ...)
from dateutil.relativedelta import relativedelta
from admin_panel.visit_policy import visit_cycle_index
next_due = install + period
while next_due <= today:
next_due = next_due + period
days_to_next = (next_due - today).days
self.assertGreater(
visit_cycle_index(install, next_due), visit_cycle_index(install, today),
msg='شمارهٔ چرخه باید در موعد تقویمی بعدی افزایش یابد',
)
def keys():
return list(
SMSQueue.objects.filter(customer=customer).values_list('dedup_key', flat=True)
)
def scan(day):
"""اسکن در «روز» داده‌شده؛ مبنای زمانی هم همان روز است (بدون وابستگی به ساعت واقعی)."""
moment = timezone.make_aware(
datetime.datetime.combine(day, datetime.time(12, 0)),
timezone.get_current_timezone(),
)
return scan_visit(day, now=moment)
# کنترل جریان: پنجرهٔ یادآوری با «امروز» واقعی (زمان پنجرهٔ اسکن) سنجیده می‌شود،
# پس خط شروع را به همان اندازه عقب می‌بریم تا این تست به تاریخ اجرا وابسته نباشد.
config = SMSProviderConfig.load()
config.scan_since = timezone.now() - timedelta(days=600)
config.allow_backlog = True
config.backlog_max = 50
config.save()
# پیش‌بردن «اکنون» به روزهای آینده به تغییر موقت ساعت ارجاعی نیاز دارد
# (بازه‌های ماهانه تقویمی‌اند و با جابه‌جایی خط شروع شبیه‌سازی نمی‌شوند).
import admin_panel.sms.scanner as scanner_module
real_timezone_now = scanner_module.timezone.now
def scan(day):
"""اسکن در «روز» داده‌شده؛ مبنای زمانی هم همان روز است (بدون وابستگی به ساعت واقعی)."""
moment = timezone.make_aware(
datetime.datetime.combine(day, datetime.time(12, 0)),
timezone.get_current_timezone(),
)
scanner_module.timezone.now = lambda: moment
try:
return scan_visit(day, now=moment)
finally:
scanner_module.timezone.now = real_timezone_now
scan(today)
first = keys()
self.assertEqual(len(first), 1)
# مدیر مورد را حذف می‌کند ⇒ در همان دوره دوباره ساخته نمی‌شود.
SMSQueue.objects.filter(customer=customer).delete()
scan(today)
self.assertEqual(keys(), [])
# حتی نزدیک پایان دوره هم برنمی‌گردد.
scan(today + timedelta(days=days_to_next - 1))
self.assertEqual(keys(), [])
# در دورهٔ بعد کلیدِ تازه ساخته می‌شود (یادآوری برای همیشه خاموش نمی‌ماند).
# نکته: شماره‌گذاری چرخه از «لنگر ثابت» (تاریخ نصب) محاسبه می‌شود، پس اولین
# چرخهٔ کاملِ بعد از حذف، چرخهٔ شمارهٔ ۲ است؛ حلقه تا رسیدن به کلید تازه
# ادامه می‌دهد تا تست به محاسبهٔ دقیق تقویمی گره نخورد.
second = None
for step in range(1, 5):
scan(today + timedelta(days=days_to_next * step))
current = keys()
if current:
second = current
break
self.assertIsNotNone(second, msg='یادآوری دورهٔ بعد باید دوباره ساخته شود')
self.assertNotEqual(second, 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_eight_sms_types(self):
self.assertEqual(len(SMSType.choices), 8)
self.assertEqual(
{code for code, _label in SMSType.choices},
{'welcome', 'survey', 'cheque_registration', 'visit',
'visit_created', 'install_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 test_install_template_is_seeded_with_auto_send_off(self):
"""قالب «ثبت نصب» فعال است ولی «ارسال خودکار» آن خاموش (نیازمند تأیید)."""
template = SMSTemplate.objects.get(sms_type=SMSType.INSTALL_CREATED)
self.assertTrue(template.is_enabled)
self.assertFalse(template.auto_send)
self.assertEqual(template.offset_days, 0)
self.assertIn('{name}', template.text)
self.assertIn('{date}', template.text)
# ---------- نظرسنجی + تشکر از پرداخت نقدی روی فاکتور ----------
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, *, is_done=True):
url = reverse('admin_panel:customerNewAction', kwargs={'c_id': customer.id})
payload = {
'actionType': action_type,
'dateTimeAnjam': '1700000000000',
'desc': 'توضیح', 'userID': self.user.id,
}
if is_done:
# اعتبارسنجی HTTP: بازدید انجام‌شده باید تاریخ اتمام داشته باشد؛
# فیلد خالی در سرور با زمان ثبت پر می‌شود.
payload['is_done'] = 'on'
payload['FinishDateTime'] = '1700000000000'
return self.client.post(url, payload)
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 _visit_sms(self, customer):
"""همهٔ ردیف‌های «ثبت بازدید» (هر وضعیتی) — برای شمارش کل."""
return SMSQueue.objects.filter(customer=customer,
sms_type=SMSType.VISIT_CREATED)
def _pending_visit_sms(self, customer):
"""فقط ردیف‌هایی که worker ممکن است بفرستد (در انتظار/تأییدشده)."""
return self._visit_sms(customer).filter(
status__in=(SMSQueue.Status.PENDING, SMSQueue.Status.APPROVED),
)
def _edit_action(self, customer, action_type, *, is_done=True):
"""ویرایش اولین اقدام مشتری با نوع جدید (از مسیر واقعی ویو)."""
from admin_panel.models import Action
action = Action.objects.filter(customer=customer).order_by('id').first()
url = reverse('admin_panel:customerEditAction',
kwargs={'c_id': customer.id, 'action_id': action.id})
payload = {
'actionType': action_type,
'dateTimeAnjam': '1700000000000',
'desc': 'ویرایش', 'userID': self.user.id,
'FinishDateTime': '1700000000000',
}
if is_done:
payload['is_done'] = 'on'
response = self.client.post(
url, payload,
HTTP_REFERER=reverse('admin_panel:customerAction',
kwargs={'c_id': customer.id}),
)
action.refresh_from_db()
return response, action
def test_editing_coordination_to_bazdid_enqueues_visit_created(self):
"""هماهنگی → بازدید در ویرایش ⇒ پیامک «ثبت بازدید» باید ساخته شود."""
customer = self._customer_with_phone()
self._post_action(customer, 'hamahangi')
self.assertEqual(self._visit_sms(customer).count(), 0)
response, action = self._edit_action(customer, 'bazdid')
self.assertEqual(response.status_code, 302)
self.assertEqual(action.actionType, 'bazdid')
self.assertEqual(self._visit_sms(customer).count(), 1)
# کلید یکتاسازی همان رکورد اقدام است (پس ویرایش دوباره تکرار نمی‌کند).
self.assertEqual(self._visit_sms(customer).first().dedup_key,
'visit_created:%d' % action.id)
def test_editing_to_bazdid_twice_does_not_duplicate_sms(self):
"""ویرایش دوبارهٔ همان رکورد نباید پیامک دوم بسازد (dedup روی action)."""
customer = self._customer_with_phone()
self._post_action(customer, 'hamahangi')
self._edit_action(customer, 'bazdid')
# برگشت به هماهنگی و تبدیل دوباره به بازدید: پیامک معلق لغو شده، ولی
# کلید دفتر dedup خورده است ⇒ پیامک تازه ساخته نمی‌شود.
self._edit_action(customer, 'hamahangi')
self._edit_action(customer, 'bazdid')
self.assertEqual(self._visit_sms(customer).count(), 1)
def test_editing_bazdid_to_other_type_cancels_pending_sms(self):
"""بازدید → هماهنگی در ویرایش ⇒ پیامک معلق «ثبت بازدید» لغو می‌شود."""
customer = self._customer_with_phone()
self._post_action(customer, 'bazdid')
self.assertEqual(self._pending_visit_sms(customer).count(), 1)
self._edit_action(customer, 'hamahangi')
# ردیف حذف نمی‌شود (سابقه می‌ماند) ولی دیگر ارسال‌شدنی نیست.
self.assertEqual(self._pending_visit_sms(customer).count(), 0)
self.assertEqual(
self._visit_sms(customer).first().status, SMSQueue.Status.CANCELLED,
)
def test_editing_other_fields_of_bazdid_sends_no_new_sms(self):
"""ویرایش عادی یک اقدام بازدید (بدون تغییر نوع) پیامک تازه نمی‌فرستد."""
customer = self._customer_with_phone()
self._post_action(customer, 'bazdid')
self.assertEqual(self._visit_sms(customer).count(), 1)
self._edit_action(customer, 'bazdid')
self.assertEqual(self._visit_sms(customer).count(), 1)
def test_editing_bazdid_keeps_other_actions_pending_sms(self):
"""لغو پیامک معلق، فقط رکورد ویرایش‌شده را هدف می‌گیرد (نه بقیهٔ بازدیدها)."""
from admin_panel.models import Action
from admin_panel.sms.triggers import notify_visit_scheduled
customer = self._customer_with_phone()
self._post_action(customer, 'bazdid') # رکورد ۱ + پیامک
first = Action.objects.filter(customer=customer).order_by('id').first()
second = Action.objects.create(
customer=customer, actionType='bazdid', status='finish',
actionDateTime=timezone.now(), finishDateTime=timezone.now(),
)
notify_visit_scheduled(customer, second, created_by=self.user) # پیامک رکورد ۲
self.assertEqual(self._pending_visit_sms(customer).count(), 2)
# رکورد دوم را از بازدید به هماهنگی تغییر می‌دهیم ⇒ فقط پیامک رکورد ۲ لغو شود.
url = reverse('admin_panel:customerEditAction',
kwargs={'c_id': customer.id, 'action_id': second.id})
self.client.post(url, {
'actionType': 'hamahangi', 'dateTimeAnjam': '1700000000000',
'FinishDateTime': '1700000000000', 'desc': '', 'userID': self.user.id,
'is_done': 'on',
}, HTTP_REFERER=reverse('admin_panel:customerAction',
kwargs={'c_id': customer.id}))
keys = set(self._pending_visit_sms(customer).values_list('dedup_key', flat=True))
self.assertEqual(keys, {'visit_created:%d' % first.id})
# ---------- ثبت اقدام از نوع «نصب» ----------
def _install_sms(self, customer):
"""همهٔ ردیف‌های «ثبت نصب» (هر وضعیتی)."""
return SMSQueue.objects.filter(customer=customer,
sms_type=SMSType.INSTALL_CREATED)
def _pending_install_sms(self, customer):
return self._install_sms(customer).filter(
status__in=(SMSQueue.Status.PENDING, SMSQueue.Status.APPROVED),
)
def test_nasb_action_enqueues_install_created(self):
"""اقدام «نصب» ⇒ پیامک «ثبت نصب» از قالب install_created."""
customer = self._customer_with_phone()
response = self._post_action(customer, 'nasb')
self.assertIn(response.status_code, (200, 302))
item = self._install_sms(customer)
self.assertEqual(item.count(), 1)
self.assertIn('نصب', item.first().message)
# auto_send قالب خاموش است ⇒ ردیف «در انتظار تأیید» می‌ماند، نه ارسال فوری.
self.assertEqual(item.first().status, SMSQueue.Status.PENDING)
# پیامک بازدید ساخته نمی‌شود (دو نوع مستقل).
self.assertEqual(self._pending_visit_sms(customer).count(), 0)
def test_nasb_action_sms_has_its_own_dedup_key(self):
"""کلید یکتاسازی با پیشوند نوع خودش است (با visit_created قاطی نمی‌شود)."""
from admin_panel.models import Action
customer = self._customer_with_phone()
self._post_action(customer, 'nasb')
action = Action.objects.filter(customer=customer).order_by('id').first()
self.assertEqual(self._install_sms(customer).first().dedup_key,
'install_created:%d' % action.id)
def test_non_nasb_action_has_no_install_created(self):
customer = self._customer_with_phone()
self._post_action(customer, 'bazdid')
self.assertFalse(self._install_sms(customer).exists())
def test_editing_to_nasb_enqueues_install_created(self):
"""هماهنگی → نصب در ویرایش ⇒ پیامک «ثبت نصب» ساخته می‌شود."""
customer = self._customer_with_phone()
self._post_action(customer, 'hamahangi')
self.assertEqual(self._install_sms(customer).count(), 0)
response, action = self._edit_action(customer, 'nasb')
self.assertEqual(response.status_code, 302)
self.assertEqual(action.actionType, 'nasb')
self.assertEqual(self._pending_install_sms(customer).count(), 1)
def test_editing_nasb_to_other_type_cancels_pending_install_sms(self):
"""نصب → هماهنگی در ویرایش ⇒ پیامک معلق «ثبت نصب» لغو می‌شود."""
customer = self._customer_with_phone()
self._post_action(customer, 'nasb')
self.assertEqual(self._pending_install_sms(customer).count(), 1)
self._edit_action(customer, 'hamahangi')
self.assertEqual(self._pending_install_sms(customer).count(), 0)
self.assertEqual(
self._install_sms(customer).first().status, SMSQueue.Status.CANCELLED,
)
def test_editing_bazdid_to_nasb_switches_sms_type(self):
"""بازدید → نصب ⇒ پیامک بازدید لغو و پیامک نصب ساخته می‌شود (نه هر دو)."""
customer = self._customer_with_phone()
self._post_action(customer, 'bazdid')
self.assertEqual(self._pending_visit_sms(customer).count(), 1)
self._edit_action(customer, 'nasb')
self.assertEqual(self._pending_visit_sms(customer).count(), 0)
self.assertEqual(self._pending_install_sms(customer).count(), 1)
def test_install_template_can_be_disabled_from_templates_page(self):
"""تنظیمات «ثبت نصب» از صفحهٔ قالب‌ها: غیرفعال‌کردن ⇒ پیامکی ساخته نمی‌شود."""
template = SMSTemplate.objects.get(sms_type=SMSType.INSTALL_CREATED)
response = self.client.post(
reverse('admin_panel:sms_template_edit', kwargs={'t_id': template.id}),
{'text': template.text, 'title': template.title},
)
self.assertEqual(response.status_code, 302)
template.refresh_from_db()
self.assertFalse(template.is_enabled)
customer = self._customer_with_phone()
self._post_action(customer, 'nasb')
self.assertFalse(self._install_sms(customer).exists())
def test_install_template_text_is_editable_from_templates_page(self):
"""ویرایش متن قالب از پنل ⇒ همان متن در پیامک ساخته‌شده می‌آید."""
template = SMSTemplate.objects.get(sms_type=SMSType.INSTALL_CREATED)
response = self.client.post(
reverse('admin_panel:sms_template_edit', kwargs={'t_id': template.id}),
{'text': '{name} عزیز، نصب دستگاه انجام شد ({date}).',
'title': 'ثبت نصب', 'is_enabled': 'on'},
)
self.assertEqual(response.status_code, 302)
customer = self._customer_with_phone()
self._post_action(customer, 'nasb')
message = self._install_sms(customer).first().message
self.assertIn('نصب دستگاه انجام شد', message)
self.assertIn(customer.name_family, message)
def test_install_sms_disabled_per_customer_is_not_created(self):
"""opt-out مشتری برای نوع install_created رعایت می‌شود."""
from admin_panel.models import CustomerSMSSetting
customer = self._customer_with_phone()
CustomerSMSSetting.objects.create(
customer=customer, sms_type=SMSType.INSTALL_CREATED, is_enabled=False,
)
self._post_action(customer, 'nasb')
self.assertFalse(self._install_sms(customer).exists())
def test_install_template_can_be_enabled_for_auto_send(self):
"""اگر مدیر «ارسال خودکار» را روشن کند، ردیف مستقیماً تأییدشده ساخته می‌شود."""
SMSTemplate.objects.filter(sms_type=SMSType.INSTALL_CREATED).update(
auto_send=True,
)
customer = self._customer_with_phone()
self._post_action(customer, 'nasb')
self.assertEqual(
self._install_sms(customer).first().status, SMSQueue.Status.APPROVED,
)
# ---------- تشکر از پرداخت (وصول چک، نسیه، سند دستی) ----------
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 InstallTemplateMigrationTests(TestCase):
"""مهاجرت ۰۰۸۱: قالب‌های موجود دست‌نخورده می‌مانند، «ثبت نصب» بدون ارسال خودکار.
توابع مهاجرت مستقیم و با همان ``apps`` واقعی صدا زده می‌شوند (بدون اجرای
schema)، تا در تراکنش تست انجام و بازگردانی شود.
"""
MODULE = 'admin_panel.migrations.0081_sms_type_install_created'
FIELDS = ('sms_type', 'title', 'text', 'is_enabled', 'auto_send',
'offset_days', 'repeat_interval_days', 'repeat_interval_months')
class _Apps:
"""جایگزین ``schema_editor.apps``: نگاشت به مدل‌های واقعی پروژه."""
@staticmethod
def get_model(app_label, model_name):
from django.apps import apps as real_apps
return real_apps.get_model(app_label, model_name)
def _migration(self):
from importlib import import_module
return import_module(self.MODULE)
def _snapshot(self):
return {row['sms_type']: row
for row in SMSTemplate.objects.values(*self.FIELDS)}
def _seed(self):
self._migration().seed_install_template(self._Apps, None)
def test_seed_leaves_existing_templates_untouched(self):
SMSTemplate.objects.filter(sms_type=SMSType.INSTALL_CREATED).delete()
# مقادیر غیرپیش‌فرض؛ اگر مهاجرت هر فیلدی را بازنویسی کند دیده می‌شود.
(SMSTemplate.objects.filter(sms_type=SMSType.VISIT).update(
title='عنوان دستی', text='متن دستی {name}', is_enabled=False,
auto_send=True, offset_days=3, repeat_interval_months=6,
))
before = self._snapshot()
self._seed()
after = self._snapshot()
for sms_type, row in before.items():
self.assertEqual(after[sms_type], row, msg=sms_type)
self.assertIn(SMSType.INSTALL_CREATED, after)
def test_seeded_install_template_has_auto_send_off(self):
SMSTemplate.objects.filter(sms_type=SMSType.INSTALL_CREATED).delete()
self._seed()
template = SMSTemplate.objects.get(sms_type=SMSType.INSTALL_CREATED)
self.assertTrue(template.is_enabled, msg='باید در پنل قالب‌ها دیده شود')
self.assertFalse(template.auto_send, msg='نباید بدون تأیید مدیر ارسال شود')
self.assertEqual(template.offset_days, 0)
self.assertIsNone(template.repeat_interval_days)
self.assertIsNone(template.repeat_interval_months)
def test_seed_is_idempotent_for_an_existing_install_template(self):
"""اگر قالب از قبل وجود داشته باشد، متن/تنظیماتش بازنویسی نمی‌شود."""
template = SMSTemplate.objects.get(sms_type=SMSType.INSTALL_CREATED)
SMSTemplate.objects.filter(pk=template.pk).update(
text='متن ویرایش‌شدهٔ مدیر {name}', auto_send=True, offset_days=5,
)
before = self._snapshot()
self._seed()
self.assertEqual(self._snapshot(), before)
def test_reverse_only_removes_the_new_type(self):
before = self._snapshot()
self._migration().drop_install_template(self._Apps, None)
after = self._snapshot()
self.assertNotIn(SMSType.INSTALL_CREATED, after)
self.assertEqual(set(after), set(before) - {SMSType.INSTALL_CREATED})
for sms_type, row in after.items():
self.assertEqual(row, before[sms_type], msg=sms_type)
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)
# ---------- انتقال تنظیمات پیامک از صفحهٔ فاکتور به این صفحه ----------
def test_factor_page_no_longer_contains_sms_settings_box(self):
"""باکس «پیامک» از صفحهٔ ایجاد فاکتور حذف شده (تنظیمات به تب پیامک رفت)."""
html = self.client.get(
reverse('admin_panel:customerNew', kwargs={'c_id': self.customer.id}),
).content.decode('utf-8')
self.assertNotIn(
reverse('admin_panel:sms_send_custom', kwargs={'c_id': self.customer.id}),
html,
)
self.assertNotIn(
reverse('admin_panel:sms_customer_toggle', kwargs={'c_id': self.customer.id}),
html,
)
def test_sms_page_holds_all_settings_controls(self):
"""همهٔ کنترل‌های تنظیمات روی صفحهٔ پیامک هستند: نوع‌ها، شمارهٔ اصلی، ارسال."""
html = self.client.get(self._url()).content.decode('utf-8')
self.assertIn(
reverse('admin_panel:sms_customer_toggle', kwargs={'c_id': self.customer.id}),
html,
)
self.assertIn(
reverse('admin_panel:sms_primary_toggle', kwargs={'c_id': self.customer.id}),
html,
)
self.assertIn(
reverse('admin_panel:sms_send_custom', kwargs={'c_id': self.customer.id}),
html,
)
# یک ردیف برای هر نوع پیامک (۸ نوع) و یک کلید for each.
self.assertEqual(html.count('name="sms_type"'), len(SMSType.choices))
self.assertEqual(html.count('role="switch"'), len(SMSType.choices))
def test_settings_live_in_modals_not_in_history_flow(self):
"""تنظیمات داخل مودال است تا فهرست سوابق شلوغ نشود."""
html = self.client.get(self._url()).content.decode('utf-8')
self.assertIn('id="smsSettingsModal"', html)
self.assertIn('id="smsCustomModal"', html)
self.assertIn('js-csmsx-open', html)
def test_stats_are_computed_for_the_header(self):
self._item(status=SMSQueue.Status.SENT)
self._item(status=SMSQueue.Status.PENDING)
self._item(status=SMSQueue.Status.FAILED)
response = self.client.get(self._url())
self.assertEqual(response.context['stat_sent'], 1)
self.assertEqual(response.context['stat_pending'], 1)
self.assertEqual(response.context['stat_failed'], 1)
self.assertEqual(response.context['total_count'], 3)
def test_toggle_redirects_back_to_sms_page_with_open_settings(self):
"""پس از تغییر تنظیم، کاربر در همان صفحه/مودال می‌ماند (نه صفحهٔ مشتری)."""
response = self.client.post(
reverse('admin_panel:sms_customer_toggle', kwargs={'c_id': self.customer.id}),
{'sms_type': SMSType.WELCOME, 'is_enabled': 'false'},
)
self.assertEqual(response.status_code, 302)
self.assertTrue(response.url.startswith(self._url()), response.url)
self.assertIn('open=settings', response.url)
self.assertIn('notice=customer_sms_saved', response.url)
def test_custom_send_redirects_back_to_sms_page(self):
mark_primary(self.customer, '09120000000')
response = self.client.post(
reverse('admin_panel:sms_send_custom', kwargs={'c_id': self.customer.id}),
{'text': 'سلام {name}'},
)
self.assertEqual(response.status_code, 302)
self.assertTrue(response.url.startswith(self._url()), response.url)
self.assertIn('notice=custom_queued', response.url)
def test_primary_toggle_redirects_back_to_sms_page(self):
response = self.client.post(
reverse('admin_panel:sms_primary_toggle', kwargs={'c_id': self.customer.id}),
{'phone': '09120000000', 'is_primary': 'true'},
)
self.assertEqual(response.status_code, 302)
self.assertTrue(response.url.startswith(self._url()), response.url)
def test_open_settings_query_keeps_modal_available(self):
response = self.client.get(self._url() + '?open=settings')
self.assertEqual(response.status_code, 200)
self.assertIn('id="smsSettingsModal"', response.content.decode('utf-8'))
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)