migration repaire

apps/customer/test_mobile_audit.py

@@ -0,0 +1,205 @@
+"""Legacy deployment diagnostics must expose conflicts without changing data."""
+import json
+from importlib import import_module
+from io import StringIO
+from unittest.mock import MagicMock, Mock, patch
+
+from django.apps import apps
+from django.contrib import admin
+from django.core.management import call_command
+from django.core.management.base import CommandError
+from django.db import connection, transaction
+from django.db.migrations.executor import MigrationExecutor
+from django.test import TransactionTestCase
+from django.test.utils import CaptureQueriesContext
+
+from apps.cuser.cuser_model import CUser
+from apps.customer.models import Customer, CustomerNote, MobileConflict
+from apps.jwt_auth.test_regressions import RegressionFixtures
+from apps.workspace.models import Workspace
+from apps.shop.models import Order
+
+
+def audit(*args):
+    output = StringIO()
+    call_command('audit_mobile_duplicates', *args, stdout=output)
+    return json.loads(output.getvalue())
+
+
+def repair():
+    migration = import_module('apps.customer.migrations.0008_customer_mobile_unique')
+    migration.check_and_normalize_mobiles(apps, Mock(connection=connection))
+
+
+class MobileAuditTests(RegressionFixtures):
+    def test_country_prefix_conflict_reports_all_members_and_dependencies_read_only(self):
+        other = Customer.objects.create(name_family='Legacy', mobile=989123456789, sys_ws=self.workspace)
+        note = CustomerNote.objects.create(customer=other, cuser_role=self.role,
+                                           desc='Keep this note', sys_ws=self.workspace)
+        before = list(Customer.objects.order_by('pk').values())
+        with CaptureQueriesContext(connection) as queries:
+            report = audit()
+        self.assertEqual(report['customer_duplicate_groups'], [{
+            'workspace_id': self.workspace.pk, 'canonical_mobile': self.customer.mobile,
+            'record_ids': [self.customer.pk, other.pk],
+        }])
+        self.assertEqual(report['duplicate_group_count'], 1)
+        self.assertEqual(report['customers'][1]['related_counts']['customer.CustomerNote.customer'], 1)
+        self.assertEqual(report['customers'][0]['panel_role']['user_id'], self.user.pk)
+        self.assertEqual(before, list(Customer.objects.order_by('pk').values()))
+        self.assertTrue(CustomerNote.objects.filter(pk=note.pk).exists())
+        writes = [query['sql'] for query in queries if query['sql'].lstrip().split()[0].upper()
+                  in {'INSERT', 'UPDATE', 'DELETE', 'ALTER', 'DROP', 'CREATE'}]
+        self.assertEqual(writes, [])
+
+    def test_cross_workspace_numbers_and_multiple_nulls_are_allowed(self):
+        Customer.objects.create(name_family='Other tenant', mobile=989123456789, sys_ws=self.other_workspace)
+        Customer.objects.create(name_family='No number 1', mobile=None, sys_ws=self.workspace)
+        Customer.objects.create(name_family='No number 2', mobile=None, sys_ws=self.workspace)
+        self.assertEqual(audit()['duplicate_group_count'], 0)
+
+    def test_user_country_prefix_conflict_is_global_and_excludes_secrets(self):
+        user = CUser.objects.create(username='legacy', name_family='Legacy', mobile=989123456789)
+        report = audit()
+        self.assertEqual(report['user_duplicate_groups'][0]['record_ids'], [self.user.pk, user.pk])
+        self.assertEqual(report['duplicate_group_count'], 1)
+        self.assertNotIn('password', report['users'][0])
+        user.refresh_from_db()
+        self.assertEqual(user.mobile, 989123456789)
+
+    def test_requested_customer_details_and_missing_ids_without_duplicates(self):
+        report = audit('--customer-ids', str(self.customer.pk), '999999')
+        self.assertEqual(report['customers'][0]['id'], self.customer.pk)
+        self.assertEqual(report['missing_requested_customer_ids'], [999999])
+        self.assertEqual(report['duplicate_group_count'], 0)
+
+    def test_invalid_database_has_clear_error(self):
+        with self.assertRaisesMessage(CommandError, 'Unknown database alias'):
+            audit('--database', 'missing')
+
+    def test_automatic_repair_preserves_notes_orders_audit_and_blocks_old_panel_session(self):
+        tokens = self.use_role()
+        other = Customer.objects.create(name_family='Legacy', mobile=989123456789, sys_ws=self.workspace)
+        note = CustomerNote.objects.create(customer=self.customer, cuser_role=self.role,
+                                           desc='Keep this note', sys_ws=self.workspace)
+        order = Order.objects.create(customer=other, total_price=100, factor_type=1, sys_ws=self.workspace)
+        repair()
+        self.assertTrue(CustomerNote.objects.filter(pk=note.pk, customer=self.customer).exists())
+        self.assertTrue(Order.objects.filter(pk=order.pk, customer=other).exists())
+        entries = MobileConflict.objects.filter(source_model='customer.Customer')
+        self.assertEqual(entries.count(), 2)
+        archived = entries.get(record_id=self.customer.pk)
+        self.assertEqual(archived.snapshot['cuser_role_id'], self.role.pk)
+        self.assertTrue(archived.snapshot['roles'][0]['customer_panel'])
+        self.assertEqual(self.client.get('/cuser/info/').status_code, 401)
+        self.assertEqual(self.post('token/refresh/', {'refresh': tokens['refresh']}).status_code, 401)
+        repair()
+        self.assertEqual(entries.count(), 2)
+        self.assertEqual(entries.get(record_id=self.customer.pk).snapshot, archived.snapshot)
+
+    def test_account_collision_disables_all_ambiguous_accounts_without_changing_their_numbers(self):
+        tokens = self.use_role()
+        other = CUser.objects.create(username='legacy', mobile=989123456789, is_active=True)
+        repair()
+        self.user.refresh_from_db()
+        other.refresh_from_db()
+        self.role.refresh_from_db()
+        self.operator.refresh_from_db()
+        self.assertFalse(self.user.is_active)
+        self.assertFalse(other.is_active)
+        self.assertFalse(self.role.is_active)
+        self.assertFalse(self.operator.is_active)
+        self.assertEqual(self.user.mobile, 9123456789)
+        self.assertEqual(other.mobile, 989123456789)
+        self.assertEqual(MobileConflict.objects.filter(source_model='cuser.CUser').count(), 2)
+        self.assertEqual(self.client.get('/cuser/info/').status_code, 401)
+        self.assertEqual(self.post('token/refresh/', {'refresh': tokens['refresh']}).status_code, 401)
+        before = list(MobileConflict.objects.order_by('pk').values())
+        repair()
+        self.assertEqual(before, list(MobileConflict.objects.order_by('pk').values()))
+
+    def test_failed_archival_rolls_back_entire_repair(self):
+        other = Customer.objects.create(name_family='Legacy', mobile=989123456789, sys_ws=self.workspace)
+        Customer.objects.create(name_family='Second group', mobile=9333333333, sys_ws=self.workspace)
+        Customer.objects.create(name_family='Second group legacy', mobile=989333333333, sys_ws=self.workspace)
+        module = import_module('apps.customer.mobile_migration_repair')
+        archive = module._archive
+        calls = 0
+
+        def failing_archive(*args, **kwargs):
+            nonlocal calls
+            calls += 1
+            if calls == 3:
+                raise RuntimeError('Simulated audit storage failure')
+            return archive(*args, **kwargs)
+
+        with self.assertRaisesMessage(RuntimeError, 'Simulated audit storage failure'):
+            with transaction.atomic(), patch.object(module, '_archive', side_effect=failing_archive):
+                repair()
+        self.customer.refresh_from_db()
+        other.refresh_from_db()
+        self.assertEqual(self.customer.mobile, 9123456789)
+        self.assertEqual(other.mobile, 989123456789)
+        self.assertEqual(MobileConflict.objects.count(), 0)
+
+    def test_admin_audit_history_cannot_be_added_changed_or_deleted(self):
+        model_admin = admin.site._registry[MobileConflict]
+        request = Mock(user=Mock(is_superuser=True, is_active=True))
+        self.assertFalse(model_admin.has_add_permission(request))
+        self.assertFalse(model_admin.has_change_permission(request))
+        self.assertFalse(model_admin.has_delete_permission(request))
+
+    def test_postgres_migration_locks_legacy_writers_and_drains_deferred_checks(self):
+        cursor = MagicMock()
+        postgres = Mock(alias='default', vendor='postgresql', ops=connection.ops)
+        postgres.cursor.return_value.__enter__ = Mock(return_value=cursor)
+        postgres.cursor.return_value.__exit__ = Mock(return_value=False)
+        migration = import_module('apps.customer.migrations.0008_customer_mobile_unique')
+        migration.check_and_normalize_mobiles(apps, Mock(connection=postgres))
+        statements = [call.args[0] for call in cursor.execute.call_args_list]
+        self.assertTrue(statements[0].startswith('LOCK TABLE '))
+        self.assertIn('SHARE ROW EXCLUSIVE MODE', statements[0])
+        self.assertEqual(statements[-1], 'SET CONSTRAINTS ALL IMMEDIATE')
+
+
+class LegacyMobileAuditTests(TransactionTestCase):
+    def test_exact_duplicates_before_migration_are_automatically_quarantined(self):
+        executor = MigrationExecutor(connection)
+        targets = executor.loader.graph.leaf_nodes()
+        executor.migrate([('customer', '0007_alter_customer_name_family')])
+        try:
+            # Real legacy databases do not have the new audit table yet.
+            with connection.schema_editor() as editor:
+                editor.delete_model(MobileConflict)
+            workspace = Workspace.objects.create(name='Legacy workspace', username='legacy-workspace')
+            first = Customer.objects.create(name_family='First', mobile=9123456789, sys_ws=workspace)
+            second = Customer.objects.create(name_family='Second', mobile=9123456789, sys_ws=workspace)
+            report = audit('--customer-ids', str(first.pk), str(second.pk))
+            self.assertEqual(report['customer_duplicate_groups'][0]['record_ids'], [first.pk, second.pk])
+            self.assertEqual(Customer.objects.filter(mobile=9123456789).count(), 2)
+            MigrationExecutor(connection).migrate(targets)
+            self.assertEqual(Customer.objects.filter(pk__in=[first.pk, second.pk]).count(), 2)
+            self.assertEqual(Customer.objects.filter(pk__in=[first.pk, second.pk], mobile=None).count(), 2)
+            self.assertEqual(MobileConflict.objects.filter(source_model='customer.Customer').count(), 2)
+            self.assertEqual(MobileConflict.objects.get(record_id=first.pk).original_mobile, 9123456789)
+            self.assertEqual(audit()['duplicate_group_count'], 0)
+        finally:
+            # Test data only: restore the normal schema for subsequent tests.
+            Customer.objects.all().delete()
+            MigrationExecutor(connection).migrate(targets)
+
+    def test_already_applied_0008_installation_creates_missing_audit_table(self):
+        executor = MigrationExecutor(connection)
+        targets = executor.loader.graph.leaf_nodes()
+        executor.migrate([('customer', '0008_customer_mobile_unique')])
+        try:
+            # Simulate an installation that applied the original 0008 before
+            # this automatic repair feature existed.
+            with connection.schema_editor() as editor:
+                editor.delete_model(MobileConflict)
+            self.assertNotIn(MobileConflict._meta.db_table, connection.introspection.table_names())
+            MigrationExecutor(connection).migrate(targets)
+            self.assertIn(MobileConflict._meta.db_table, connection.introspection.table_names())
+            self.assertEqual(MobileConflict.objects.count(), 0)
+        finally:
+            MigrationExecutor(connection).migrate(targets)