Added an API endpoint to support automation of IESG ballot position posting, at /api/iesg/position. Added tests for the API endpoint, and updated the apikey validation decorator tests. Tweaked the decorator to handle a weakness found during testing.

 - Legacy-Id: 14429

Author: levkowetz

ietf/doc/tests_ballot.py

@@ -12,7 +12,7 @@
 from ietf.group.models import Group, Role
 from ietf.name.models import BallotPositionName
 from ietf.iesg.models import TelechatDate
-from ietf.person.models import Person
+from ietf.person.models import Person, PersonalApiKey
 from ietf.utils.test_utils import TestCase, unicontent
 from ietf.utils.mail import outbox, empty_outbox
 from ietf.utils.test_data import make_test_data
@@ -82,6 +82,65 @@ def test_edit_position(self):
         self.assertEqual(draft.docevent_set.count(), events_before + 2)
         self.assertTrue("Ballot comment text updated" in pos.desc)
 
+    def test_api_set_position(self):
+        draft = make_test_data()
+        url = urlreverse('ietf.doc.views_ballot.api_set_position')
+        ad = Person.objects.get(name="Areað Irector")
+        ad.user.last_login = datetime.datetime.now()
+        ad.user.save()
+        apikey = PersonalApiKey.objects.create(endpoint=url, person=ad)
+
+        # vote
+        events_before = draft.docevent_set.count()
+        
+        r = self.client.post(url, dict(
+                                        apikey=apikey.hash(),
+                                        doc=draft.name,
+                                        position="discuss",
+                                        discuss=" This is a discussion test. \n ",
+                                        comment=" This is a test. \n ")
+            )
+        self.assertEqual(r.status_code, 200)
+
+        pos = draft.latest_event(BallotPositionDocEvent, ad=ad)
+        self.assertEqual(pos.pos.slug, "discuss")
+        self.assertTrue(" This is a discussion test." in pos.discuss)
+        self.assertTrue(pos.discuss_time != None)
+        self.assertTrue(" This is a test." in pos.comment)
+        self.assertTrue(pos.comment_time != None)
+        self.assertTrue("New position" in pos.desc)
+        self.assertEqual(draft.docevent_set.count(), events_before + 3)
+
+        # recast vote
+        events_before = draft.docevent_set.count()
+        r = self.client.post(url, dict(apikey=apikey.hash(), doc=draft.name, position="noobj"))
+        self.assertEqual(r.status_code, 200)
+
+        pos = draft.latest_event(BallotPositionDocEvent, ad=ad)
+        self.assertEqual(pos.pos.slug, "noobj")
+        self.assertEqual(draft.docevent_set.count(), events_before + 1)
+        self.assertTrue("Position for" in pos.desc)
+        
+        # clear vote
+        events_before = draft.docevent_set.count()
+        r = self.client.post(url, dict(apikey=apikey.hash(), doc=draft.name, position="norecord"))
+        self.assertEqual(r.status_code, 200)
+
+        pos = draft.latest_event(BallotPositionDocEvent, ad=ad)
+        self.assertEqual(pos.pos.slug, "norecord")
+        self.assertEqual(draft.docevent_set.count(), events_before + 1)
+        self.assertTrue("Position for" in pos.desc)
+
+        # change comment
+        events_before = draft.docevent_set.count()
+        r = self.client.post(url, dict(apikey=apikey.hash(), doc=draft.name, position="norecord", comment="New comment."))
+        self.assertEqual(r.status_code, 200)
+
+        pos = draft.latest_event(BallotPositionDocEvent, ad=ad)
+        self.assertEqual(pos.pos.slug, "norecord")
+        self.assertEqual(draft.docevent_set.count(), events_before + 2)
+        self.assertTrue("Ballot comment text updated" in pos.desc)
+
     def test_edit_position_as_secretary(self):
         draft = make_test_data()
         url = urlreverse('ietf.doc.views_ballot.edit_position', kwargs=dict(name=draft.name,

ietf/doc/views_ballot.py

@@ -7,6 +7,7 @@
 from django.conf import settings
 from django.http import HttpResponse, HttpResponseForbidden, HttpResponseRedirect, Http404
 from django.shortcuts import render, get_object_or_404, redirect
+from django.template.defaultfilters import striptags
 from django.template.loader import render_to_string
 from django.urls import reverse as urlreverse
 from django.views.decorators.csrf import csrf_exempt
@@ -108,6 +109,74 @@ def clean_discuss(self):
            raise forms.ValidationError("You must enter a non-empty discuss")
        return entered_discuss
 
+def save_position(form, doc, ballot, ad, login=None):
+    # save the vote
+    if login is None:
+        login = ad
+    clean = form.cleaned_data
+
+    old_pos = doc.latest_event(BallotPositionDocEvent, type="changed_ballot_position", ad=ad, ballot=ballot)
+    pos = BallotPositionDocEvent(doc=doc, rev=doc.rev, by=login)
+    pos.type = "changed_ballot_position"
+    pos.ballot = ballot
+    pos.ad = ad
+    pos.pos = clean["position"]
+    pos.comment = clean["comment"].rstrip()
+    pos.comment_time = old_pos.comment_time if old_pos else None
+    pos.discuss = clean["discuss"].rstrip()
+    if not pos.pos.blocking:
+        pos.discuss = ""
+    pos.discuss_time = old_pos.discuss_time if old_pos else None
+
+    changes = []
+    added_events = []
+    # possibly add discuss/comment comments to history trail
+    # so it's easy to see what's happened
+    old_comment = old_pos.comment if old_pos else ""
+    if pos.comment != old_comment:
+        pos.comment_time = pos.time
+        changes.append("comment")
+
+        if pos.comment:
+            e = DocEvent(doc=doc, rev=doc.rev)
+            e.by = ad # otherwise we can't see who's saying it
+            e.type = "added_comment"
+            e.desc = "[Ballot comment]\n" + pos.comment
+            added_events.append(e)
+
+    old_discuss = old_pos.discuss if old_pos else ""
+    if pos.discuss != old_discuss:
+        pos.discuss_time = pos.time
+        changes.append("discuss")
+
+        if pos.pos.blocking:
+            e = DocEvent(doc=doc, rev=doc.rev, by=login)
+            e.by = ad # otherwise we can't see who's saying it
+            e.type = "added_comment"
+            e.desc = "[Ballot %s]\n" % pos.pos.name.lower()
+            e.desc += pos.discuss
+            added_events.append(e)
+
+    # figure out a description
+    if not old_pos and pos.pos.slug != "norecord":
+        pos.desc = u"[Ballot Position Update] New position, %s, has been recorded for %s" % (pos.pos.name, pos.ad.plain_name())
+    elif old_pos and pos.pos != old_pos.pos:
+        pos.desc = "[Ballot Position Update] Position for %s has been changed to %s from %s" % (pos.ad.plain_name(), pos.pos.name, old_pos.pos.name)
+
+    if not pos.desc and changes:
+        pos.desc = u"Ballot %s text updated for %s" % (u" and ".join(changes), ad.plain_name())
+
+    # only add new event if we actually got a change
+    if pos.desc:
+        if login != ad:
+            pos.desc += u" by %s" % login.plain_name()
+
+        pos.save()
+
+        for e in added_events:
+            e.save() # save them after the position is saved to get later id for sorting order
+    
+
 @role_required('Area Director','Secretariat')
 def edit_position(request, name, ballot_id):
     """Vote and edit discuss and comment on document as Area Director."""
@@ -128,77 +197,14 @@ def edit_position(request, name, ballot_id):
             raise Http404
         ad = get_object_or_404(Person, pk=ad_id)
 
-    old_pos = doc.latest_event(BallotPositionDocEvent, type="changed_ballot_position", ad=ad, ballot=ballot)
-
     if request.method == 'POST':
         if not has_role(request.user, "Secretariat") and not ad.role_set.filter(name="ad", group__type="area", group__state="active"):
             # prevent pre-ADs from voting
             return HttpResponseForbidden("Must be a proper Area Director in an active area to cast ballot")
 
         form = EditPositionForm(request.POST, ballot_type=ballot.ballot_type)
         if form.is_valid():
-            # save the vote
-            clean = form.cleaned_data
-
-            pos = BallotPositionDocEvent(doc=doc, rev=doc.rev, by=login)
-            pos.type = "changed_ballot_position"
-            pos.ballot = ballot
-            pos.ad = ad
-            pos.pos = clean["position"]
-            pos.comment = clean["comment"].rstrip()
-            pos.comment_time = old_pos.comment_time if old_pos else None
-            pos.discuss = clean["discuss"].rstrip()
-            if not pos.pos.blocking:
-                pos.discuss = ""
-            pos.discuss_time = old_pos.discuss_time if old_pos else None
-
-            changes = []
-            added_events = []
-            # possibly add discuss/comment comments to history trail
-            # so it's easy to see what's happened
-            old_comment = old_pos.comment if old_pos else ""
-            if pos.comment != old_comment:
-                pos.comment_time = pos.time
-                changes.append("comment")
-
-                if pos.comment:
-                    e = DocEvent(doc=doc, rev=doc.rev)
-                    e.by = ad # otherwise we can't see who's saying it
-                    e.type = "added_comment"
-                    e.desc = "[Ballot comment]\n" + pos.comment
-                    added_events.append(e)
-
-            old_discuss = old_pos.discuss if old_pos else ""
-            if pos.discuss != old_discuss:
-                pos.discuss_time = pos.time
-                changes.append("discuss")
-
-                if pos.pos.blocking:
-                    e = DocEvent(doc=doc, rev=doc.rev, by=login)
-                    e.by = ad # otherwise we can't see who's saying it
-                    e.type = "added_comment"
-                    e.desc = "[Ballot %s]\n" % pos.pos.name.lower()
-                    e.desc += pos.discuss
-                    added_events.append(e)
-
-            # figure out a description
-            if not old_pos and pos.pos.slug != "norecord":
-                pos.desc = u"[Ballot Position Update] New position, %s, has been recorded for %s" % (pos.pos.name, pos.ad.plain_name())
-            elif old_pos and pos.pos != old_pos.pos:
-                pos.desc = "[Ballot Position Update] Position for %s has been changed to %s from %s" % (pos.ad.plain_name(), pos.pos.name, old_pos.pos.name)
-
-            if not pos.desc and changes:
-                pos.desc = u"Ballot %s text updated for %s" % (u" and ".join(changes), ad.plain_name())
-
-            # only add new event if we actually got a change
-            if pos.desc:
-                if login != ad:
-                    pos.desc += u" by %s" % login.plain_name()
-
-                pos.save()
-
-                for e in added_events:
-                    e.save() # save them after the position is saved to get later id for sorting order
+            save_position(form, doc, ballot, ad, login)
 
             if request.POST.get("send_mail"):
                 qstr=""
@@ -213,6 +219,7 @@ def edit_position(request, name, ballot_id):
                 return HttpResponseRedirect(return_to_url)
     else:
         initial = {}
+        old_pos = doc.latest_event(BallotPositionDocEvent, type="changed_ballot_position", ad=ad, ballot=ballot)
         if old_pos:
             initial['position'] = old_pos.pos.slug
             initial['discuss'] = old_pos.discuss
@@ -240,10 +247,37 @@ def edit_position(request, name, ballot_id):
 @role_required('Area Director', 'Secretariat')
 @csrf_exempt
 def api_set_position(request):
+    def err(code, text):
+        return HttpResponse(text, status=code, content_type='text/plain')
     if request.method == 'POST':
-        pass
+        ad = request.user.person
+        name = request.POST.get('doc')
+        if not name:
+            return err(400, "Missing document name")
+        try:
+            doc = Document.objects.get(docalias__name=name)
+        except Document.DoesNotExist:
+            return err(404, "Document not found")
+        position_names = BallotPositionName.objects.values_list('slug', flat=True)
+        position = request.POST.get('position')
+        if not position:
+            return err(400, "Missing parameter: position, one of: %s " % ','.join(position_names))
+        if not position in position_names:
+            return err(400, "Bad position name, must be one of: %s " % ','.join(position_names))
+        ballot = doc.active_ballot()
+        if not ballot:
+            return err(404, "No open ballot found")
+        form = EditPositionForm(request.POST, ballot_type=ballot.ballot_type)
+        if form.is_valid():
+            save_position(form, doc, ballot, ad)
+        else:
+            debug.type('form.errors')
+            debug.show('form.errors')
+            errors = form.errors
+            summary = ','.join([ "%s: %s" % (f, striptags(errors[f])) for f in errors ])
+            return err(400, "Form not valid: %s" % summary)
     else:
-        return HttpResponse("Method not allowed", status=405, content_type='text/plain')
+        return err(405, "Method not allowed")
 
     return HttpResponse("Done", status=200, content_type='text/plain')
 

ietf/ietfauth/tests.py

@@ -1,5 +1,6 @@
+# Copyright The IETF Trust 2017, All Rights Reserved
 # -*- coding: utf-8 -*-
-from __future__ import unicode_literals
+from __future__ import unicode_literals, print_function
 
 import os, shutil, time, datetime
 from urlparse import urlsplit
@@ -101,7 +102,7 @@ def username_in_htpasswd_file(self, username):
                 if l.startswith(username + ":"):
                     return True
         with open(settings.HTPASSWD_FILE) as f:
-            print f.read()
+            print(f.read())
 
         return False
 
@@ -552,7 +553,7 @@ def test_apikey_management(self):
         self.assertEqual(len(q('td code')), len(PERSON_API_KEY_ENDPOINTS)) # key hash
         self.assertEqual(len(q('td a:contains("Disable")')), len(PERSON_API_KEY_ENDPOINTS)-1)
 
-    def test_apikey_usage(self):
+    def test_apikey_errors(self):
         BAD_KEY = "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
 
         person = PersonFactory()
@@ -571,10 +572,6 @@ def test_apikey_usage(self):
         for key in person.apikeys.all()[:3]:
             url = key.endpoint
 
-            # successful access
-            r = self.client.post(url, {'apikey':key.hash(), 'dummy':'dummy',})
-            self.assertEqual(r.status_code, 200)
-
             # bad method
             r = self.client.put(url, {'apikey':key.hash()})
             self.assertEqual(r.status_code, 405)
@@ -638,7 +635,7 @@ def test_send_apikey_report(self):
 
         self.assertEqual(len(outbox), len(PERSON_API_KEY_ENDPOINTS))
         for mail in outbox:
-            body = mail.get_payload()
+            body = mail.get_payload(decode=True).decode('utf-8')
             self.assertIn("API key usage", mail['subject'])
             self.assertIn(" %s times" % count, body)
             self.assertIn(date, body)

ietf/utils/decorators.py

@@ -60,7 +60,7 @@ def err(code, text):
     person = key.person
     last_login = person.user.last_login
     time_limit = (datetime.datetime.now() - datetime.timedelta(days=settings.UTILS_APIKEY_GUI_LOGIN_LIMIT_DAYS))
-    if last_login < time_limit:
+    if last_login == None or last_login < time_limit:
         return err(400, "Too long since last regular login")
     # Log in
     login(request, person.user)