#342 Read content from stdin

chcemvediet/apps/anonymization/management/commands/attachment_anonymization.py

@@ -1,3 +1,4 @@
+import sys
 from optparse import make_option
 
 import magic
@@ -11,19 +12,19 @@
 
 
 class Command(BaseCommand):
- args = u'attachment [file]'
+ args = u'attachment_id [file]'
  help = squeeze(u"""
- Creates anonymization for the specified Attachment. By default, the file path is read
- from stdin. File path can be explicitly passed as an command argument. Anonymization
- created this way will be marked as successful. Only one successful anonymization can be
- assigned to the Attachment.
+ Creates anonymization for the specified Attachment. The content source is file, that can
+ be passed as an argument, or stdin. Preferred source is file. If file is not specified
+ and stdin is empty, the command will fail. Anonymization created this way will be marked
+ as successful. Only one successful anonymization can be assigned to the Attachment.
  """)
 
  option_list = BaseCommand.option_list + (
  make_option(u'--content_type',
  help=squeeze(u"""
- Content type of file, e.g. "application/pdf". Automatically computed if
- not specified.
+ Content type of file, e.g. "application/pdf". Automatically guessed from
+ the file content if not specified.
  """)
  ),
  make_option(u'--debug',
@@ -44,35 +45,44 @@ class Command(BaseCommand):
  def handle(self, *args, **options):
  if not args:
  raise CommandError(u'attachment_anonymization takes at least 1 argument (0 given).')
- if len(args) > 2:
+ elif len(args) > 2:
  raise CommandError(
  u'attachment_anonymization takes at most 2 arguments ({} given).'.format(len(args))
- )
- pk = args[0]
- filename = args[1] if len(args) == 2 else raw_input(u'File:')
- if not filename:
- raise CommandError(u'Missing source file name.')
- with open(filename, u'rb') as file:
+ )
+
+ 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(squeeze(u"""
+ Anonymization files already exist. Use the --force option to overwrite
+ them.
+ """))
+
+ if len(args) == 2:
+ filename = args[1]
  try:
- attachment = Attachment.objects.attached_to(Action).get(pk=pk)
- attachments_finalization = (AttachmentFinalization.objects
- .filter(attachment=attachment)
- .successful())
- if options[u'force'] or not attachments_finalization:
- attachments_finalization.delete()
+ with open(filename, u'rb') as file:
  content = file.read()
-  content_type = magic.from_buffer(content, mime=True)
-  AttachmentFinalization.objects.create(
-  attachment=attachment,
-  successful=True,
-  file=ContentFile(content),
-  content_type=options[u'content_type'] or content_type,
- debug=options[u'debug'],
-  )
-  else:
-  raise CommandError(squeeze(u"""
-  Anonymization files already exist. Use the --force option to overwrite
-  them.
-  """))
- except Attachment.DoesNotExist:
-  raise CommandError(u'Attachment instance with pk {} does not exist.'.format(pk))
+ except IOError as e:
+ raise CommandError(u'Could not open file: {}.'.format(e))
+ elif not sys.stdin.isatty():
+ content = sys.stdin.read()
+ else:
+ raise CommandError(u'Missing content source.')
+
+ 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/tests/test_management_commands.py

@@ -43,24 +43,40 @@ def test_attachment_argument_may_not_be_omitted(self):
  call_command(u'attachment_anonymization')
 
  def test_non_existent_attachment_raises_exception(self):
- with self.assertRaisesMessage(CommandError, u'Attachment instance with pk -1 does not exist.'):
+ 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_file_argument_is_read_from_stdin_if_omitted(self):
+ def test_content_is_read_from_stdin_if_file_argument_is_omitted(self):
  self.addCleanup(setattr, sys, u'stdin', sys.stdin)
- sys.stdin = StringIO(self.filename)
+ 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'\n')
- with self.assertRaisesMessage(CommandError, u'Missing source file name.'):
+ with self.assertRaisesMessage(CommandError, u'Missing content source.'):
  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_file_is_invalid(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)

misc/anonymization.md

@@ -90,14 +90,16 @@ Properties:
 Computed Properties:
 * `content`: String; May be NULL; May be empty; Read-only.
 
+## Commands
 
-## `attachment_anonymization`
+### `attachment_anonymization`
 
   $ env/bin/python manage.py attachment_anonymization [options] attachment_id [file]
 
 
-Creates AttachmentFinalization instance for the specified Attachment. By default, the file path is
-read from stdin. You can pass file path explicitly as an argument.
+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 file is not 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.