[tests] add scgi tests

personal/stbuehler/wip
Stefan Bühler 9 years ago
parent ff7b73c8a9
commit 20e0eb88b6
  1. 3
      tests/base.py
  2. 2
      tests/requests.py
  3. 131
      tests/t-scgi.py

@ -383,6 +383,9 @@ instance {{
binary "{Env.worker}";
config "{Env.lighttpdconf}";
modules "{Env.plugindir}";
# env ( "G_SLICE=always-malloc", "G_DEBUG=gc-friendly" );
# wrapper ("/usr/bin/valgrind", "--leak-check=full", "--show-reachable=yes", "--leak-resolution=high" );
}}
allow-listen {{ ip "127.0.0.1:{Env.port}"; }}

@ -54,7 +54,7 @@ class CurlRequest(TestBase):
c.setopt(pycurl.URL, self.SCHEME + ("://127.0.0.1:%i" % (Env.port + self.PORT)) + self.URL)
c.setopt(pycurl.HTTPHEADER, ["Host: " + self.vhost])
c.setopt(pycurl.NOSIGNAL, 1)
c.setopt(pycurl.TIMEOUT, 10)
c.setopt(pycurl.TIMEOUT, 2)
b = StringIO.StringIO()
c.setopt(pycurl.WRITEFUNCTION, b.write)
c.setopt(pycurl.HEADERFUNCTION, self._recv_header)

@ -0,0 +1,131 @@
# -*- coding: utf-8 -*-
from base import GroupTest
from requests import CurlRequest
from service import Service
import socket
import os
import base
SCGI_ENVCHECK="""#!/usr/bin/env python
# -*- coding: utf-8 -*-
import socket
import traceback
servsocket = socket.fromfd(0, socket.AF_UNIX, socket.SOCK_STREAM)
def parsereq(data):
a = data.split(':', 1)
if len(a) != 2: return False
hlen, rem = a
hlen = int(hlen)
if hlen < 16: raise Exception("invalid request")
if len(rem) < hlen + 1: return False
if rem[hlen] != ',' or rem[hlen-1] != '\\0': raise Exception("invalid request")
header = rem[:hlen-1]
body = rem[hlen+1:]
header = header.split('\\0')
if len(header) < 4: raise Exception("invalid request: not enough header entries")
if header[0] != "CONTENT_LENGTH": raise Exception("invalid request: missing CONTENT_LENGTH")
clen = int(header[1])
if len(body) < clen: return False
env = dict()
while len(header) > 0:
if len(header) == 1: raise Exception("invalid request: missing value for key")
key, value = header[0:2]
header = header[2:]
if '' == key: raise Exception("invalid request: empty key")
if env.has_key(key): raise Exception("invalid request: duplicate key")
env[key] = value
if not env.has_key('SCGI') or env['SCGI'] != '1':
raise Exception("invalid request: missing/broken SCGI=1 header")
return {'env': env, 'body': body}
while 1:
conn, addr = servsocket.accept()
result_status = 200
result = ''
try:
print 'Connected from', addr
data = ''
header = False
while not header:
newdata = conn.recv(1024)
if len(newdata) == 0: raise Exception("invalid request: unexpected EOF")
data += newdata
header = parsereq(data)
envvar = header['env']['QUERY_STRING']
result = header['env'][envvar]
except Exception as e:
print traceback.format_exc()
result_status = 500
result = str(e)
try:
conn.sendall("Status: " + str(result_status) + "\\r\\nContent-Type: text/plain\\r\\n\\r\\n")
conn.sendall(result)
conn.close()
except:
print traceback.format_exc()
"""
class SCGI(Service):
name = "scgi"
binary = [ None ]
def __init__(self):
super(SCGI, self).__init__()
self.sockfile = os.path.join(base.Env.dir, "tmp", "sockets", self.name + ".sock")
def Prepare(self):
sockdir = self.tests.PrepareDir(os.path.join("tmp", "sockets"))
program = self.tests.PrepareFile('scgi-envcheck.py', SCGI_ENVCHECK, mode = 0755)
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
sock.bind(os.path.relpath(self.sockfile))
sock.listen(8)
self.binary = [ program ]
self.fork(*self.binary, inp = sock)
def Cleanup(self):
if None != self.sockfile:
try:
os.remove(self.sockfile)
except BaseException, e:
print >>sys.stderr, "Couldn't delete socket '%s': %s" % (self.sockfile, e)
self.tests.CleanupDir(os.path.join("tmp", "sockets"))
self.tests.CleanupFile('scgi-envcheck.py')
class TestPathInfo1(CurlRequest):
URL = "/scgi/abc/xyz?PATH_INFO"
EXPECT_RESPONSE_BODY = "/abc/xyz"
EXPECT_RESPONSE_CODE = 200
class TestRequestUri1(CurlRequest):
URL = "/scgi/abc/xyz?REQUEST_URI"
EXPECT_RESPONSE_BODY = "/scgi/abc/xyz?REQUEST_URI"
EXPECT_RESPONSE_CODE = 200
class Test(GroupTest):
group = [
TestPathInfo1,
TestRequestUri1,
]
config = """
run_scgi;
"""
def FeatureCheck(self):
scgi = SCGI()
self.plain_config = """
setup {{ module_load "mod_scgi"; }}
run_scgi = {{
core.wsgi ( "/scgi", {{ scgi "unix:{socket}"; }} );
}};
""".format(socket = scgi.sockfile)
self.tests.add_service(scgi)
return True
Loading…
Cancel
Save