Tests: simple python application.

2 files changed, 134 insertions, 0 deletions

diff --git a/test/test_python_application.py b/test/test_python_application.py
new file mode 100644
index 00000000..a693002b
--- /dev/null
+++ b/test/test_python_application.py
@@ -0,0 +1,91 @@
+import unittest

+import unit

+

+class TestUnitApplication(unit.TestUnitControl):

+

+    def setUpClass():

+        u = unit.TestUnit()

+

+        u.check_modules('python')

+        u.check_version('0.4')

+

+    def test_python_application(self):

+        code, name = """

+

+def application(environ, start_response):

+

+    content_length = int(environ.get('CONTENT_LENGTH', 0))

+    body = bytes(environ['wsgi.input'].read(content_length))

+

+    start_response('200 OK', [

+        ('Content-Type', environ.get('CONTENT_TYPE')),

+        ('Content-Length', str(len(body))),

+        ('Request-Method', environ.get('REQUEST_METHOD')),

+        ('Request-Uri', environ.get('REQUEST_URI')),

+        ('Path-Info', environ.get('PATH_INFO')),

+        ('Http-Host', environ.get('HTTP_HOST')),

+        ('Remote-Addr', environ.get('REMOTE_ADDR')),

+        ('Server-Name', environ.get('SERVER_NAME')),

+        ('Server-Port', environ.get('SERVER_PORT')),

+        ('Server-Protocol', environ.get('SERVER_PROTOCOL')),

+        ('Custom-Header', environ.get('HTTP_CUSTOM_HEADER'))

+    ])

+    return [body]

+

+""", 'py_app'

+

+        self.python_application(name, code)

+

+        self.put('/', """

+            {

+                "listeners": {

+                    "*:7080": {

+                        "application": "app"

+                    }

+                },

+                "applications": {

+                    "app": {

+                        "type": "python",

+                        "workers": 1,

+                        "path": "%s",

+                        "module": "wsgi"

+                    }

+                }

+            }

+            """ % (self.testdir + '/' + name))

+

+        body = 'Test body string.'

+

+        r = unit.TestUnitHTTP.post(headers={

+            'Host': 'localhost',

+            'Content-Type': 'text/html',

+            'Content-Length': str(len(body)),

+            'Custom-Header': 'blah'

+        }, body=body)

+

+        self.assertEqual(r.status_code, 200, 'status')

+        self.assertEqual(r.headers['Content-Length'], str(len(body)),

+            'header content length')

+        self.assertEqual(r.headers['Content-Type'], 'text/html',

+            'header content type')

+        self.assertEqual(r.headers['Request-Method'], 'POST',

+            'header request method')

+        self.assertEqual(r.headers['Request-Uri'], '/', 'header request uri')

+        self.assertEqual(r.headers['Path-Info'], '/', 'header path info')

+        self.assertEqual(r.headers['Http-Host'], 'localhost',

+            'header http host')

+        self.assertEqual(r.headers['Remote-Addr'], '127.0.0.1',

+            'header remote addr')

+

+        self.assertTry('assertEqual', 'header server port',

+            r.headers['Server-Port'], '7080')

+

+        self.assertEqual(r.headers['Server-Protocol'], 'HTTP/1.1',

+            'header server protocol')

+        self.assertEqual(r.headers['Custom-Header'], 'blah',

+            'header custom header')

+        self.assertEqual(r.content, str.encode(body), 'body')

+

+

+if __name__ == '__main__':

+    unittest.main()

diff --git a/test/unit.py b/test/unit.py
index b510b42c..d3d00fe0 100644
--- a/test/unit.py
+++ b/test/unit.py
@@ -7,6 +7,7 @@ import shutil
 import socket

 import tempfile

 import unittest

+from requests import Request, Session

 from subprocess import call

 from multiprocessing import Process

 

@@ -94,6 +95,12 @@ class TestUnit(unittest.TestCase):
             self.testdir + '/unit.log', self.testdir + '/control.unit.sock'):

             exit("Could not start unit")

 

+    def python_application(self, name, code):

+        os.mkdir(self.testdir + '/' + name)

+

+        with open(self.testdir + '/' + name + '/wsgi.py', 'w') as f:

+            f.write(code)

+

     def _stop(self):

         with open(self.testdir + '/unit.pid', 'r') as f:

             pid = f.read().rstrip()

@@ -200,3 +207,39 @@ class TestUnitControl(TestUnit):
     def _body_json(self, resp):

         m = re.search('.*?\x0d\x0a?\x0d\x0a?(.*)', resp, re.M | re.S)

         return json.loads(m.group(1))

+

+class TestUnitHTTP():

+

+    @classmethod

+    def http(self, method, host='127.0.0.1:7080', uri='/', **kwargs):

+        if 'sess' in kwargs:

+            sess = kwargs['sess']

+        else:

+            sess = Session()

+

+        body = None

+        if 'body' in kwargs:

+            body = kwargs['body']

+

+        headers = None

+        if 'headers' in kwargs:

+            headers = kwargs['headers']

+

+        req = Request('POST', 'http://' + host + uri, headers=headers)

+        prepped = req.prepare()

+

+        prepped.body = body

+

+        r = sess.send(prepped)

+

+        if 'keep' not in kwargs:

+            sess.close()

+            return r

+

+        return (r, sess)

+

+    def get(**kwargs):

+        return TestUnitHTTP.http('GET', **kwargs)

+

+    def post(**kwargs):

+        return TestUnitHTTP.http('POST', **kwargs)