Skip to content

Commit

Permalink
#342 Read content from stdin
Browse files Browse the repository at this point in the history
  • Loading branch information
@viliambalaz
viliambalaz committed Jan 21, 2021
1 parent 22db3c1 commit 537cd1d
Show file tree
Hide file tree
Showing 3 changed files with 72 additions and 44 deletions.
80 changes: 45 additions & 35 deletions chcemvediet/apps/anonymization/management/commands/attachment_anonymization.py
View file
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import sys
from optparse import make_option

import magic
Expand All @@ -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',
Expand All @@ -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'],
)
28 changes: 22 additions & 6 deletions chcemvediet/apps/anonymization/tests/test_management_commands.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
8 changes: 5 additions & 3 deletions misc/anonymization.md
View file
Original file line number Diff line number Diff line change
Expand Up @@ -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.
Expand Down

0 comments on commit 537cd1d

Please sign in to comment.