Test for invalid UTF-8

django_kepi/views.py

@@ -303,6 +303,11 @@ class InboxView(django.views.View):
                     status = 415, # unsupported media type
                     reason = 'Invalid JSON',
                     )
+        except UnicodeDecodeError:
+            return HttpResponse(
+                    status = 400, # bad request
+                    reason = 'Invalid UTF-8',
+                    )
 
         capture = django_kepi.validation.IncomingMessage(
                 date = request.META['HTTP_DATE'],

tests/test_inbox.py

@@ -80,7 +80,6 @@ class TestInbox(TestCase):
         self.assertFalse(
                 IncomingMessage.objects.all().exists())
 
-    @httpretty.activate
     def test_malformed_json(self):
 
         keys = json.load(open('tests/keys/keys-0001.json', 'r'))
@@ -89,6 +88,7 @@ class TestInbox(TestCase):
                 f_actor = REMOTE_FRED,
                 secret = keys['private'],
                 )
+        # we don't use the body it returns
 
         broken_json = json.dumps(body)[1:]
 
@@ -110,6 +110,36 @@ class TestInbox(TestCase):
         self.assertFalse(
                 IncomingMessage.objects.all().exists())
 
+    def test_invalid_utf8(self):
+
+        keys = json.load(open('tests/keys/keys-0001.json', 'r'))
+
+        body, headers = test_message_body_and_headers(
+                f_actor = REMOTE_FRED,
+                secret = keys['private'],
+                )
+        # we don't use the body it returns
+
+        invalid_utf8 = b"\xa0\xa1"
+
+        c = Client()
+        result = c.post(
+                path = INBOX_PATH,
+                content_type = headers['content-type'],
+                data = invalid_utf8,
+                HTTP_DATE = headers['date'],
+                HOST = headers['host'],
+                HTTP_SIGNATURE = headers['signature'],
+                )
+
+        self.assertEqual(
+                result.status_code,
+                400, # bad request
+                )
+
+        self.assertFalse(
+                IncomingMessage.objects.all().exists())
+
     @skip("broken; find out why")
     def test_all_parts_known(self):