Add coverage for consumption template migrations

src/documents/migrations/1044_workflow_workflowaction_workflowtrigger_and_more.py

@@ -54,7 +54,7 @@ def remove_workflow_permissions(apps, schema_editor):
         group.permissions.remove(*workflow_permissions)
 
 
-def migrate_consumption_templates(apps, schema_editor):  # pragma: no cover
+def migrate_consumption_templates(apps, schema_editor):
     """
     Migrate consumption templates to workflows. At this point ConsumptionTemplate still exists
     but objects are not returned as their true model so we have to manually do that
@@ -142,7 +142,7 @@ def migrate_consumption_templates(apps, schema_editor):  # pragma: no cover
             workflow.save()
 
 
-def unmigrate_consumption_templates(apps, schema_editor):  # pragma: no cover
+def unmigrate_consumption_templates(apps, schema_editor):
     model_name = "ConsumptionTemplate"
     app_name = "documents"
 
@@ -152,7 +152,7 @@ def unmigrate_consumption_templates(apps, schema_editor):  # pragma: no cover
         template = ConsumptionTemplate.objects.create(
             name=workflow.name,
             order=workflow.order,
-            sources=workflow.sources,
+            sources=workflow.triggers.first().sources,
             filter_path=workflow.triggers.first().filter_path,
             filter_filename=workflow.triggers.first().filter_filename,
             filter_mailrule=workflow.triggers.first().filter_mailrule,

src/documents/tests/test_migration_workflows.py

@@ -1,5 +1,4 @@
-from django.contrib.auth import get_user_model
+from documents.data_models import DocumentSource
-
 from documents.tests.utils import TestMigrations
 
 
@@ -8,8 +7,8 @@ class TestMigrateWorkflow(TestMigrations):
     migrate_to = "1044_workflow_workflowaction_workflowtrigger_and_more"
 
     def setUpBeforeMigration(self, apps):
-        User = get_user_model()
+        User = apps.get_model("auth", "User")
-        Group = apps.get_model("auth.Group")
+        Group = apps.get_model("auth", "Group")
         self.Permission = apps.get_model("auth", "Permission")
         self.user = User.objects.create(username="user1")
         self.group = Group.objects.create(name="group1")
@@ -17,19 +16,65 @@ class TestMigrateWorkflow(TestMigrations):
         self.user.user_permissions.add(permission.id)
         self.group.permissions.add(permission.id)
 
-    def test_users_with_add_documents_get_add_workflow(self):
+        # create a CT to migrate
+        c = apps.get_model("documents", "Correspondent").objects.create(
+            name="Correspondent Name",
+        )
+        dt = apps.get_model("documents", "DocumentType").objects.create(
+            name="DocType Name",
+        )
+        t1 = apps.get_model("documents", "Tag").objects.create(name="t1")
+        sp = apps.get_model("documents", "StoragePath").objects.create(path="/test/")
+        cf1 = apps.get_model("documents", "CustomField").objects.create(
+            name="Custom Field 1",
+            data_type="string",
+        )
+
+        user2 = User.objects.create(username="user2")
+        user3 = User.objects.create(username="user3")
+        group2 = Group.objects.create(name="group2")
+
+        model_name = "ConsumptionTemplate"
+        app_name = "documents"
+        ConsumptionTemplate = apps.get_model(app_label=app_name, model_name=model_name)
+
+        ct = ConsumptionTemplate.objects.create(
+            name="Template 1",
+            order=0,
+            sources=f"{DocumentSource.ApiUpload},{DocumentSource.ConsumeFolder},{DocumentSource.MailFetch}",
+            filter_filename="*simple*",
+            filter_path="*/samples/*",
+            assign_title="Doc from {correspondent}",
+            assign_correspondent=c,
+            assign_document_type=dt,
+            assign_storage_path=sp,
+            assign_owner=user2,
+        )
+
+        ct.assign_tags.add(t1)
+        ct.assign_view_users.add(user3)
+        ct.assign_view_groups.add(group2)
+        ct.assign_change_users.add(user3)
+        ct.assign_change_groups.add(group2)
+        ct.assign_custom_fields.add(cf1)
+        ct.save()
+
+    def test_users_with_add_documents_get_add_and_workflow_templates_get_migrated(self):
         permission = self.Permission.objects.get(codename="add_workflow")
-        self.assertTrue(self.user.has_perm(f"documents.{permission.codename}"))
+        self.assertTrue(permission in self.user.user_permissions.all())
         self.assertTrue(permission in self.group.permissions.all())
 
+        Workflow = self.apps.get_model("documents", "Workflow")
+        self.assertEqual(Workflow.objects.all().count(), 1)
+
 
 class TestReverseMigrateWorkflow(TestMigrations):
     migrate_from = "1044_workflow_workflowaction_workflowtrigger_and_more"
     migrate_to = "1043_alter_savedviewfilterrule_rule_type"
 
     def setUpBeforeMigration(self, apps):
-        User = get_user_model()
+        User = apps.get_model("auth", "User")
-        Group = apps.get_model("auth.Group")
+        Group = apps.get_model("auth", "Group")
         self.Permission = apps.get_model("auth", "Permission")
         self.user = User.objects.create(username="user1")
         self.group = Group.objects.create(name="group1")
@@ -40,10 +85,37 @@ class TestReverseMigrateWorkflow(TestMigrations):
             self.user.user_permissions.add(permission.id)
             self.group.permissions.add(permission.id)
 
-    def test_remove_workflow_permissions(self):
+        Workflow = apps.get_model("documents", "Workflow")
+        WorkflowTrigger = apps.get_model("documents", "WorkflowTrigger")
+        WorkflowAction = apps.get_model("documents", "WorkflowAction")
+
+        trigger = WorkflowTrigger.objects.create(
+            type=0,
+            sources=[DocumentSource.ConsumeFolder],
+            filter_path="*/path/*",
+            filter_filename="*file*",
+        )
+
+        action = WorkflowAction.objects.create(
+            assign_title="assign title",
+        )
+        workflow = Workflow.objects.create(
+            name="workflow 1",
+            order=0,
+        )
+        workflow.triggers.set([trigger])
+        workflow.actions.set([action])
+        workflow.save()
+
+    def test_remove_workflow_permissions_and_migrate_workflows_to_consumption_templates(
+        self,
+    ):
         permission = self.Permission.objects.filter(
             codename="add_workflow",
         ).first()
         if permission is not None:
-            self.assertFalse(self.user.has_perm(f"documents.{permission.codename}"))
+            self.assertFalse(permission in self.user.user_permissions.all())
             self.assertFalse(permission in self.group.permissions.all())
+
+        ConsumptionTemplate = self.apps.get_model("documents", "ConsumptionTemplate")
+        self.assertEqual(ConsumptionTemplate.objects.all().count(), 1)