lgcosta/paperless-ngx
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix tests for workflows, working nested triggers/actions

Browse Source
This commit is contained in:
shamoon 2023-12-24 08:53:53 -08:00
parent 6984fcf821
commit e328d5aa95
9 changed files with 158 additions and 41 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
src/documents/management/commands/document_exporter.py
View File

@ -23,7 +23,6 @@ from guardian.models import UserObjectPermission
from documents.file_handling import delete_empty_directories
from documents.file_handling import generate_filename
from documents.models import ConsumptionTemplate
from documents.models import Correspondent
from documents.models import CustomField
from documents.models import CustomFieldInstance
@ -35,6 +34,9 @@ from documents.models import SavedViewFilterRule
from documents.models import StoragePath
from documents.models import Tag
from documents.models import UiSettings
from documents.models import Workflow
from documents.models import WorkflowAction
from documents.models import WorkflowTrigger
from documents.settings import EXPORTER_ARCHIVE_NAME
from documents.settings import EXPORTER_FILE_NAME
from documents.settings import EXPORTER_THUMBNAIL_NAME
@ -284,7 +286,15 @@ class Command(BaseCommand):
)
manifest += json.loads(
serializers.serialize("json", ConsumptionTemplate.objects.all()),
serializers.serialize("json", WorkflowTrigger.objects.all()),
)
manifest += json.loads(
serializers.serialize("json", WorkflowAction.objects.all()),
)
manifest += json.loads(
serializers.serialize("json", Workflow.objects.all()),
)
manifest += json.loads(

6
src/documents/migrations/1044_workflow_workflowaction_workflowtrigger_and_more.py
View File

@ -44,7 +44,7 @@ def add_workflow_permissions(apps, schema_editor):
def remove_workflow_permissions(apps, schema_editor):
workflow_permissions = Permission.objects.filter(
codename__contains="workflow_permissions",
codename__contains="workflow",
)
for user in User.objects.all():
@ -54,7 +54,7 @@ def remove_workflow_permissions(apps, schema_editor):
group.permissions.remove(*workflow_permissions)
def migrate_consumption_templates(apps, schema_editor):
def migrate_consumption_templates(apps, schema_editor): # pragma: no cover
"""
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):
workflow.save()
def unmigrate_consumption_templates(apps, schema_editor):
def unmigrate_consumption_templates(apps, schema_editor): # pragma: no cover
model_name = "ConsumptionTemplate"
app_name = "documents"

2
src/documents/models.py
View File

@ -948,7 +948,7 @@ class WorkflowTrigger(models.Model):
verbose_name_plural = _("workflow triggers")
def __str__(self):
return f"WorfklowTrigger: {self.pk}"
return f"WorkflowTrigger {self.pk}"
class WorkflowAction(models.Model):

74
src/documents/serialisers.py
View File

@ -1262,6 +1262,7 @@ class BulkEditObjectPermissionsSerializer(serializers.Serializer, SetPermissions
class WorkflowTriggerSerializer(serializers.ModelSerializer):
id = serializers.IntegerField(required=False)
sources = fields.MultipleChoiceField(
choices=WorkflowTrigger.DocumentSourceChoices.choices,
allow_empty=False,
@ -1320,6 +1321,7 @@ class WorkflowTriggerSerializer(serializers.ModelSerializer):
class WorkflowActionSerializer(serializers.ModelSerializer):
id = serializers.IntegerField(required=False)
assign_correspondent = CorrespondentField(allow_null=True, required=False)
assign_tags = TagsField(many=True, allow_null=True, required=False)
assign_document_type = DocumentTypeField(allow_null=True, required=False)
@ -1370,36 +1372,70 @@ class WorkflowSerializer(serializers.ModelSerializer):
"actions",
]
def create(self, validated_data: Any) -> Any:
if "triggers" in validated_data:
# WorkflowTrigger.objects.update_or_create(triggers)
triggers = validated_data.pop("triggers")
if "actions" in validated_data:
# WorkflowAction.objects.update_or_create(actions)
actions = validated_data.pop("actions")
instance = super().create(validated_data)
def update_triggers_and_actions(self, instance: Workflow, triggers, actions):
set_triggers = []
set_actions = []
if triggers is not None:
for trigger in triggers:
print(trigger)
trigger_instance = WorkflowTrigger.objects.filter(**trigger).first()
if trigger_instance is not None:
set_triggers.append(trigger_instance)
trigger_instance, _ = WorkflowTrigger.objects.update_or_create(
id=trigger["id"],
defaults=trigger,
)
set_triggers.append(trigger_instance)
if actions is not None:
for action in actions:
print(action)
action_instance = WorkflowAction.objects.filter(**action).first()
if action_instance is not None:
set_actions.append(action_instance)
assign_tags = action.pop("assign_tags", None)
assign_view_users = action.pop("assign_view_users", None)
assign_view_groups = action.pop("assign_view_groups", None)
assign_change_users = action.pop("assign_change_users", None)
assign_change_groups = action.pop("assign_change_groups", None)
assign_custom_fields = action.pop("assign_custom_fields", None)
action_instance, _ = WorkflowAction.objects.update_or_create(
id=trigger["id"],
defaults=action,
)
if assign_tags is not None:
action_instance.assign_tags.set(assign_tags)
if assign_view_users is not None:
action_instance.assign_view_users.set(assign_view_users)
if assign_view_groups is not None:
action_instance.assign_view_groups.set(assign_view_groups)
if assign_change_users is not None:
action_instance.assign_change_users.set(assign_change_users)
if assign_change_groups is not None:
action_instance.assign_change_groups.set(assign_change_groups)
if assign_custom_fields is not None:
action_instance.assign_custom_fields.set(assign_custom_fields)
set_actions.append(action_instance)
instance.triggers.set(set_triggers)
instance.actions.set(set_actions)
instance.save()
def create(self, validated_data: Any) -> Workflow:
if "triggers" in validated_data:
triggers = validated_data.pop("triggers")
if "actions" in validated_data:
actions = validated_data.pop("actions")
instance = super().create(validated_data)
self.update_triggers_and_actions(instance, triggers, actions)
return instance
def update(self, instance: Any, validated_data: Any) -> Workflow:
if "triggers" in validated_data:
triggers = validated_data.pop("triggers")
if "actions" in validated_data:
actions = validated_data.pop("actions")
instance = super().update(instance, validated_data)
self.update_triggers_and_actions(instance, triggers, actions)
return instance

2
src/documents/tests/test_api_workflows.py
View File

@ -141,6 +141,7 @@ class TestApiWorkflows(DirectoriesMixin, APITestCase):
"order": 1,
"triggers": [
{
"id": trigger_response.data["id"],
"sources": [DocumentSource.ApiUpload],
"type": trigger_response.data["type"],
"filter_filename": trigger_response.data["filter_filename"],
@ -148,6 +149,7 @@ class TestApiWorkflows(DirectoriesMixin, APITestCase):
],
"actions": [
{
"id": action_response.data["id"],
"assign_title": action_response.data["assign_title"],
},
],

31
src/documents/tests/test_management_exporter.py
View File

@ -21,7 +21,6 @@ from guardian.models import UserObjectPermission
from guardian.shortcuts import assign_perm
from documents.management.commands import document_exporter
from documents.models import ConsumptionTemplate
from documents.models import Correspondent
from documents.models import CustomField
from documents.models import CustomFieldInstance
@ -31,6 +30,9 @@ from documents.models import Note
from documents.models import StoragePath
from documents.models import Tag
from documents.models import User
from documents.models import Workflow
from documents.models import WorkflowAction
from documents.models import WorkflowTrigger
from documents.sanity_checker import check_sanity
from documents.settings import EXPORTER_FILE_NAME
from documents.tests.utils import DirectoriesMixin
@ -109,7 +111,16 @@ class TestExportImport(DirectoriesMixin, FileSystemAssertsMixin, TestCase):
self.d4.storage_path = self.sp1
self.d4.save()
self.ct1 = ConsumptionTemplate.objects.create(name="CT 1", filter_path="*")
self.trigger = WorkflowTrigger.objects.create(
type=WorkflowTrigger.WorkflowTriggerType.CONSUMPTION,
sources=[1],
filter_filename="*",
)
self.action = WorkflowAction.objects.create(assign_title="new title")
self.workflow = Workflow.objects.create(name="Workflow 1", order="0")
self.workflow.triggers.add(self.trigger)
self.workflow.actions.add(self.action)
self.workflow.save()
super().setUp()
@ -168,7 +179,7 @@ class TestExportImport(DirectoriesMixin, FileSystemAssertsMixin, TestCase):
manifest = self._do_export(use_filename_format=use_filename_format)
self.assertEqual(len(manifest), 172)
self.assertEqual(len(manifest), 189)
# dont include consumer or AnonymousUser users
self.assertEqual(
@ -262,7 +273,7 @@ class TestExportImport(DirectoriesMixin, FileSystemAssertsMixin, TestCase):
self.assertEqual(Document.objects.get(id=self.d4.id).title, "wow_dec")
self.assertEqual(GroupObjectPermission.objects.count(), 1)
self.assertEqual(UserObjectPermission.objects.count(), 1)
self.assertEqual(Permission.objects.count(), 124)
self.assertEqual(Permission.objects.count(), 136)
messages = check_sanity()
# everything is alright after the test
self.assertEqual(len(messages), 0)
@ -694,15 +705,15 @@ class TestExportImport(DirectoriesMixin, FileSystemAssertsMixin, TestCase):
os.path.join(self.dirs.media_dir, "documents"),
)
self.assertEqual(ContentType.objects.count(), 31)
self.assertEqual(Permission.objects.count(), 124)
self.assertEqual(ContentType.objects.count(), 34)
self.assertEqual(Permission.objects.count(), 136)
manifest = self._do_export()
with paperless_environment():
self.assertEqual(
len(list(filter(lambda e: e["model"] == "auth.permission", manifest))),
124,
136,
)
# add 1 more to db to show objects are not re-created by import
Permission.objects.create(
@ -710,7 +721,7 @@ class TestExportImport(DirectoriesMixin, FileSystemAssertsMixin, TestCase):
codename="test_perm",
content_type_id=1,
)
self.assertEqual(Permission.objects.count(), 125)
self.assertEqual(Permission.objects.count(), 137)
# will cause an import error
self.user.delete()
@ -719,5 +730,5 @@ class TestExportImport(DirectoriesMixin, FileSystemAssertsMixin, TestCase):
with self.assertRaises(IntegrityError):
call_command("document_importer", "--no-progress-bar", self.target)
self.assertEqual(ContentType.objects.count(), 31)
self.assertEqual(Permission.objects.count(), 125)
self.assertEqual(ContentType.objects.count(), 34)
self.assertEqual(Permission.objects.count(), 137)

19
src/documents/tests/test_migration_consumption_templates.py
View File

@ -33,11 +33,18 @@ class TestReverseMigrateConsumptionTemplate(TestMigrations):
self.Permission = apps.get_model("auth", "Permission")
self.user = User.objects.create(username="user1")
self.group = Group.objects.create(name="group1")
permission = self.Permission.objects.get(codename="add_consumptiontemplate")
self.user.user_permissions.add(permission.id)
self.group.permissions.add(permission.id)
permission = self.Permission.objects.filter(
codename="add_consumptiontemplate",
).first()
if permission is not None:
self.user.user_permissions.add(permission.id)
self.group.permissions.add(permission.id)
def test_remove_consumptiontemplate_permissions(self):
permission = self.Permission.objects.get(codename="add_consumptiontemplate")
self.assertFalse(self.user.has_perm(f"documents.{permission.codename}"))
self.assertFalse(permission in self.group.permissions.all())
permission = self.Permission.objects.filter(
codename="add_consumptiontemplate",
).first()
# can be None ? now that CTs removed
if permission is not None:
self.assertFalse(self.user.has_perm(f"documents.{permission.codename}"))
self.assertFalse(permission in self.group.permissions.all())

49
src/documents/tests/test_migration_workflows.py Normal file
View File

@ -0,0 +1,49 @@
from django.contrib.auth import get_user_model
from documents.tests.utils import TestMigrations
class TestMigrateWorkflow(TestMigrations):
migrate_from = "1043_alter_savedviewfilterrule_rule_type"
migrate_to = "1044_workflow_workflowaction_workflowtrigger_and_more"
def setUpBeforeMigration(self, apps):
User = get_user_model()
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")
permission = self.Permission.objects.get(codename="add_document")
self.user.user_permissions.add(permission.id)
self.group.permissions.add(permission.id)
def test_users_with_add_documents_get_add_workflow(self):
permission = self.Permission.objects.get(codename="add_workflow")
self.assertTrue(self.user.has_perm(f"documents.{permission.codename}"))
self.assertTrue(permission in self.group.permissions.all())
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()
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")
permission = self.Permission.objects.filter(
codename="add_workflow",
).first()
if permission is not None:
self.user.user_permissions.add(permission.id)
self.group.permissions.add(permission.id)
def test_remove_workflow_permissions(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.group.permissions.all())

2
src/documents/tests/test_workflows.py
View File

@ -116,6 +116,8 @@ class TestWorkflows(DirectoriesMixin, FileSystemAssertsMixin, TestCase):
w.save()
self.assertEqual(w.__str__(), "Workflow: Workflow 1")
self.assertEqual(trigger.__str__(), "WorkflowTrigger 1")
self.assertEqual(action.__str__(), "WorkflowAction 1")
test_file = self.SAMPLE_DIR / "simple.pdf"