Refactoring the credential importing

staff/templates/staff_import_overview.html

@@ -0,0 +1,29 @@
+{% extends "base.html" %}
+{% load url from future %}
+{% load i18n %}
+
+{% block content %}
+<h1>{% trans "Import Overview" %}</h1>
+<table class="table table-striped table-bordered table-condensed">
+<thead>
+<tr>
+    <th>{% trans "Actions" %}</th>
+    <th>{% trans "Title" %}</th>
+    <th>{% trans "User" %}</th>
+</tr>
+</thead>
+<tbody>
+{% for e in entries %}
+<tr>
+    <td>
+        <a class="btn btn-mini btn-success" href="{% url "staff.views.import_process" forloop.counter0 %}">{% trans "Import" %}</a>
+        <a class="btn btn-mini btn-danger" href="{% url "staff.views.import_ignore" forloop.counter0 %}">{% trans "Skip" %}</a>
+    </td>
+    <td>{{ e.title }}</td>
+    <td>{{ e.username }}</td>
+</tr>
+{% endfor %}
+</tbody>
+</table>
+{% endblock %}
+

staff/templates/staff_process_import.html → staff/templates/staff_import_process.html

@@ -6,12 +6,10 @@
 {% block content %}
 
 {% include "cred_part_passgenmodal.html" %}
-{% include "cred_part_logomodal.html" with icons=icons txtfield='iconname' imgfield='logodisplay' %}
-<h1>{% trans "Import" %}</h1>
+{% include "cred_part_logomodal.html" with icons=icons txtfield='#id_iconname' imgfield='#logodisplay' %}
+<h1>{% trans "Password Import" %}</h1>
 
-<p>{% blocktrans %}There are {{count}} passwords left to import. If you wish you can{% endblocktrans %} <a href="{% url 'staff.views.process_import' %}">{% trans "skip importing this password" %}</a>.</p>
-
-<form class="form-horizontal" enctype="multipart/form-data" autocomplete="off" action="{{ action }}{% if next %}?next={{ next }}{% endif %}" method="post">{% csrf_token %}
+<form class="form-horizontal" enctype="multipart/form-data" autocomplete="off" method="post">{% csrf_token %}
     {% for field in form %}
         <div class="control-group">
             {{ field.errors }}

staff/tests/test_import.py

@@ -4,6 +4,7 @@
 from django.test.utils import override_settings
 
 from ratticweb.tests.helper import TestData
+from cred.models import Cred
 
 
 class ImportTests(TestCase):
@@ -41,17 +42,19 @@ def test_upload_keepass(self):
         self.assertEqual(data['group'], self.gp.id)
 
         # Check the right credentials are in there
-        cred = data['entries'][0]
-        attcred = data['entries'][3]
+        cred = filter(lambda x: x['title'] == 'dans id', data['entries'])[0]
         self.assertEqual(cred['title'], 'dans id')
         self.assertEqual(cred['password'], 'CeidAcHuhy')
         self.assertEqual(sorted(cred['tags']), sorted(['Internet', 'picasa.com']))
+
+        # Check for attachments
+        attcred = filter(lambda x: x['title'] == 'Attachment Test', data['entries'])[0]
         self.assertEqual(attcred['filename'], 'test.txt')
         self.assertEqual(attcred['filecontent'], 'This is a test file.\n')
 
     def test_process_import_no_data(self):
         # With no data we expect a 404
-        resp = self.data.staff.get(reverse('staff.views.process_import'))
+        resp = self.data.staff.get(reverse('staff.views.import_overview'))
         self.assertEqual(resp.status_code, 404)
 
     def test_process_import_last_entry(self):
@@ -64,13 +67,13 @@ def test_process_import_last_entry(self):
         session.save()
 
         # Try to import
-        resp = self.data.staff.get(reverse('staff.views.process_import'))
+        resp = self.data.staff.get(reverse('staff.views.import_overview'))
 
         # Check we were redirected home
         self.assertRedirects(resp, reverse('staff.views.home'), 302, 200)
         self.assertNotIn('imported_data', self.data.staff.session)
 
-    def test_process_import_entry_import(self):
+    def test_process_import_entry_ignore(self):
         # Setup session test data
         entry = {
             'title': 'Test',
@@ -82,6 +85,31 @@ def test_process_import_entry_import(self):
             'filename': None,
             'filecontent': '',
         }
+        session = self.data.staff.session
+        session['imported_data'] = {}
+        session['imported_data']['group'] = self.gp.id
+        session['imported_data']['entries'] = [entry, ]
+        session.save()
+
+        # Load the import screen
+        resp = self.data.staff.get(reverse('staff.views.import_ignore', args=(0, )))
+
+        # Check things worked
+        self.assertRedirects(resp, reverse('staff.views.import_overview'), 302, 302)
+        self.assertNotIn('imported_data', self.data.staff.session)
+
+    def test_process_import_entry_import(self):
+        # Setup session test data
+        entry = {
+            'title': 'TestImportFunction',
+            'username': 'dan',
+            'description': '',
+            'password': 'pass',
+            'url': 'http://example.com/',
+            'tags': ['tag1', 'tag2'],
+            'filename': None,
+            'filecontent': '',
+        }
         entries = [entry, ]
         session = self.data.staff.session
         session['imported_data'] = {}
@@ -90,12 +118,27 @@ def test_process_import_entry_import(self):
         session.save()
 
         # Load the import screen
-        resp = self.data.staff.get(reverse('staff.views.process_import'))
+        resp = self.data.staff.get(reverse('staff.views.import_process', args=(0, )))
 
         # Check things worked
-        self.assertIn('imported_data', self.data.staff.session)
-        self.assertEquals(len(self.data.staff.session['imported_data']['entries']), 0)
-        self.assertTemplateUsed(resp, 'staff_process_import.html')
+        self.assertTemplateUsed(resp, 'staff_import_process.html')
+        self.assertTrue(resp.context['form'].is_valid())
+
+        # Perform the save
+        post = {}
+        for i in resp.context['form']:
+            if i.value() is not None:
+                post[i.name] = i.value()
+        resp = self.data.staff.post(
+            reverse('staff.views.import_process', args=(0, )),
+            post,
+        )
+
+        # Test the results
+        self.assertRedirects(resp, reverse('staff.views.import_overview'), 302, 302)
+        c = Cred.objects.get(title='TestImportFunction')
+        self.assertEquals(c.url, 'http://example.com/')
+        self.assertEquals(c.password, 'pass')
 
 
 ImportTests = override_settings(PASSWORD_HASHERS=('django.contrib.auth.hashers.MD5PasswordHasher',))(ImportTests)

staff/urls.py

@@ -5,23 +5,36 @@
 urlpatterns = patterns('staff.views',
     # Views in views.py
     url(r'^$', 'home'),
+
+    # User/Group Management
     url(r'^userdetail/(?P<uid>\d+)/$', 'userdetail'),
     url(r'^removetoken/(?P<uid>\d+)/$', 'removetoken'),
     url(r'^groupdetail/(?P<gid>\d+)/$', 'groupdetail'),
+
+    # Auditing
     url(r'^audit-by-cred/(?P<cred_id>\d+)/$', 'audit_by_cred'),
     url(r'^audit-by-user/(?P<user_id>\d+)/$', 'audit_by_user'),
     url(r'^audit-by-days/(?P<days_ago>\d+)/$', 'audit_by_days'),
+
+    # Importing
     url(r'^import/keepass/$', 'upload_keepass'),
-    url(r'^import/process/$', 'process_import'),
+    url(r'^import/process/$', 'import_overview'),
+    url(r'^import/process/(?P<import_id>\d+)/$', 'import_process'),
+    url(r'^import/process/(?P<import_id>\d+)/ignore/$', 'import_ignore'),
+
+    # Undeletion
     url(r'^credundelete/(?P<cred_id>\d+)/$', 'credundelete'),
 )
 
 # URLs that we don't want with LDAP
 if not settings.LDAP_ENABLED:
     urlpatterns += patterns('staff.views',
+        # Group Management
         url(r'^groupadd/$', 'groupadd'),
         url(r'^groupedit/(?P<gid>\d+)/$', 'groupedit'),
         url(r'^groupdelete/(?P<gid>\d+)/$', 'groupdelete'),
+
+        # User Management
         url(r'^useradd/$', NewUser.as_view(), name="user_add"),
         url(r'^useredit/(?P<pk>\d+)/$', UpdateUser.as_view(), name="user_edit"),
         url(r'^userdelete/(?P<uid>\d+)/$', 'userdelete'),

staff/views.py

@@ -225,89 +225,128 @@ def upload_keepass(request):
             request.session['imported_data'] = data
 
             # Start the user processing entries
-            return HttpResponseRedirect(reverse('staff.views.process_import'))
+            return HttpResponseRedirect(reverse('staff.views.import_overview'))
     else:
         form = KeepassImportForm(request.user)
     return render(request, 'staff_keepassimport.html', {'form': form})
 
 
-def process_import(request):
+@staff_member_required
+def import_overview(request):
     # If there was no session data, return 404
     if 'imported_data' not in request.session.keys():
         raise Http404
 
-    # If there are no creds left to import
-    if len(request.session['imported_data']['entries']) == 0:
-        # Clear data and go back to staff home
+    # Get the entries to import
+    entries = request.session['imported_data']['entries']
+
+    # If there is nothing left, go back home
+    if len(entries) == 0:
         del request.session['imported_data']
+        request.session.save()
         return HttpResponseRedirect(reverse('staff.views.home'))
 
-    # If we have a submission from the user
+    return render(request, 'staff_import_overview.html', {
+        'entries': entries,
+    })
+
+
+@staff_member_required
+def import_ignore(request, import_id):
+    # If there was no session data, return 404
+    if 'imported_data' not in request.session.keys():
+        raise Http404
+
+    # Get the entry we are concerned with
+    try:
+        del request.session['imported_data']['entries'][int(import_id)]
+        request.session.save()
+    except IndexError:
+        raise Http404
+
+    return HttpResponseRedirect(reverse('staff.views.import_overview'))
+
+
+@staff_member_required
+def import_process(request, import_id):
+    # If there was no session data, return 404
+    if 'imported_data' not in request.session.keys():
+        raise Http404
+
+    # Get the entry we are concerned with
+    try:
+        entry = request.session['imported_data']['entries'][int(import_id)]
+    except IndexError:
+        raise Http404
+
+    # Get the group
+    groupid = request.session['imported_data']['group']
+    try:
+        group = Group.objects.get(pk=groupid)
+    except Group.DoesNotExist:
+        del request.session['imported_data']
+        raise Http404
+
     if request.method == 'POST':
-        files = request.FILES
+        # Try and import what we have now
+
+        # Did the user upload a new attachment
+        if entry['filename'] and 'attachment' not in request.FILES.keys():
+            sfile = SimpleUploadedFile(entry['filename'], bytes(entry['filecontent']))
+            request.FILES['attachment'] = sfile
 
-        if 'imported_attachment' in request.session.keys() and 'attachment' not in files.keys():
-            fdata = request.session['imported_attachment']
-            sfile = SimpleUploadedFile(fdata['filename'], bytes(fdata['content']))
-            files['attachment'] = sfile
+        # Build the form
+        form = CredForm(request.user, request.POST, request.FILES)
 
-        form = CredForm(request.user, request.POST, files)
+        # Do we have enough data to save?
         if form.is_valid():
-            # Save the new credential
-            form.save()
 
-            if 'imported_attachment' in request.session.keys():
-                del request.session['imported_attachment']
+            # Save the credential
+            form.save()
 
-            # Add an audit record
+            # Write the audit log
             CredAudit(
                 audittype=CredAudit.CREDADD,
                 cred=form.instance,
                 user=request.user,
             ).save()
 
-            # Import another
-            return HttpResponseRedirect(reverse('staff.views.process_import'))
+            # Remove the entry we're importing
+            del request.session['imported_data']['entries'][int(import_id)]
+            request.session.save()
+
+            # Go back to the overview
+            return HttpResponseRedirect(reverse('staff.views.import_overview'))
 
-    # If we didn't recieve any data
     else:
-        # Get a new entry
-        newcred = request.session['imported_data']['entries'].pop()
-        request.session.save()
+        # Init the cred, and create the form
+        processed = dict(entry)
 
         # Create all the tags
         tlist = []
-        for t in newcred['tags']:
+        for t in processed['tags']:
             (tag, create) = Tag.objects.get_or_create(name=t)
             tlist.append(tag)
-        newcred['tags'] = tlist
+        processed['tags'] = tlist
 
         # Setup the group
-        groupid = request.session['imported_data']['group']
-        try:
-            newcred['group'] = Group.objects.get(pk=groupid)
-        except Group.DoesNotExist:
-            del request.session['imported_data']
-            raise Http404
-
-        # Move any attachment temporarily
-        if newcred['filename']:
-            request.session['imported_attachment'] = {
-                'filename': newcred['filename'],
-                'content': newcred['filecontent'],
-            }
-            del newcred['filename']
-            del newcred['filecontent']
+        processed['group'] = group
+
+        # If the icon is empty set it
+        if 'iconname' not in processed.keys():
+            processed['iconname'] = 'Key.png'
+
+        # Remove the attachment
+        if processed['filename']:
+            del processed['filename']
+            del processed['filecontent']
 
-        # Display the form
-        form = CredForm(request.user, newcred, {})
+        # Create the form
+        form = CredForm(request.user, processed, {})
 
-    # Display the edit form
-    return render(request, 'staff_process_import.html', {
+    return render(request, 'staff_import_process.html', {
         'form': form,
-        'action': reverse('staff.views.process_import'),
         'icons': get_icon_list(),
-        'count': len(request.session['imported_data']['entries']),
     })
 
 

tests/runner.py

@@ -1,11 +1,16 @@
 from django_nose import NoseTestSuiteRunner
 from django.conf import settings
 
-# South debug logs are irritating
 import logging
+
+# South debug logs are irritating
 south_logger=logging.getLogger('south')
 south_logger.setLevel(logging.INFO)
 
+# So are the KeepassDB ones
+keepassdb_logger=logging.getLogger('keepassdb')
+keepassdb_logger.setLevel(logging.INFO)
+
 
 class ExcludeAppsTestSuiteRunner(NoseTestSuiteRunner):
     """Override the default django 'test' command, exclude from testing