chore: completed Django migration for creating contacts from industry…

… users, tested against prod

bc_obps/registration/migrations/0078_make_contacts_from_industry_users.py

@@ -1,6 +1,7 @@
 # Generated by Django 5.0.11 on 2025-02-14 01:29
 
-from django.db import migrations
+from typing import Dict
+from django.db import migrations, models
 
 
 """
@@ -9,17 +10,33 @@
 """
 
 
+def count_stats(User, Contact, UserOperator) -> Dict[str, int]:
+    total_users = User.objects.count()
+    industry_users = User.objects.filter(app_role__role_name="industry_user").count()
+    contacts = Contact.objects.count()
+    approved_user_operators = UserOperator.objects.filter(status="Approved").count()
+
+    return {
+        'total_users': total_users,
+        'industry_users': industry_users,
+        'contacts': contacts,
+        'approved_user_operators': approved_user_operators,
+    }
+
+
 def create_contacts_from_prod_industry_users(apps, schema_monitor):
     # import the required django models
     User = apps.get_model('registration', 'User')
     Contact = apps.get_model('registration', 'Contact')
     UserOperator = apps.get_model('registration', 'UserOperator')
     BusinessRole = apps.get_model('registration', 'BusinessRole')
 
+    before_stats = count_stats(User, Contact, UserOperator)
+
     # filter on Users to fetch only those with app_role_id == "industry_user"
     # and status == "Approved", and whose email doesn't appear in the Contact table
     approved_industry_users = (
-        User.objects.filter(app_role__role_name="industry_user", user_operators__status=UserOperator.Statuses.APPROVED)
+        User.objects.filter(app_role__role_name="industry_user", user_operators__status="Approved")
         .exclude(email__in=Contact.objects.values_list("email", flat=True))
         .distinct()
     )
@@ -35,12 +52,51 @@ def create_contacts_from_prod_industry_users(apps, schema_monitor):
             phone_number=user.phone_number,
             business_role=operation_representative_role,
             address=None,
-            operator=user.user_operator.operator,
+            operator=user.user_operators.first().operator,
         )
         for user in approved_industry_users
     ]
     Contact.objects.bulk_create(new_contacts)
 
+    print(f'\n\n\n{len(new_contacts)} new contacts created')
+
+    after_stats = count_stats(User, Contact, UserOperator)
+
+    assertions(before_stats, after_stats, Contact, User)
+
+
+def assertions(before_stats, after_stats, Contact, User):
+    print("\n\n\n*** BEFORE MIGRATION ***")
+    print(before_stats)
+    print("*** AFTER MIGRATION ***")
+    print(after_stats)
+    print("\n\n\n")
+
+    # assert that there are at least as many contacts as there are user_operators with "Approved" status
+    # (since every industry_user should also be a contact, but there may be some contacts
+    # who aren't users)
+    assert after_stats.get('contacts') >= after_stats.get('approved_user_operators')
+
+    # assert that no internal users (role "cas_x") appear in the Contacts table
+    assert not Contact.objects.filter(
+        email__in=User.objects.filter(app_role__role_name__startswith="cas_").values_list("email", flat=True)
+    ).exists()
+
+    # assert that each combination of (email_address, operator_id) is unique
+    # assert not Contact.objects.values("email", "operator_id").annotate(count=models.Count("id")).filter(count__gt=1)
+    # NOTE: this assertion fails because there's duplicate contact data in the prod dataset.
+    # This is because a new Contact was created for each "Point of Contact" that an operation has in Reg1
+
+    # to see a printout of the duplicated contacts:
+    print(Contact.objects.values("email", "operator_id").annotate(count=models.Count("id")).filter(count__gt=1))
+
+    # basic assertion - there should be at least as many contacts after the migration as before (probably more)
+    assert after_stats.get('contacts') >= before_stats.get('contacts')
+
+    # basic assertion - no new users should have been created (and none deleted)
+    assert after_stats.get('total_users') == before_stats.get('total_users')
+    assert after_stats.get('industry_users') == before_stats.get('industry_users')
+
 
 class Migration(migrations.Migration):
     dependencies = [