Practice script: set Reg1 operation statuses to "Draft"

bc_obps/registration/migrations/0079_migrate_reg1_operation_statuses.py

@@ -0,0 +1,111 @@
+# Generated by Django 5.0.11 on 2025-02-25 01:37
+
+from django.db import migrations
+from typing import Dict
+import datetime
+import uuid
+
+"""
+One-time forward-only migration to be applied to prod data.
+Purpose: set the `status` field of Operations (created in Reg1) to match the statuses used in Administration/Registration2 modules.
+Also archives operations with a "Declined" status (only 1 as of Feb 19, 2025)
+Reg1 Status        | Reg2 Status
+-------------------+-------------------
+PENDING            | DRAFT
+APPROVED           | DRAFT
+CHANGES_REQUESTED  | DRAFT
+NOT_STARTED        | NOT_STARTED
+DRAFT              | DRAFT
+DECLINED           | Archive the operation
+"""
+
+
+def count_stats(Operation) -> Dict[str, int]:
+    total = Operation.objects.count()
+    declined = Operation.objects.filter(status="Declined").count()
+    approved = Operation.objects.filter(status="Approved").count()
+    not_started = Operation.objects.filter(status="Not Started").count()
+    draft = Operation.objects.filter(status="Draft").count()
+    changes_requested = Operation.objects.filter(status="Changes Requested").count()
+    pending = Operation.objects.filter(status="Pending").count()
+    registered = Operation.objects.filter(status="Registered").count()
+    return {
+        'total': total,
+        'declined': declined,
+        'approved': approved,
+        'not_started': not_started,
+        'draft': draft,
+        'changes_requested': changes_requested,
+        'pending': pending,
+        'registered': registered,
+    }
+
+
+def migrate_reg1_operation_statuses(apps, schema_monitor):
+    # import the required Django model
+    Operation = apps.get_model('registration', 'Operation')
+    User = apps.get_model('registration', 'User')
+
+    before_stats = count_stats(Operation)
+    declined_operations_before = Operation.objects.filter(status="Declined")
+    print("\n\n\nDeclined operations - before migration:")
+    for op in declined_operations_before:
+        print(f'{op.id}: archived_by_id={op.archived_by_id}, archived_at={op.archived_at}')
+
+    pending_operations_updated = Operation.objects.filter(status="Pending").update(status="Draft")
+    approved_operations_updated = Operation.objects.filter(status="Approved").update(status="Draft")
+    changes_requested_operations_updated = Operation.objects.filter(status="Changes Requested").update(status="Draft")
+    filtered_users = User.objects.filter(first_name="Patricia", last_name="Russell")
+
+    assert filtered_users.count() == 1
+    po_user = filtered_users.first()
+
+    declined_operations_archived = Operation.objects.filter(status="Declined").update(
+        archived_by_id=po_user.user_guid, archived_at=datetime.datetime.now(datetime.timezone.utc)
+    )  # declined operations will be archived, per PR's instruction and using her user GUID
+    after_stats = count_stats(Operation)
+    declined_operations_after = Operation.objects.filter(status="Declined")
+    updates = {
+        'pending_ops_updated': pending_operations_updated,
+        'approved_ops_updated': approved_operations_updated,
+        'changes_requested_ops_updated': changes_requested_operations_updated,
+        'declined_operations_archived': declined_operations_archived,
+    }
+    assertions(before_stats, after_stats, updates, declined_operations_after)
+
+
+def assertions(before_stats, after_stats, updates, declined_operations_after):
+    print("\n\n\n*** BEFORE MIGRATION ***")
+    print(before_stats)
+    print("*** AFTER MIGRATION ***")
+    print(after_stats)
+    print("*** UPDATES ***")
+    print(updates)
+    print("\n\n\n")
+    # assert that the total number of operations does not change before and after the migration is applied
+    assert before_stats.get('total') == after_stats.get('total')
+    # assert that there are no more operations with a status of "Pending", "Approved", or "Changes Requested"
+    assert after_stats.get('pending') == 0
+    assert after_stats.get('approved') == 0
+    assert after_stats.get('changes_requested') == 0
+    # assert that the number of Declined operations before is the same as after the migration is applied
+    assert before_stats.get('declined') == after_stats.get('declined')
+    print("\n\n\nDeclined operations - after migration:")
+    for op in declined_operations_after:
+        print(f'{op.id}: archived_by_id={op.archived_by_id}, archived_at={op.archived_at}')
+    # assert that the Declined operations are marked as archived after the migration
+    for op in declined_operations_after:
+        assert op.archived_at is not None
+        assert op.archived_by_id == uuid.UUID('c3bc1b69-15de-44ac-b03b-982bf3163406')
+    # assert that the number of "Not Started" operations remains the same
+    assert before_stats.get('not_started') == after_stats.get('not_started')
+    # assert that the number of "Registered" operations is zero both before and after the migration is applied
+    assert before_stats.get('registered') == 0
+    assert after_stats.get('registered') == 0
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ('registration', '0078_V1_22_0'),
+    ]
+    operations = [migrations.RunPython(migrate_reg1_operation_statuses, migrations.RunPython.noop, elidable=True)]