#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

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'],
+        )

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')
 

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)

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.
+
 
 <sub>*\* Features that are marked ~~strikethrough~~ are not implemented yet.*</sub>