384 lines
17 KiB
Python
384 lines
17 KiB
Python
"""تست‌های دستور ``normalize_customer_sms_phones``.
|
||
|
||
همهٔ تست‌ها روی دیتابیس تستی Django اجرا می‌شوند؛ نه به ``data9/kt.sqlite3`` دست
|
||
می‌زنند و نه فایل وضعیت داخل ریشهٔ پروژه می‌نویسند (مسیر وضعیت موقت است).
|
||
"""
|
||
import io
|
||
import json
|
||
import shutil
|
||
import tempfile
|
||
from pathlib import Path
|
||
from unittest import mock
|
||
|
||
from django.core.management import call_command
|
||
from django.core.management.base import CommandError
|
||
from django.db.models.query import QuerySet
|
||
from django.test import TestCase
|
||
|
||
from admin_panel.models import Customer, PhoneNumber
|
||
from admin_panel.management.commands.normalize_customer_sms_phones import (
|
||
FIELD_MAX_LENGTH, digit_string, is_landline)
|
||
from admin_panel.sms.recipients import resolve_recipients
|
||
|
||
from .factories import make_customer
|
||
|
||
|
||
def make_phone(customer, number, title='تلفن'):
|
||
return PhoneNumber.objects.create(
|
||
customer=customer, title=title, phone_number=number)
|
||
|
||
|
||
class PureHelperTests(TestCase):
|
||
"""توابع کمکی اعتبارسنجی/تفکیک (بدون دیتابیس واقعی)."""
|
||
|
||
def test_digit_string_converts_persian_and_arabic_digits(self):
|
||
self.assertEqual(digit_string('۰۹۱۲-۳۴۵'), '0912345')
|
||
self.assertEqual(digit_string('٠٩١٢ ٣٤٥'), '0912345')
|
||
self.assertEqual(digit_string('(0912) 345'), '0912345')
|
||
self.assertEqual(digit_string(None), '')
|
||
self.assertEqual(digit_string(' '), '')
|
||
|
||
def test_landline_detection(self):
|
||
for raw in ('02112345678', '۰۲۱۱۲۳۴۵۶۷۸', '031-33445566', '05138490000',
|
||
'57387608', '57473005'): # ۸رقمی = ثابت محلی
|
||
self.assertTrue(is_landline(raw), msg=raw)
|
||
for raw in ('09123456789', '9123456789', '12345', '', None,
|
||
'+989123456789', '0012345678'):
|
||
self.assertFalse(is_landline(raw), msg=repr(raw))
|
||
|
||
|
||
class CommandTestCase(TestCase):
|
||
"""زیرساخت مشترک: دیتابیس تستی + فایل وضعیت موقت."""
|
||
|
||
def setUp(self):
|
||
super().setUp()
|
||
self.tmpdir = tempfile.mkdtemp(prefix='smsnorm-')
|
||
self.addCleanup(shutil.rmtree, self.tmpdir, ignore_errors=True)
|
||
self.checkpoint = Path(self.tmpdir) / '.sms_normalization_checkpoint.json'
|
||
|
||
def run_cmd(self, **options):
|
||
options.setdefault('checkpoint', str(self.checkpoint))
|
||
options.setdefault('stdout', io.StringIO())
|
||
call_command('normalize_customer_sms_phones', **options)
|
||
return options['stdout'].getvalue()
|
||
|
||
def stored(self):
|
||
return dict(Customer.objects.values_list('id', 'sms_primary_phones'))
|
||
|
||
|
||
class DryRunTests(CommandTestCase):
|
||
"""پیش‌فرض: هیچ دیتایی در دیتابیس یا فایل وضعیت نوشته نمی‌شود."""
|
||
|
||
def test_dry_run_writes_nothing(self):
|
||
customer = make_customer('خشکی')
|
||
Customer.objects.filter(pk=customer.pk).update(
|
||
phone1='09120000001', sms_primary_phones='+989120000009')
|
||
make_phone(customer, '09120000002')
|
||
|
||
out = self.run_cmd()
|
||
customer.refresh_from_db()
|
||
self.assertEqual(customer.sms_primary_phones, '+989120000009')
|
||
self.assertFalse(self.checkpoint.exists())
|
||
self.assertIn('dry-run', out)
|
||
self.assertIn('قابل‌به‌روزرسانی', out)
|
||
|
||
def test_dry_run_reports_counts(self):
|
||
customer = make_customer('گزارش')
|
||
Customer.objects.filter(pk=customer.pk).update(phone1='09120000001')
|
||
PhoneNumber.objects.create(
|
||
customer=customer, title='t', phone_number='02112345678')
|
||
out = self.run_cmd()
|
||
self.assertIn('مشتریان بررسی‌شده: 1', out)
|
||
self.assertIn('شماره‌های فیلترشده: ثابت=1', out)
|
||
def test_dry_run_flag_conflicts_with_apply(self):
|
||
customer = make_customer('تناقض فلگ')
|
||
with self.assertRaises(CommandError):
|
||
self.run_cmd(dry_run=True, apply=True)
|
||
customer.refresh_from_db()
|
||
self.assertEqual(customer.sms_primary_phones, '')
|
||
|
||
def test_dry_run_ignores_reset_checkpoint(self):
|
||
self.checkpoint.write_text(json.dumps({'last_id': 5}), encoding='utf-8')
|
||
self.run_cmd(reset_checkpoint=True)
|
||
self.assertTrue(self.checkpoint.exists())
|
||
|
||
|
||
class ApplyTests(CommandTestCase):
|
||
"""اجرای واقعی: تجمیع، حذف تکراری، فیلتر نامعتبر، ایمنی داده."""
|
||
|
||
def test_aggregates_all_sources_in_priority_order(self):
|
||
customer = make_customer('چند منبع')
|
||
Customer.objects.filter(pk=customer.pk).update(
|
||
phone1='09120000001', phone2='09120000002', phone3='09120000003')
|
||
PhoneNumber.objects.create(customer=customer, title='a', phone_number='09120000004')
|
||
PhoneNumber.objects.create(customer=customer, title='b', phone_number='09120000005')
|
||
|
||
self.run_cmd(apply=True)
|
||
customer.refresh_from_db()
|
||
self.assertEqual(
|
||
customer.sms_primary_phones,
|
||
'+989120000001,+989120000002,+989120000003,+989120000004,+989120000005',
|
||
)
|
||
|
||
def test_dedupes_preserving_priority_order(self):
|
||
customer = make_customer('تکراری')
|
||
Customer.objects.filter(pk=customer.pk).update(
|
||
phone1='09120000001', phone2='989120000001') # همان شماره
|
||
PhoneNumber.objects.create(
|
||
customer=customer, title='a', phone_number='+98 912 000 0002')
|
||
PhoneNumber.objects.create(
|
||
customer=customer, title='b', phone_number='۰۹۱۲۰۰۰۰۰۰۲') # تکراری
|
||
|
||
out = self.run_cmd(apply=True)
|
||
customer.refresh_from_db()
|
||
self.assertEqual(customer.sms_primary_phones,
|
||
'+989120000001,+989120000002')
|
||
self.assertIn('شماره‌های تکراری حذف‌شده: 2', out)
|
||
|
||
def test_converts_international_and_local_variants(self):
|
||
customer = make_customer('قالب‌ها')
|
||
Customer.objects.filter(pk=customer.pk).update(phone1='+989120000001')
|
||
PhoneNumber.objects.create(customer=customer, title='a', phone_number='00989120000002')
|
||
PhoneNumber.objects.create(customer=customer, title='b', phone_number='989120000003')
|
||
PhoneNumber.objects.create(customer=customer, title='c', phone_number='9120000004')
|
||
PhoneNumber.objects.create(customer=customer, title='d', phone_number='0912-000-0005')
|
||
|
||
self.run_cmd(apply=True)
|
||
customer.refresh_from_db()
|
||
self.assertEqual(
|
||
customer.sms_primary_phones,
|
||
'+989120000001,+989120000002,+989120000003,+989120000004,+989120000005',
|
||
)
|
||
|
||
def test_filters_landlines_and_invalid_values(self):
|
||
customer = make_customer('نامعتبر')
|
||
Customer.objects.filter(pk=customer.pk).update(
|
||
phone1='09120000001', phone2='02112345678', phone3='0')
|
||
PhoneNumber.objects.create(customer=customer, title='a', phone_number='57387608')
|
||
PhoneNumber.objects.create(customer=customer, title='b', phone_number='اعظمی')
|
||
PhoneNumber.objects.create(customer=customer, title='c', phone_number='')
|
||
|
||
out = self.run_cmd(apply=True)
|
||
customer.refresh_from_db()
|
||
self.assertEqual(customer.sms_primary_phones, '+989120000001')
|
||
self.assertIn('ثابت=2', out) # 02112345678 + 57387608
|
||
self.assertIn('نامعتبر=2', out) # '0' + 'اعظمی' (ردیف خالی شمرده نمی‌شود)
|
||
|
||
def test_source_fields_and_phone_rows_are_untouched(self):
|
||
customer = make_customer('ایزوله', address='تهران')
|
||
Customer.objects.filter(pk=customer.pk).update(
|
||
phone1='09120000001', phone2='02112345678')
|
||
row = make_phone(customer, '09120000002', title='موبایل')
|
||
|
||
self.run_cmd(apply=True)
|
||
customer.refresh_from_db()
|
||
row.refresh_from_db()
|
||
self.assertEqual(customer.phone1, '09120000001')
|
||
self.assertEqual(customer.phone2, '02112345678')
|
||
self.assertEqual(customer.name_family, 'ایزوله')
|
||
self.assertEqual(customer.address, 'تهران')
|
||
self.assertEqual(row.phone_number, '09120000002')
|
||
self.assertEqual(row.title, 'موبایل')
|
||
self.assertEqual(PhoneNumber.objects.filter(customer=customer).count(), 1)
|
||
|
||
def test_unchanged_rows_are_not_rewritten(self):
|
||
customer = make_customer('بدون تغییر')
|
||
Customer.objects.filter(pk=customer.pk).update(
|
||
phone1='09120000001', sms_primary_phones='+989120000001')
|
||
|
||
writes = []
|
||
original = QuerySet.bulk_update
|
||
|
||
def spy(self, objs, fields, batch_size=None):
|
||
writes.append(len(objs))
|
||
return original(self, objs, fields, batch_size)
|
||
|
||
with mock.patch.object(QuerySet, 'bulk_update', spy):
|
||
out = self.run_cmd(apply=True)
|
||
self.assertEqual(writes, []) # هیچ نوشتنی رخ نداد
|
||
customer.refresh_from_db()
|
||
self.assertEqual(customer.sms_primary_phones, '+989120000001')
|
||
self.assertIn('بدون تغییر: 1', out)
|
||
|
||
def test_empty_result_never_erases_existing_value(self):
|
||
customer = make_customer('محافظت')
|
||
Customer.objects.filter(pk=customer.pk).update(
|
||
phone1='02112345678', sms_primary_phones='+989120000009')
|
||
|
||
out = self.run_cmd(apply=True)
|
||
customer.refresh_from_db()
|
||
self.assertEqual(customer.sms_primary_phones, '+989120000009')
|
||
self.assertIn('محافظت شد', out)
|
||
|
||
def test_idempotent_second_run_updates_nothing(self):
|
||
customer = make_customer('تکرار اجرا')
|
||
Customer.objects.filter(pk=customer.pk).update(phone1='09120000001')
|
||
|
||
self.run_cmd(apply=True)
|
||
first = self.stored()
|
||
out = self.run_cmd(apply=True, reset_checkpoint=True)
|
||
self.assertEqual(self.stored(), first)
|
||
self.assertIn('مشتریان به‌روزشده: 0', out)
|
||
|
||
|
||
|
||
class OverflowTests(CommandTestCase):
|
||
"""سرریز طول: truncate نمی‌شود، رکورد دست‌نخورده می‌ماند و در گزارش می‌آید."""
|
||
|
||
def test_overflow_leaves_customer_untouched(self):
|
||
numbers = ['09120000%03d' % i for i in range(1, 11)] # 10×13 = 139 > 100
|
||
customer = make_customer('سرریز')
|
||
Customer.objects.filter(pk=customer.pk).update(
|
||
phone1=numbers[0], phone2=numbers[1])
|
||
for number in numbers[2:]:
|
||
PhoneNumber.objects.create(customer=customer, title='t', phone_number=number)
|
||
|
||
out = self.run_cmd(apply=True)
|
||
customer.refresh_from_db()
|
||
self.assertEqual(customer.sms_primary_phones, '')
|
||
self.assertIn(f'سرریز طول (>{FIELD_MAX_LENGTH} کاراکتر، دست‌نخورده): 1', out)
|
||
self.assertIn(f'شناسه‌های سرریز: {customer.id}', out)
|
||
|
||
def test_exactly_under_max_length_is_saved(self):
|
||
# 7 شماره × 13 کاراکتر + 6 کاما = 97 ≤ 100
|
||
numbers = ['09120000%03d' % i for i in range(1, 8)]
|
||
customer = make_customer('لبه طول')
|
||
Customer.objects.filter(pk=customer.pk).update(phone1=numbers[0])
|
||
for number in numbers[1:]:
|
||
PhoneNumber.objects.create(customer=customer, title='t', phone_number=number)
|
||
|
||
out = self.run_cmd(apply=True)
|
||
customer.refresh_from_db()
|
||
self.assertEqual(len(customer.sms_primary_phones), 97)
|
||
self.assertEqual(customer.sms_primary_phones.count(','), 6)
|
||
self.assertIn('سرریز طول', out)
|
||
|
||
|
||
class BatchAndCheckpointTests(CommandTestCase):
|
||
"""keyset pagination، batch اتمیک، و بازیابی از فایل وضعیت."""
|
||
|
||
def _customers_with_phone(self, count):
|
||
pks = []
|
||
for i in range(count):
|
||
customer = make_customer(f'مشتری {i}')
|
||
Customer.objects.filter(pk=customer.pk).update(phone1='09120%06d' % (i + 1))
|
||
pks.append(customer.pk)
|
||
return pks
|
||
|
||
|
||
def test_all_customers_covered_with_small_batches(self):
|
||
pks = self._customers_with_phone(7)
|
||
out = self.run_cmd(apply=True, batch_size=2)
|
||
self.assertEqual(
|
||
Customer.objects.filter(pk__in=pks).exclude(sms_primary_phones='').count(), 7)
|
||
self.assertIn('batchهای تکمیل‌شده: 4', out) # ۷ مشتری / ۲ = ۴ batch
|
||
|
||
def test_checkpoint_written_after_each_batch(self):
|
||
self._customers_with_phone(5)
|
||
self.run_cmd(apply=True, batch_size=2)
|
||
data = json.loads(self.checkpoint.read_text(encoding='utf-8'))
|
||
self.assertEqual(data['last_id'], Customer.objects.order_by('-id').first().pk)
|
||
self.assertEqual(data['batches'], 3)
|
||
self.assertTrue(data['applied'])
|
||
|
||
def test_resume_continues_from_checkpoint(self):
|
||
pks = self._customers_with_phone(4)
|
||
self.checkpoint.write_text(json.dumps({'last_id': pks[1]}), encoding='utf-8')
|
||
|
||
self.run_cmd(apply=True, resume=True)
|
||
self.assertEqual(Customer.objects.get(pk=pks[0]).sms_primary_phones, '')
|
||
self.assertEqual(
|
||
Customer.objects.filter(pk__in=pks[2:]).exclude(sms_primary_phones='').count(), 2)
|
||
|
||
def test_resume_without_checkpoint_fails(self):
|
||
with self.assertRaises(CommandError):
|
||
self.run_cmd(apply=True, resume=True)
|
||
|
||
def test_resume_with_invalid_checkpoint_fails(self):
|
||
self.checkpoint.write_text('{not json', encoding='utf-8')
|
||
with self.assertRaises(CommandError):
|
||
self.run_cmd(apply=True, resume=True)
|
||
|
||
def test_start_id_takes_precedence(self):
|
||
pks = self._customers_with_phone(4)
|
||
self.checkpoint.write_text(json.dumps({'last_id': 0}), encoding='utf-8')
|
||
self.run_cmd(apply=True, resume=True, start_id=pks[2])
|
||
self.assertEqual(Customer.objects.get(pk=pks[0]).sms_primary_phones, '')
|
||
self.assertEqual(Customer.objects.get(pk=pks[3]).sms_primary_phones,
|
||
'+989120000004')
|
||
|
||
def test_reset_checkpoint_restarts_from_beginning(self):
|
||
self._customers_with_phone(3)
|
||
self.run_cmd(apply=True, batch_size=1)
|
||
self.checkpoint.write_text(json.dumps({'last_id': 10**9}), encoding='utf-8')
|
||
self.run_cmd(apply=True, reset_checkpoint=True, batch_size=1)
|
||
data = json.loads(self.checkpoint.read_text(encoding='utf-8'))
|
||
self.assertEqual(data['batches'], 3) # از ابتدا شروع شد
|
||
|
||
def test_batch_failure_keeps_previous_batches_and_checkpoint(self):
|
||
pks = self._customers_with_phone(3)
|
||
original = QuerySet.bulk_update
|
||
calls = iter([1, 2])
|
||
|
||
def flaky(self, objs, fields, batch_size=None):
|
||
if next(calls, 2) == 2:
|
||
raise RuntimeError('boom')
|
||
return original(self, objs, fields, batch_size)
|
||
|
||
with mock.patch.object(QuerySet, 'bulk_update', flaky):
|
||
with self.assertRaises(CommandError):
|
||
self.run_cmd(apply=True, batch_size=1)
|
||
|
||
# batch اول commit شده، batch دوم rollback شده، مشتری سوم دست‌نخورده
|
||
self.assertEqual(Customer.objects.get(pk=pks[0]).sms_primary_phones,
|
||
'+989120000001')
|
||
self.assertEqual(Customer.objects.get(pk=pks[1]).sms_primary_phones, '')
|
||
self.assertEqual(Customer.objects.get(pk=pks[2]).sms_primary_phones, '')
|
||
data = json.loads(self.checkpoint.read_text(encoding='utf-8'))
|
||
self.assertEqual(data['last_id'], pks[0])
|
||
|
||
def test_resume_after_failure_completes_remaining(self):
|
||
pks = self._customers_with_phone(3)
|
||
original = QuerySet.bulk_update
|
||
calls = iter([1, 2])
|
||
|
||
def flaky(self, objs, fields, batch_size=None):
|
||
if next(calls, 2) == 2:
|
||
raise RuntimeError('boom')
|
||
return original(self, objs, fields, batch_size)
|
||
|
||
with mock.patch.object(QuerySet, 'bulk_update', flaky):
|
||
with self.assertRaises(CommandError):
|
||
self.run_cmd(apply=True, batch_size=1)
|
||
|
||
self.run_cmd(apply=True, resume=True, batch_size=1)
|
||
for pk in pks:
|
||
self.assertNotEqual(Customer.objects.get(pk=pk).sms_primary_phones, '')
|
||
|
||
|
||
class IntegrationAndEfficiencyTests(CommandTestCase):
|
||
"""سازگاری با مسیر پیامک موجود + جلوگیری از N+1."""
|
||
|
||
def test_panel_resolution_reads_back_aggregated_numbers(self):
|
||
customer = make_customer('گیرنده')
|
||
Customer.objects.filter(pk=customer.pk).update(
|
||
phone1='09120000001', phone2='09120000002')
|
||
self.run_cmd(apply=True)
|
||
customer.refresh_from_db()
|
||
# قالب ذخیره‌شده باید همان چیزی باشد که مسیر پیامک انتظار دارد.
|
||
self.assertEqual(resolve_recipients(customer),
|
||
['+989120000001', '+989120000002'])
|
||
|
||
def test_query_count_is_constant_per_batch(self):
|
||
for i in range(3):
|
||
customer = make_customer(f'کوئری {i}')
|
||
Customer.objects.filter(pk=customer.pk).update(phone1='09120%06d' % (i + 1))
|
||
for j in range(5):
|
||
PhoneNumber.objects.create(
|
||
customer=customer, title='t',
|
||
phone_number='09130%06d' % (i * 10 + j))
|
||
|
||
with self.assertNumQueries(3): # ۱ کوئری سقف دامنه + ۱ مشتریان + ۱ prefetch
|
||
self.run_cmd()
|
||
|