"""تست‌های دستور ``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()