Added unit test for unix sockets

tests/unittests/server_unix.py

@@ -0,0 +1,107 @@
+#!/usr/bin/env python
+# -*- coding: utf-8 -*-
+
+import fapws._evwsgi as evwsgi
+from fapws import base
+import time
+import sys
+from fapws.contrib import views, zip, log
+import mybase
+
+
+def env(environ, start_response):
+    print environ
+    start_response('200 OK', [('Content-Type','text/html')])
+    res=[]
+    for key,val in environ.items():
+        val=str(val).replace('\r','\\r')
+        val=val.replace('\n','\\n')
+        res.append("%s:%s\n" % (key,val))
+    return res
+
+def hello(environ, start_response):
+    start_response('200 OK', [('Content-Type','text/html')])
+    return ["Hello"," world!!"]
+
+def iteration(environ, start_response):
+    start_response('200 OK', [('Content-Type','text/plain')])
+    yield "Hello"
+    yield " "
+    yield "world!!"
+
+def tuplehello(environ, start_response):
+    start_response('200 OK', [('Content-Type','text/html')])
+    return ("Hello"," world!!")
+
+@log.Log()
+def staticlong(environ, start_response):
+    try:
+        f=open("long.txt", "rb")
+    except:
+        f=["Page not found"]
+    start_response('200 OK', [('Content-Type','text/html')])
+    return f
+
+def embedlong(environ, start_response):
+    try:
+        c=open("long.txt", "rb").read()
+    except:
+        c=["Page not found"]
+    start_response('200 OK', [('Content-Type','text/html')])
+    return base.split_len(c,32768)
+
+def staticshort(environ, start_response):
+    f=open("short.txt", "rb")
+    start_response('200 OK', [('Content-Type','text/html')])
+    return f
+
+def testpost(environ, start_response):
+    print environ
+    if "multipart/form-data" in environ['HTTP_CONTENT_TYPE']:
+        res=environ["wsgi.input"].getvalue()
+    elif "application/x-www-form-urlencoded" in environ['HTTP_CONTENT_TYPE']:
+        res=environ["fapws.params"]
+    else:
+        res={}
+    return ["OK. params are:%s" % (res)]
+
+@zip.Gzip()    
+def staticlongzipped(environ, start_response):
+    try:
+        f=open("long.txt", "rb")
+    except:
+        f=["Page not found"]
+    start_response('200 OK', [('Content-Type','text/html')])
+    return f
+
+def badscript(environ, start_response):
+    start_reponse('200 OK', [('Content-Type','text/html')])
+    return ["Hello world!!"]
+
+
+
+
+def start():
+    evwsgi.start("\0/org/fapws3/server", "unix")
+    evwsgi.set_base_module(mybase)
+
+
+    evwsgi.wsgi_cb(("/env", env))
+    evwsgi.wsgi_cb(("/hello", hello))
+    evwsgi.wsgi_cb(("/tuplehello", tuplehello))
+    evwsgi.wsgi_cb(("/iterhello", iteration))
+    evwsgi.wsgi_cb(("/longzipped", staticlongzipped))
+    evwsgi.wsgi_cb(("/long", staticlong))
+    evwsgi.wsgi_cb(("/elong", embedlong))
+    evwsgi.wsgi_cb(("/short", staticshort))
+    staticform=views.Staticfile("test.html")
+    evwsgi.wsgi_cb(("/staticform", staticform))
+    evwsgi.wsgi_cb(("/testpost", testpost))
+    evwsgi.wsgi_cb(("/badscript", badscript))
+
+    evwsgi.set_debug(0)    
+    evwsgi.run()
+
+
+if __name__=="__main__":
+    start()

tests/unittests/test_unix.py

@@ -0,0 +1,212 @@
+# -*- coding: utf-8 -*-
+import httplib
+import urllib
+import os.path
+
+import pycurl
+import os
+import socket
+
+successes=0
+failures=0
+
+
+def test(search, test, data):
+    global successes, failures
+    if not test:
+        print "TEST PROBLEM"
+        failures+=1
+    elif search not in data:
+        print """RESPONSE PROBLEM, we don't find "%s" """ % search
+        print data
+        failures+=1
+    else:
+        print "SUCCESS"
+        successes+=1
+
+class UHTTPConnection(httplib.HTTPConnection):
+    """Subclass of Python library HTTPConnection that
+       uses a unix-domain socket.
+       borrowed from http://7bits.nl/blog/2007/08/15/http-on-unix-sockets-with-python
+    """
+
+    def __init__(self, path):
+        httplib.HTTPConnection.__init__(self, 'localhost')
+        self.path = path
+
+    def connect(self):
+        sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+        sock.connect(self.path)
+        self.sock = sock
+
+con = UHTTPConnection("\0/org/fapws3/server")
+
+if 1:
+  print "=== Normal get ==="
+  con.request("GET", "/env/param?key=val")
+  response=con.getresponse()
+  content=response.read()
+  test("SCRIPT_NAME:/env",response.status==200,content) 
+  test("PATH_INFO:/param",response.status==200,content)
+  test("REQUEST_METHOD:GET",response.status==200,content) 
+  test("SERVER_PROTOCOL:HTTP/1.1",response.status==200,content) 
+  test("wsgi.url_scheme:HTTP",response.status==200,content) 
+  test("QUERY_STRING:key=val",response.status==200,content) 
+  test("fapws.params:{'key': ['val']}",response.status==200,content) 
+
+  print "=== URL not found ==="
+  con.request("GET", "/wrongpage")
+  response=con.getresponse()
+  content=response.read()
+  test("Page not found", response.status==500, content)
+
+  print "=== Get Hello world ==="
+  con.request("GET", "/hello")
+  response=con.getresponse()
+  content=response.read()
+  test("Hello world!!", response.status==200, content)
+
+  print "=== Get Iter Hello world ==="
+  con.request("GET", "/iterhello")
+  response=con.getresponse()
+  content=response.read()
+  test("Hello world!!", response.status==200, content)
+
+  print "=== Get tuple Hello world ==="
+  con.request("GET", "/tuplehello")
+  response=con.getresponse()
+  content=response.read()
+  test("Hello world!!", response.status==200, content)
+
+  print "=== Get long file ==="
+  con.request("GET", "/long")
+  response=con.getresponse()
+  content=response.read()
+  test("azerty", len(content)==os.path.getsize("long.txt"), content)
+
+  print "=== Get long file zipped ==="
+  headers={"Accept-Encoding":"gzip"}
+  con.request("GET", "/longzipped", "", headers)
+  response=con.getresponse()
+  content=response.read()
+  header=response.getheader('content-encoding')
+  test("azerty", header=="gzip", "azerty")
+
+  print "=== Get split long file ==="
+  con.request("GET", "/elong")
+  response=con.getresponse()
+  content=response.read()
+  test("azerty", len(content)==os.path.getsize("long.txt"), content)
+
+  print "=== Get cached file ==="
+  headers={"if-None-Match":str(os.path.getmtime('test.html'))}
+  con.request("GET", "/staticform", "", headers)
+  response=con.getresponse()
+  test("304", response.status==304, "304")
+
+  print "=== Post without length ==="
+  params = urllib.urlencode({'var1': 'value1', 'var2': 'value2'})
+  con.request("POST", "/testpost")
+  response=con.getresponse()
+  content=response.read()
+  test("Length Required", response.status==411, content)
+
+  print "=== Post with length ==="
+  params = urllib.urlencode({'var1': 'value1', 'var2': 'value2'})
+  headers = {"Content-type": "application/x-www-form-urlencoded", 
+            "Accept": "text/plain"}
+  con.request("POST", "/testpost", params, headers) #in this case httplib send automatically the content-length header
+  response=con.getresponse()
+  content=response.read()
+  test("OK. params are:{'var1': ['value1'], 'var2': ['value2']}", response.status==200, content)
+
+  print "=== Post with multipart (disabled for unix sockets) ==="
+  #pf = [('field1', 'this is a test using httppost & stuff'),
+  #    ('field2', (pycurl.FORM_FILE, 'short.txt'))
+  #   ]
+
+  #response=""
+  #def echo(data):
+  #  global response
+  #  response+=data
+
+  #fpath='/tmp/short.txt'
+  #if os.path.isfile(fpath):
+  #  os.remove(fpath) #we remove the data stored by the MultipartFormData
+
+  #c = pycurl.Curl()
+  #c.setopt(c.URL, 'http://127.0.0.1:8080/testpost')
+  #c.setopt(c.WRITEFUNCTION, echo)
+  #c.setopt(c.HTTPPOST, pf)
+  #c.perform()
+  #c.close()
+  #test("OK. params are:{'field2': ['/tmp/short.txt', {'Content-Type': 'text/plain', 'size': 14L}], 'field1': ['this is a test using httppost & stuff']}", 1==1, response)
+
+  print "=== Options ==="
+  con.request("OPTIONS", "/")
+  response=con.getresponse()
+  content=response.read()
+  resp_allow=response.getheader('allow', None)
+  test("Options", response.status==200 and resp_allow[:3]=='GET', "Options")
+
+  print "=== Bad header: 2 semi-column ==="
+  headers={'Badkey:': "Value"}
+  con.request("GET", "/env", "", headers)
+  response=con.getresponse()
+  content=response.read()
+  test("HTTP_BADKEY:: Value", response.status==200 , content)
+
+  print "=== Bad header: key with CR  ==="
+  headers={'Bad\nkey': "Value"}
+  con.request("GET", "/env", "", headers)
+  response=con.getresponse()
+  content=response.read()
+  test("KEY:Value", response.status==200 , content)
+
+  print "=== Bad header: value with CR  ==="
+  headers={'Badkey': "Val\nue"}
+  con.request("GET", "/env", "", headers)
+  response=con.getresponse()
+  content=response.read()
+  test("HTTP_BADKEY:Val", response.status==200 , content)
+
+  print "=== Bad header: value with CRLF  ==="
+  headers={'Badkey': "Val\r\nue"}
+  con.request("GET", "/env", "", headers)
+  response=con.getresponse()
+  content=response.read()
+  test("HTTP_BADKEY:Val", response.status==200 , content)
+
+  print "=== Bad command  ==="
+  con.request("GIT", "/env")
+  response=con.getresponse()
+  content=response.read()
+  test("Not Implemented", response.status==501 , content)
+
+  print "=== Bad first line  ==="
+  con.request("GET", "/env\r\n")
+  response=con.getresponse()
+  content=response.read()
+  test("SCRIPT_NAME:/env", response.status==200 , content)
+
+  print "=== Bad script  ==="
+  con.request("GET", "/badscript")
+  response=con.getresponse()
+  content=response.read()
+  test("Traceback", response.status==500 , content)
+
+  print "=== Bad command ==="
+  con.request("!çàù","")
+  response=con.getresponse()
+  content=response.read()
+  test("Not Implemented", response.status==501 , content)
+
+
+
+print "=================="
+print "TOTAL successes:", successes
+print "TOTAL failures:", failures
+print 
+print 
+
+