add asgi tests

Author: nggit

tests/asgi_server.py

@@ -119,15 +119,28 @@ async def app(scope, receive, send):
         # a response splitting attempt
         headers.append((b'referer', scope['path'].encode('utf-8')))
 
-    await send({
-        'type': 'http.response.start',
-        'status': 200,
-        'headers': headers
-    })
+    if scope['path'].startswith('/start'):
+        await send({
+            'type': 'http.response.start',
+            'status': 200,
+            'headers': headers
+        })
+
+    if not scope['path'].startswith('/body'):
+        await send({
+            'type': 'http.response.start',
+            'status': 200,
+            'headers': headers
+        })
+
+    if scope['path'].startswith('/invalid'):
+        await send({'type': 'invalid.message'})
+
     await send({
         'type': 'http.response.body',
         'body': b'Hello world!'
     })
 
 if __name__ == '__main__':
-    tremolo.run(app, host=ASGI_HOST, port=ASGI_PORT, debug=True, worker_num=2)
+    tremolo.run(app, host=ASGI_HOST, port=ASGI_PORT, debug=True, worker_num=2,
+                experimental=True, client_max_body_size=100 * 1048576)

tests/test_asgi_server.py

@@ -36,6 +36,39 @@ def test_app_exits_early(self):
                          b'HTTP/1.0 500 Internal Server Error')
         self.assertEqual(body, b'Internal Server Error')
 
+    def test_double_start(self):
+        header, body = getcontents(
+            host=ASGI_HOST,
+            port=ASGI_PORT,
+            raw=b'GET /start HTTP/1.0\r\n\r\n'
+        )
+
+        self.assertEqual(header[:header.find(b'\r\n')],
+                         b'HTTP/1.0 500 Internal Server Error')
+        self.assertEqual(body, b'response already started')
+
+    def test_body_before_start(self):
+        header, body = getcontents(
+            host=ASGI_HOST,
+            port=ASGI_PORT,
+            raw=b'GET /body HTTP/1.0\r\n\r\n'
+        )
+
+        self.assertEqual(header[:header.find(b'\r\n')],
+                         b'HTTP/1.0 500 Internal Server Error')
+        self.assertEqual(body, b'response not started')
+
+    def test_invalid_message_type(self):
+        header, body = getcontents(
+            host=ASGI_HOST,
+            port=ASGI_PORT,
+            raw=b'GET /invalid HTTP/1.0\r\n\r\n'
+        )
+
+        self.assertEqual(header[:header.find(b'\r\n')],
+                         b'HTTP/1.0 500 Internal Server Error')
+        self.assertEqual(body, b'invalid ASGI message type')
+
     def test_get_ok_10(self):
         header, body = getcontents(host=ASGI_HOST,
                                    port=ASGI_PORT,
@@ -136,9 +169,6 @@ def test_sec_response_splitting(self):
         self.assertEqual(header[:header.find(b'\r\n')],
                          b'HTTP/1.1 500 Internal Server Error')
         self.assertFalse(b'\r\nContent-Type: text/plain' in header)
-        self.assertTrue(
-            b'name or value cannot contain illegal characters' in body
-        )
 
     def test_websocket(self):
         header, body = getcontents(