From ba421eaa1094322289b8f277704cf170a0cb4ae9 Mon Sep 17 00:00:00 2001 From: Viliam Balaz Date: Sat, 23 Jan 2021 19:19:08 +0100 Subject: [PATCH] #342 Create attachment_anonymization management command (#344) * #342 Create attachment_anonymization management command * #342 Remade attachment_anonymization management command * #342 Read content from stdin * #342 Remove sys.stdin.isatty * #342 Use transaction.atomic * #342 Use transaction.atomic --- .../apps/anonymization/management/__init__.py | 0 .../management/commands/__init__.py | 0 .../commands/attachment_anonymization.py | 81 ++++++++++++ chcemvediet/apps/anonymization/models.py | 3 + .../apps/anonymization/tests/__init__.py | 0 .../tests/test_management_commands.py | 115 ++++++++++++++++++ misc/anonymization.md | 22 ++++ 7 files changed, 221 insertions(+) create mode 100644 chcemvediet/apps/anonymization/management/__init__.py create mode 100644 chcemvediet/apps/anonymization/management/commands/__init__.py create mode 100644 chcemvediet/apps/anonymization/management/commands/attachment_anonymization.py create mode 100644 chcemvediet/apps/anonymization/tests/__init__.py create mode 100644 chcemvediet/apps/anonymization/tests/test_management_commands.py diff --git a/chcemvediet/apps/anonymization/management/__init__.py b/chcemvediet/apps/anonymization/management/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/chcemvediet/apps/anonymization/management/commands/__init__.py b/chcemvediet/apps/anonymization/management/commands/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/chcemvediet/apps/anonymization/management/commands/attachment_anonymization.py b/chcemvediet/apps/anonymization/management/commands/attachment_anonymization.py new file mode 100644 index 000000000..158f5ce77 --- /dev/null +++ b/chcemvediet/apps/anonymization/management/commands/attachment_anonymization.py @@ -0,0 +1,81 @@ +import sys +from optparse import make_option + +import magic +from django.core.files.base import ContentFile +from django.core.management.base import BaseCommand, CommandError +from django.db import transaction + +from poleno.attachments.models import Attachment +from poleno.utils.misc import squeeze +from chcemvediet.apps.anonymization.models import AttachmentFinalization +from chcemvediet.apps.inforequests.models import Action + + +class Command(BaseCommand): + args = u'attachment_id [file]' + help = squeeze(u""" + Creates anonymization for the specified Attachment. The anonymized content is read from + the given file or from stdin if no file is specified. If no file is specified and stdin + is empty, the command will fail. + """) + + option_list = BaseCommand.option_list + ( + make_option(u'--content_type', + help=squeeze(u""" + Content type of file, e.g. "application/pdf". Automatically guessed from + the file content if not specified. + """) + ), + make_option(u'--debug', + default=u'', + help=u'Debug message to the newly created anonymization. Empty by default.' + ), + make_option(u'--force', + action=u'store_true', + help=u'Overwrite any existing anonymization for the attachment.' + ), + ) + + @transaction.atomic + def handle(self, *args, **options): + if not args: + raise CommandError(u'attachment_anonymization takes at least 1 argument (0 given).') + elif len(args) > 2: + raise CommandError( + u'attachment_anonymization takes at most 2 arguments ({} given).'.format(len(args)) + ) + + attachment_pk = args[0] + try: + attachment = Attachment.objects.attached_to(Action).get(pk=attachment_pk) + except (Attachment.DoesNotExist, ValueError): + raise CommandError( + u'Attachment instance with pk "{}" does not exist.'.format(attachment_pk) + ) + attachments_finalization = (AttachmentFinalization.objects + .filter(attachment=attachment) + .successful()) + if not options[u'force'] and attachments_finalization: + raise CommandError(u'Anonymization already exists. Use the --force to overwrite it.') + + if len(args) == 2: + filename = args[1] + try: + with open(filename, u'rb') as file: + content = file.read() + except IOError as e: + raise CommandError(u'Could not open file: {}.'.format(e)) + else: + content = sys.stdin.read() + if not content: + raise CommandError(u'No content given.') + + attachments_finalization.delete() + AttachmentFinalization.objects.create( + attachment=attachment, + successful=True, + file=ContentFile(content), + content_type=options[u'content_type'] or magic.from_buffer(content, mime=True), + debug=options[u'debug'], + ) diff --git a/chcemvediet/apps/anonymization/models.py b/chcemvediet/apps/anonymization/models.py index a60c22845..5120019a0 100644 --- a/chcemvediet/apps/anonymization/models.py +++ b/chcemvediet/apps/anonymization/models.py @@ -322,6 +322,9 @@ def __unicode__(self): return format(self.pk) class AttachmentFinalizationQuerySet(QuerySet): + def successful(self): + return self.filter(successful=True) + def order_by_pk(self): return self.order_by(u'pk') diff --git a/chcemvediet/apps/anonymization/tests/__init__.py b/chcemvediet/apps/anonymization/tests/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/chcemvediet/apps/anonymization/tests/test_management_commands.py b/chcemvediet/apps/anonymization/tests/test_management_commands.py new file mode 100644 index 000000000..62e644500 --- /dev/null +++ b/chcemvediet/apps/anonymization/tests/test_management_commands.py @@ -0,0 +1,115 @@ +import os +import sys +from StringIO import StringIO +from testfixtures import TempDirectory + +from django.core.management import call_command +from django.core.management.base import CommandError +from django.test import TestCase + +from chcemvediet.apps.anonymization.models import AttachmentFinalization +from chcemvediet.tests import ChcemvedietTestCaseMixin + + +class AttachmentAnonymizationManagementCommandTest(ChcemvedietTestCaseMixin, TestCase): + + def _pre_setup(self): + super(AttachmentAnonymizationManagementCommandTest, self)._pre_setup() + self.tempdir = TempDirectory() + self.tempdir.write(u'testfile.txt', u'Default testing content') + self.filename = os.path.join(self.tempdir.path, u'testfile.txt') + self.attachment = self._create_attachment() + + def _post_teardown(self): + self.tempdir.cleanup() + super(AttachmentAnonymizationManagementCommandTest, self)._post_teardown() + + def _create_attachment(self, **kwargs): + return super(AttachmentAnonymizationManagementCommandTest, self)._create_attachment( + generic_object=self.action, + **kwargs + ) + + + def test_attachment_and_file_arguments(self): + call_command(u'attachment_anonymization', self.attachment.pk, self.filename) + attachment_finalization = AttachmentFinalization.objects.get(attachment=self.attachment) + self.assertEqual(attachment_finalization.attachment.pk, self.attachment.pk) + self.assertEqual(attachment_finalization.file.read(), u'Default testing content') + self.assertEqual(attachment_finalization.successful, True) + + def test_attachment_argument_may_not_be_omitted(self): + with self.assertRaisesMessage(CommandError, u'attachment_anonymization takes at least 1 argument (0 given).'): + call_command(u'attachment_anonymization') + + def test_non_existing_attachment_raises_exception(self): + with self.assertRaisesMessage(CommandError, u'Attachment instance with pk "-1" does not exist.'): + call_command(u'attachment_anonymization', u'-1', self.filename) + + def test_invalid_attachment_id_raises_exception(self): + with self.assertRaisesMessage(CommandError, u'Attachment instance with pk "invalid_id" does not exist.'): + call_command(u'attachment_anonymization', u'invalid_id', self.filename) + + def test_command_with_too_many_arguments(self): + with self.assertRaisesMessage(CommandError, u'attachment_anonymization takes at most 2 arguments (3 given).'): + call_command(u'attachment_anonymization', self.attachment.pk, self.filename, u'filename2') + + def test_content_is_read_from_stdin_if_file_argument_is_omitted(self): + self.addCleanup(setattr, sys, u'stdin', sys.stdin) + sys.stdin = StringIO(u'Content from stdin.') + call_command(u'attachment_anonymization', self.attachment.pk) + attachment_finalization = AttachmentFinalization.objects.get(attachment=self.attachment) + self.assertEqual(attachment_finalization.file.read(), u'Content from stdin.') + + def test_file_argument_and_stdin_together_may_not_be_omitted(self): + self.addCleanup(setattr, sys, u'stdin', sys.stdin) + sys.stdin = StringIO(u'') + with self.assertRaisesMessage(CommandError, u'No content given.'): + call_command(u'attachment_anonymization', self.attachment.pk) + + def test_preferred_content_source_is_file(self): + self.addCleanup(setattr, sys, u'stdin', sys.stdin) + sys.stdin = StringIO(u'Content from stdin.') + call_command(u'attachment_anonymization', self.attachment.pk, self.filename) + attachment_finalization = AttachmentFinalization.objects.get(attachment=self.attachment) + self.assertEqual(attachment_finalization.file.read(), u'Default testing content') + + def test_invalid_file_raises_exception(self): + filename = u'/tmp/invalid.txt' + with self.assertRaisesMessage(CommandError, u'Could not open file: '.format(filename)): + call_command(u'attachment_anonymization', self.attachment.pk, filename) + + def test_content_type_option(self): + call_command(u'attachment_anonymization', self.attachment.pk, self.filename, content_type=u'application/pdf') + attachment_finalization = AttachmentFinalization.objects.get(attachment=self.attachment) + self.assertEqual(attachment_finalization.content_type, u'application/pdf') + + def test_content_type_option_default_value_if_omitted(self): + call_command(u'attachment_anonymization', self.attachment.pk, self.filename) + attachment_finalization = AttachmentFinalization.objects.get(attachment=self.attachment) + self.assertEqual(attachment_finalization.content_type, u'text/plain') + + def test_debug_option(self): + call_command(u'attachment_anonymization', self.attachment.pk, self.filename, debug=u'debug') + attachment_finalization = AttachmentFinalization.objects.get(attachment=self.attachment) + self.assertEqual(attachment_finalization.debug, u'debug') + + def test_debug_option_default_value_if_omitted(self): + call_command(u'attachment_anonymization', self.attachment.pk, self.filename) + attachment_finalization = AttachmentFinalization.objects.get(attachment=self.attachment) + self.assertEqual(attachment_finalization.debug, u'') + + def test_force_option(self): + call_command(u'attachment_anonymization', self.attachment.pk, self.filename) + attachment_finalization1 = AttachmentFinalization.objects.get(attachment=self.attachment) + call_command(u'attachment_anonymization', self.attachment.pk, self.filename, force=True) + attachment_finalization2 = AttachmentFinalization.objects.get(attachment=self.attachment) + self.assertNotEqual(attachment_finalization1.pk, attachment_finalization2.pk) + with self.assertRaisesMessage(AttachmentFinalization.DoesNotExist, u'AttachmentFinalization matching query does not exist'): + AttachmentFinalization.objects.get(pk=attachment_finalization1.pk) + + def test_existing_attachment_finalization_raises_exception_if_force_option_is_omitted(self): + call_command(u'attachment_anonymization', self.attachment.pk, self.filename) + attachment_finalization = AttachmentFinalization.objects.get(attachment=self.attachment) + with self.assertRaisesMessage(CommandError, u'Anonymization already exists. Use the --force to overwrite it.'): + call_command(u'attachment_anonymization', self.attachment.pk, self.filename) diff --git a/misc/anonymization.md b/misc/anonymization.md index 2807c28ba..03ba61e26 100644 --- a/misc/anonymization.md +++ b/misc/anonymization.md @@ -90,5 +90,27 @@ Properties: Computed Properties: * `content`: String; May be NULL; May be empty; Read-only. +## Commands + +### `attachment_anonymization` + + $ env/bin/python manage.py attachment_anonymization [options] attachment_id [file] + + +Creates AttachmentFinalization instance for the specified Attachment. The content source is file, +that can be passed as an argument, or stdin. Preferred source is file. If no file is specified and +stdin is empty, the command will fail. + +AttachmentFinalization created this way will be marked as successful. Only one successful +AttachmentFinalization can be assigned to the Attachment. + +* `--content_type=CONTENT_TYPE`: Content type of file, e.g. "application/pdf". Automatically + computed if not specified. +* `--debug=DEBUG`: Debug message to the newly created instance. Empty by default. +* `--force`: The command refuses to anonymize attachment if a successful anonymization already + exists. This flag disables this check. Deletes all existing successful + AttachmentFinalizations and creates new one. Unsuccessful AttachmentFinalizations will + stay unaffected. + *\* Features that are marked ~~strikethrough~~ are not implemented yet.*