jingrow/gunicorn
1
0
Fork 0
mirror of https://github.com/frappe/gunicorn.git synced 2026-01-14 11:09:11 +08:00
Code Issues Packages Projects Releases Wiki Activity

Imported simplehttp request tests.

Browse Source
This commit is contained in:
Paul J. Davis 2010-05-30 21:12:29 -04:00
parent 516adafcbe
commit 27cc404f1f
48 changed files with 724 additions and 803 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
gunicorn/app/pasterapp.py
View File

@ -22,7 +22,7 @@ class PasterApplication(Application):
cfgfname = os.path.normpath(os.path.join(os.getcwd(), args[0]))
cfgfname = os.path.abspath(cfgfname)
if not os.path.exists(cfgfname):
parser.error("Config file not found.")
parser.error("Config file not found: %s" % cfgfname)
self.cfgurl = 'config:%s' % cfgfname
self.relpath = os.path.dirname(cfgfname)

187
gunicorn/http/request.py
View File

@ -1,187 +0,0 @@
# -*- coding: utf-8 -
#
# This file is part of gunicorn released under the MIT license.
# See the NOTICE for more information.
import errno
import logging
import os
import re
import socket
try:
from cStringIO import StringIO
except ImportError:
from StringIO import StringIO
import sys
from urllib import unquote
from simplehttp import RequestParser
from gunicorn import __version__
from gunicorn.http.parser import Parser
from gunicorn.http.response import Response, KeepAliveResponse
from gunicorn.http.tee import TeeInput
from gunicorn.util import CHUNK_SIZE
NORMALIZE_SPACE = re.compile(r'(?:\r\n)?[ \t]+')
class RequestError(Exception):
pass
class Request(object):
RESPONSE_CLASS = Response
SERVER_VERSION = "gunicorn/%s" % __version__
DEFAULTS = {
"wsgi.url_scheme": 'http',
"wsgi.input": StringIO(),
"wsgi.errors": sys.stderr,
"wsgi.version": (1, 0),
"wsgi.multithread": False,
"wsgi.multiprocess": True,
"wsgi.run_once": False,
"SCRIPT_NAME": "",
"SERVER_SOFTWARE": "gunicorn/%s" % __version__
}
def __init__(self, cfg, socket, client_address, server_address):
self.cfg = cfg
self.socket = socket
self.client_address = client_address
self.server_address = server_address
self.response_status = None
self.response_headers = []
self._version = 11
self.parser = RequestParser(self.socket)
self.log = logging.getLogger(__name__)
self.response = None
self.response_chunked = False
self.headers_sent = False
self.req = None
def read(self):
environ = {}
headers = []
ended = False
req = None
self.req = req = self.parser.next()
##self.log.debug("%s", self.parser.status)
self.log.debug("Headers:\n%s" % req.headers)
# authors should be aware that REMOTE_HOST and REMOTE_ADDR
# may not qualify the remote addr:
# http://www.ietf.org/rfc/rfc3875
client_address = self.client_address or "127.0.0.1"
forward_address = client_address
server_address = self.server_address
script_name = os.environ.get("SCRIPT_NAME", "")
content_type = ""
content_length = ""
for hdr_name, hdr_value in req.headers:
name = hdr_name.lower()
if name == "expect":
# handle expect
if hdr_value.lower() == "100-continue":
self.socket.send("HTTP/1.1 100 Continue\r\n\r\n")
elif name == "x-forwarded-for":
forward_address = hdr_value
elif name == "host":
host = hdr_value
elif name == "script_name":
script_name = hdr_value
elif name == "content-type":
content_type = hdr_value
elif name == "content-length":
content_length = hdr_value
else:
continue
# This value should evaluate true if an equivalent application
# object may be simultaneously invoked by another process, and
# should evaluate false otherwise. In debug mode we fall to one
# worker so we comply to pylons and other paster app.
wsgi_multiprocess = self.cfg.workers > 1
if isinstance(forward_address, basestring):
# we only took the last one
# http://en.wikipedia.org/wiki/X-Forwarded-For
if "," in forward_address:
forward_adress = forward_address.split(",")[-1].strip()
remote_addr = forward_address.split(":")
if len(remote_addr) == 1:
remote_addr.append('')
else:
remote_addr = forward_address
if isinstance(server_address, basestring):
server_address = server_address.split(":")
if len(server_address) == 1:
server_address.append('')
path_info = req.path
if script_name:
path_info = path_info.split(script_name, 1)[-1]
environ = {
"wsgi.url_scheme": url_scheme,
"wsgi.input": req.body,
"wsgi.errors": sys.stderr,
"wsgi.version": (1, 0),
"wsgi.multithread": False,
"wsgi.multiprocess": wsgi_multiprocess,
"wsgi.run_once": False,
"SCRIPT_NAME": script_name,
"SERVER_SOFTWARE": self.SERVER_VERSION,
"REQUEST_METHOD": req.method,
"PATH_INFO": unquote(path_info),
"QUERY_STRING": req.query,
"RAW_URI": req.path,
"CONTENT_TYPE": content_type,
"CONTENT_LENGTH": content_length,
"REMOTE_ADDR": remote_addr[0],
"REMOTE_PORT": str(remote_addr[1]),
"SERVER_NAME": server_address[0],
"SERVER_PORT": str(server_address[1]),
"SERVER_PROTOCOL": req.version
}
for key, value in req.headers:
key = 'HTTP_' + key.upper().replace('-', '_')
if key not in ('HTTP_CONTENT_TYPE', 'HTTP_CONTENT_LENGTH'):
environ[key] = value
return environ
def start_response(self, status, headers, exc_info=None):
if exc_info:
try:
if self.response and self.response.headers_sent:
raise exc_info[0], exc_info[1], exc_info[2]
finally:
exc_info = None
elif self.response is not None:
raise AssertionError("Response headers already set!")
self.response = self.RESPONSE_CLASS(self, status, headers)
return self.response.write
class KeepAliveRequest(Request):
RESPONSE_CLASS = KeepAliveResponse
def read(self):
try:
return super(KeepAliveRequest, self).read()
except socket.error, e:
if e[0] == errno.ECONNRESET:
return
raise

214
gunicorn/http/tee.py
View File

@ -1,214 +0,0 @@
# -*- coding: utf-8 -
#
# This file is part of gunicorn released under the MIT license.
# See the NOTICE for more information.
"""
TeeInput replace old FileInput. It use a file
if size > MAX_BODY or memory. It's now possible to rewind
read or restart etc ... It's based on TeeInput from Gunicorn.
"""
import os
try:
from cStringIO import StringIO
except ImportError:
from StringIO import StringIO
import tempfile
from gunicorn import util
class UnexpectedEOF(object):
""" exception raised when remote closed the connection """
class TeeInput(object):
CHUNK_SIZE = util.CHUNK_SIZE
def __init__(self, cfg, socket, parser, buf):
self.cfg = cfg
self.buf = StringIO()
self.parser = parser
self._sock = socket
self._is_socket = True
self._len = parser.content_len
if not self.parser.content_len and not self.parser.is_chunked:
self.tmp = StringIO()
elif self._len and self._len < util.MAX_BODY:
self.tmp = StringIO()
else:
self.tmp = tempfile.TemporaryFile(dir=self.cfg.tmp_upload_dir)
if len(buf.getvalue()) > 0:
chunk, self.buf = parser.filter_body(buf)
if chunk:
self.tmp.write(chunk)
self.tmp.flush()
self._finalize()
self.tmp.seek(0)
del buf
@property
def len(self):
if self._len: return self._len
if self._is_socket:
pos = self.tmp.tell()
self.tmp.seek(0, 2)
while True:
if not self._tee(self.CHUNK_SIZE):
break
self.tmp.seek(pos)
self._len = self._tmp_size()
return self._len
def seek(self, offset, whence=0):
""" naive implementation of seek """
if self._is_socket:
self.tmp.seek(0, 2)
while True:
if not self._tee(self.CHUNK_SIZE):
break
self.tmp.seek(offset, whence)
def flush(self):
self.tmp.flush()
def read(self, length=-1):
""" read """
if not self._is_socket:
return self.tmp.read(length)
if length < 0:
buf = StringIO()
buf.write(self.tmp.read())
while True:
chunk = self._tee(self.CHUNK_SIZE)
if not chunk:
break
buf.write(chunk)
return buf.getvalue()
else:
dest = StringIO()
diff = self._tmp_size() - self.tmp.tell()
if not diff:
dest.write(self._tee(length))
return self._ensure_length(dest, length)
else:
l = min(diff, length)
dest.write(self.tmp.read(l))
return self._ensure_length(dest, length)
def readline(self, size=-1):
if not self._is_socket:
return self.tmp.readline()
orig_size = self._tmp_size()
if self.tmp.tell() == orig_size:
if not self._tee(self.CHUNK_SIZE):
return ''
self.tmp.seek(orig_size)
# now we can get line
line = self.tmp.readline()
if line.find("\n") >=0:
return line
buf = StringIO()
buf.write(line)
while True:
orig_size = self.tmp.tell()
data = self._tee(self.CHUNK_SIZE)
if not data:
break
self.tmp.seek(orig_size)
buf.write(self.tmp.readline())
if data.find("\n") >= 0:
break
return buf.getvalue()
def readlines(self, sizehint=0):
total = 0
lines = []
line = self.readline()
while line:
lines.append(line)
total += len(line)
if 0 < sizehint <= total:
break
line = self.readline()
return lines
def next(self):
r = self.readline()
if not r:
raise StopIteration
return r
__next__ = next
def __iter__(self):
return self
def get_socket(self):
return self._sock.dup()
def _tee(self, length):
""" fetch partial body"""
buf2 = self.buf
buf2.seek(0, 2)
while True:
chunk, buf2 = self.parser.filter_body(buf2)
if chunk:
self.tmp.write(chunk)
self.tmp.flush()
self.tmp.seek(0, 2)
self.buf = StringIO()
self.buf.write(buf2.getvalue())
return chunk
if self.parser.body_eof():
break
if not self._is_socket:
if self.parser.is_chunked:
data = buf2.getvalue()
if data.find("\r\n") >= 0:
continue
raise UnexpectedEOF("remote closed the connection")
data = self._sock.recv(length)
if not data:
self._is_socket = False
buf2.write(data)
self._finalize()
return ""
def _finalize(self):
""" here we wil fetch final trailers
if any."""
if self.parser.body_eof():
self.buf = StringIO()
self._is_socket = False
def _tmp_size(self):
if hasattr(self.tmp, 'fileno'):
return int(os.fstat(self.tmp.fileno())[6])
else:
return len(self.tmp.getvalue())
def _ensure_length(self, dest, length):
if not len(dest.getvalue()) or not self._len:
return dest.getvalue()
while True:
if len(dest.getvalue()) >= length:
break
data = self._tee(length - len(dest.getvalue()))
if not data:
break
dest.write(data)
return dest.getvalue()

7
gunicorn/http/wsgi.py
View File

@ -27,6 +27,7 @@ def create(req, sock, client, server, debug=False):
# http://www.ietf.org/rfc/rfc3875
client = client or "127.0.0.1"
forward = client
url_scheme = "http"
script_name = os.environ.get("SCRIPT_NAME", "")
content_type = ""
content_length = ""
@ -38,6 +39,10 @@ def create(req, sock, client, server, debug=False):
sock.send("HTTP/1.1 100 Continue\r\n\r\n")
elif name == "x-forwarded-for":
forward = hdr_value
elif name == "x-forwarded-protocol" and value.lower() == "ssl":
url_scheme = "https"
elif name == "x-forwarded-ssl" and value.lower() == "on":
url_scheme = "https"
elif name == "host":
server = hdr_value
elif name == "script_name":
@ -76,7 +81,7 @@ def create(req, sock, client, server, debug=False):
path_info = path_info.split(script_name, 1)[1]
environ = {
"wsgi.url_scheme": 'http',
"wsgi.url_scheme": url_scheme,
"wsgi.input": req.body,
"wsgi.errors": sys.stderr,
"wsgi.version": (1, 0),

213
tests/001-test-parser.py
View File

@ -4,203 +4,22 @@
# See the NOTICE for more information.
import t
import treq
@t.request("001.http")
def test_001(buf, p):
headers = []
buf2 = p.filter_headers(headers, buf)
t.ne(buf2, False)
t.eq(p.method, "PUT")
t.eq(p.version, (1,0))
t.eq(p.path, "/stuff/here")
t.eq(p.query_string, "foo=bar")
t.eq(sorted(p.headers), [
('Content-Length', '14'),
('Content-Type', 'application/json'),
('Server', 'http://127.0.0.1:5984')
])
body, tr = p.filter_body(buf2)
t.eq(body, '{"nom": "nom"}')
t.eq(p.body_eof(), True)
import glob
import os
dirname = os.path.dirname(__file__)
reqdir = os.path.join(dirname, "requests")
@t.request("002.http")
def test_002(buf, p):
headers = []
buf2 = p.filter_headers(headers, buf)
t.ne(buf2, False)
t.eq(p.method, "GET")
t.eq(p.version, (1, 1))
t.eq(p.path, "/test")
t.eq(p.query_string, "")
t.eq(sorted(p.headers), [
("Accept", "*/*"),
("Host", "0.0.0.0=5000"),
("User-Agent", "curl/7.18.0 (i486-pc-linux-gnu) libcurl/7.18.0 OpenSSL/0.9.8g zlib/1.2.3.3 libidn/1.1")
])
body, tr = p.filter_body(buf2)
t.eq(body, "")
def load_py(fname):
config = globals().copy()
config["uri"] = treq.uri
execfile(fname, config)
return config["request"]
@t.request("003.http")
def test_003(buf, p):
headers = []
buf2 = p.filter_headers(headers, buf)
t.ne(buf2, False)
t.eq(p.method, "GET")
t.eq(p.version, (1, 1))
t.eq(p.path, "/favicon.ico")
t.eq(p.query_string, "")
t.eq(sorted(p.headers), [
("Accept", "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8"),
("Accept-Charset", "ISO-8859-1,utf-8;q=0.7,*;q=0.7"),
("Accept-Encoding", "gzip,deflate"),
("Accept-Language", "en-us,en;q=0.5"),
("Connection", "keep-alive"),
("Host", "0.0.0.0=5000"),
("Keep-Alive", "300"),
("User-Agent", "Mozilla/5.0 (X11; U; Linux i686; en-US; rv:1.9) Gecko/2008061015 Firefox/3.0"),
])
body, tr = p.filter_body(buf2)
t.eq(body, "")
@t.request("004.http")
def test_004(buf, p):
headers = []
buf2 = p.filter_headers(headers, buf)
t.ne(buf2, False)
t.eq(p.method, "GET")
t.eq(p.version, (1, 1))
t.eq(p.path, "/dumbfuck")
t.eq(p.query_string, "")
t.eq(p.headers, [("Aaaaaaaaaaaaa", "++++++++++")])
body, tr = p.filter_body(buf2)
t.eq(body, "")
@t.request("005.http")
def test_005(buf, p):
headers = []
buf2 = p.filter_headers(headers, buf)
t.ne(buf2, False)
t.eq(p.method, "GET")
t.eq(p.version, (1, 1))
t.eq(p.path, "/forums/1/topics/2375")
t.eq(p.query_string, "page=1")
t.eq(p.fragment, "posts-17408")
body, tr = p.filter_body(buf2)
t.eq(body, "")
@t.request("006.http")
def test_006(buf, p):
headers = []
buf2 = p.filter_headers(headers, buf)
t.ne(buf2, False)
t.eq(p.method, "GET")
t.eq(p.version, (1, 1))
t.eq(p.path, "/get_no_headers_no_body/world")
t.eq(p.query_string, "")
t.eq(p.fragment, "")
body, tr = p.filter_body(buf2)
t.eq(body, "")
@t.request("007.http")
def test_007(buf, p):
headers = []
buf2 = p.filter_headers(headers, buf)
t.ne(buf2, False)
t.eq(p.method, "GET")
t.eq(p.version, (1, 1))
t.eq(p.path, "/get_one_header_no_body")
t.eq(p.query_string, "")
t.eq(p.fragment, "")
t.eq(p.headers, [('Accept', '*/*')])
body, tr = p.filter_body(buf2)
t.eq(body, "")
@t.request("008.http")
def test_008(buf, p):
headers = []
buf2 = p.filter_headers(headers, buf)
t.ne(buf2, False)
t.eq(p.method, "GET")
t.eq(p.version, (1, 0))
t.eq(p.path, "/get_funky_content_length_body_hello")
t.eq(p.query_string, "")
t.eq(p.fragment, "")
t.eq(p.headers, [('Content-Length', '5')])
body, tr = p.filter_body(buf2)
t.eq(body, "HELLO")
@t.request("009.http")
def test_009(buf, p):
headers = []
buf2 = p.filter_headers(headers, buf)
t.ne(buf2, False)
t.eq(p.method, "POST")
t.eq(p.version, (1, 1))
t.eq(p.path, "/post_identity_body_world")
t.eq(p.query_string, "q=search")
t.eq(p.fragment, "hey")
t.eq(sorted(p.headers), [
('Accept', '*/*'),
('Content-Length', '5'),
('Transfer-Encoding', 'identity')
])
body, tr = p.filter_body(buf2)
t.eq(body, "World")
@t.request("010.http")
def test_010(buf, p):
headers = []
buf2 = p.filter_headers(headers, buf)
t.ne(buf2, False)
t.eq(p.method, "POST")
t.eq(p.version, (1, 1))
t.eq(p.path, "/post_chunked_all_your_base")
t.eq(p.headers, [('Transfer-Encoding', 'chunked')])
t.eq(p.is_chunked, True)
t.eq(p._chunk_eof, False)
t.ne(p.body_eof(), True)
body = ""
while not p.body_eof():
chunk, buf2 = p.filter_body(buf2)
#print chunk
if chunk:
body += chunk
t.eq(body, "all your base are belong to us")
@t.request("011.http")
def test_011(buf, p):
headers = []
buf2 = p.filter_headers(headers, buf)
t.ne(buf2, False)
t.eq(p.method, "POST")
t.eq(p.version, (1, 1))
t.eq(p.path, "/two_chunks_mult_zero_end")
t.eq(p.headers, [('Transfer-Encoding', 'chunked')])
t.eq(p.is_chunked, True)
t.eq(p._chunk_eof, False)
t.ne(p.body_eof(), True)
body = ""
while not p.body_eof():
chunk, buf2 = p.filter_body(buf2)
if chunk:
body += chunk
t.eq(body, "hello world")
@t.request("017.http")
def test_012(buf, p):
headers = []
buf2 = p.filter_headers(headers, buf)
t.ne(buf2, False)
t.eq(p.method, "GET")
t.eq(p.version, (1, 0))
t.eq(p.path, "/stuff/here")
t.eq(p.query_string, "foo=bar")
t.eq(p.is_chunked, False)
t.eq(p._chunk_eof, False)
t.eq(p.body_eof(), True)
t.eq(p.headers, [("If-Match", "bazinga!, large-sound")])
def test_http_parser():
for fname in glob.glob(os.path.join(reqdir, "*.http")):
expect = load_py(os.path.splitext(fname)[0] + ".py")
req = treq.request(fname, expect)
for case in req.gen_cases():
yield case

284
tests/002-test-request.py
View File

@ -6,146 +6,146 @@
import os
import t
from gunicorn.http import tee
@t.http_request("001.http")
def test_001(req):
e = req.read()
t.eq(e['CONTENT_LENGTH'], '14')
t.eq(e['wsgi.version'], (1,0))
t.eq(e['REQUEST_METHOD'], 'PUT')
t.eq(e['PATH_INFO'], '/stuff/here')
t.eq(e['CONTENT_TYPE'], 'application/json')
t.eq(e['QUERY_STRING'], 'foo=bar')
t.eq(isinstance(e['wsgi.input'], tee.TeeInput), True)
body = e['wsgi.input'].read()
t.eq(body, '{"nom": "nom"}')
@t.http_request("002.http")
def test_002(req):
e = req.read()
t.eq(e['REQUEST_METHOD'], 'GET')
t.eq(e['PATH_INFO'], "/test")
t.eq(e['QUERY_STRING'], "")
t.eq(e['HTTP_ACCEPT'], "*/*")
t.eq(e['HTTP_HOST'], "0.0.0.0=5000")
t.eq(e['HTTP_USER_AGENT'], "curl/7.18.0 (i486-pc-linux-gnu) libcurl/7.18.0 OpenSSL/0.9.8g zlib/1.2.3.3 libidn/1.1")
body = e['wsgi.input'].read()
t.eq(body, '')
@t.http_request("003.http")
def test_003(req):
e = req.read()
t.eq(e['REQUEST_METHOD'], 'GET')
t.eq(e['PATH_INFO'], "/favicon.ico")
t.eq(e['QUERY_STRING'], "")
t.eq(e['HTTP_ACCEPT'], "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8")
t.eq(e['HTTP_KEEP_ALIVE'], "300")
body = e['wsgi.input'].read()
t.eq(body, '')
@t.http_request("004.http")
def test_004(req):
e = req.read()
t.eq(e['REQUEST_METHOD'], 'GET')
t.eq(e['PATH_INFO'], "/dumbfuck")
t.eq(e['QUERY_STRING'], "")
body = e['wsgi.input'].read()
t.eq(body, '')
@t.http_request("005.http")
def test_005(req):
e = req.read()
t.eq(e['REQUEST_METHOD'], 'GET')
t.eq(e['PATH_INFO'], "/forums/1/topics/2375")
t.eq(e['QUERY_STRING'], "page=1")
body = e['wsgi.input'].read()
t.eq(body, '')
@t.http_request("006.http")
def test_006(req):
e = req.read()
t.eq(e['REQUEST_METHOD'], 'GET')
t.eq(e['PATH_INFO'], "/get_no_headers_no_body/world")
t.eq(e['QUERY_STRING'], "")
body = e['wsgi.input'].read()
t.eq(body, '')
@t.http_request("007.http")
def test_007(req):
e = req.read()
t.eq(e['REQUEST_METHOD'], 'GET')
t.eq(e['PATH_INFO'], "/get_one_header_no_body")
t.eq(e['QUERY_STRING'], "")
t.eq(e['HTTP_ACCEPT'], "*/*")
body = e['wsgi.input'].read()
t.eq(body, '')
@t.http_request("008.http")
def test_008(req):
e = req.read()
t.eq(e['REQUEST_METHOD'], 'GET')
t.eq(e['PATH_INFO'], "/get_funky_content_length_body_hello")
t.eq(e['QUERY_STRING'], "")
t.eq(e['CONTENT_LENGTH'], '5')
body = e['wsgi.input'].read()
t.eq(body, "HELLO")
@t.http_request("009.http")
def test_009(req):
e = req.read()
t.eq(e['REQUEST_METHOD'], 'POST')
t.eq(e['PATH_INFO'], "/post_identity_body_world")
t.eq(e['QUERY_STRING'], "q=search")
t.eq(e['CONTENT_LENGTH'], '5')
body = e['wsgi.input'].read()
t.eq(body, "World")
@t.http_request("010.http")
def test_010(req):
e = req.read()
t.eq(e['REQUEST_METHOD'], 'POST')
t.eq(e['PATH_INFO'], "/post_chunked_all_your_base")
t.eq(e['HTTP_TRANSFER_ENCODING'], "chunked")
t.eq(e['CONTENT_LENGTH'], '30')
body = e['wsgi.input'].read()
t.eq(body, "all your base are belong to us")
@t.http_request("011.http")
def test_011(req):
e = req.read()
t.eq(e['REQUEST_METHOD'], 'POST')
t.eq(e['PATH_INFO'], "/two_chunks_mult_zero_end")
t.eq(e['HTTP_TRANSFER_ENCODING'], "chunked")
t.eq(e['CONTENT_LENGTH'], '11')
body = e['wsgi.input'].read()
t.eq(body, "hello world")
@t.http_request("017.http")
def test_017(req):
e = req.read()
t.eq(e['REQUEST_METHOD'], 'GET')
t.eq(e['PATH_INFO'], "/stuff/here")
t.eq(e["HTTP_IF_MATCH"], "bazinga!, large-sound")
t.eq(e["wsgi.input"].read(), "")
@t.http_request("017.http")
def test_018(req):
os.environ['SCRIPT_NAME'] = "/stuff"
e = req.read()
t.eq(e['REQUEST_METHOD'], 'GET')
t.eq(e['SCRIPT_NAME'], "/stuff")
t.eq(e['PATH_INFO'], "/here")
t.eq(e["wsgi.input"].read(), "")
# from gunicorn.http import tee
#
# @t.http_request("001.http")
# def test_001(req):
# e = req.read()
# t.eq(e['CONTENT_LENGTH'], '14')
# t.eq(e['wsgi.version'], (1,0))
# t.eq(e['REQUEST_METHOD'], 'PUT')
# t.eq(e['PATH_INFO'], '/stuff/here')
# t.eq(e['CONTENT_TYPE'], 'application/json')
# t.eq(e['QUERY_STRING'], 'foo=bar')
#
# t.eq(isinstance(e['wsgi.input'], tee.TeeInput), True)
# body = e['wsgi.input'].read()
# t.eq(body, '{"nom": "nom"}')
#
# @t.http_request("002.http")
# def test_002(req):
# e = req.read()
# t.eq(e['REQUEST_METHOD'], 'GET')
# t.eq(e['PATH_INFO'], "/test")
# t.eq(e['QUERY_STRING'], "")
# t.eq(e['HTTP_ACCEPT'], "*/*")
# t.eq(e['HTTP_HOST'], "0.0.0.0=5000")
# t.eq(e['HTTP_USER_AGENT'], "curl/7.18.0 (i486-pc-linux-gnu) libcurl/7.18.0 OpenSSL/0.9.8g zlib/1.2.3.3 libidn/1.1")
# body = e['wsgi.input'].read()
# t.eq(body, '')
#
# @t.http_request("003.http")
# def test_003(req):
# e = req.read()
#
# t.eq(e['REQUEST_METHOD'], 'GET')
# t.eq(e['PATH_INFO'], "/favicon.ico")
# t.eq(e['QUERY_STRING'], "")
# t.eq(e['HTTP_ACCEPT'], "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8")
# t.eq(e['HTTP_KEEP_ALIVE'], "300")
#
# body = e['wsgi.input'].read()
# t.eq(body, '')
#
# @t.http_request("004.http")
# def test_004(req):
# e = req.read()
# t.eq(e['REQUEST_METHOD'], 'GET')
# t.eq(e['PATH_INFO'], "/dumbfuck")
# t.eq(e['QUERY_STRING'], "")
# body = e['wsgi.input'].read()
# t.eq(body, '')
#
#
# @t.http_request("005.http")
# def test_005(req):
# e = req.read()
# t.eq(e['REQUEST_METHOD'], 'GET')
# t.eq(e['PATH_INFO'], "/forums/1/topics/2375")
# t.eq(e['QUERY_STRING'], "page=1")
# body = e['wsgi.input'].read()
# t.eq(body, '')
#
#
# @t.http_request("006.http")
# def test_006(req):
# e = req.read()
# t.eq(e['REQUEST_METHOD'], 'GET')
# t.eq(e['PATH_INFO'], "/get_no_headers_no_body/world")
# t.eq(e['QUERY_STRING'], "")
# body = e['wsgi.input'].read()
# t.eq(body, '')
#
#
# @t.http_request("007.http")
# def test_007(req):
# e = req.read()
# t.eq(e['REQUEST_METHOD'], 'GET')
# t.eq(e['PATH_INFO'], "/get_one_header_no_body")
# t.eq(e['QUERY_STRING'], "")
# t.eq(e['HTTP_ACCEPT'], "*/*")
# body = e['wsgi.input'].read()
# t.eq(body, '')
#
#
# @t.http_request("008.http")
# def test_008(req):
# e = req.read()
# t.eq(e['REQUEST_METHOD'], 'GET')
# t.eq(e['PATH_INFO'], "/get_funky_content_length_body_hello")
# t.eq(e['QUERY_STRING'], "")
# t.eq(e['CONTENT_LENGTH'], '5')
# body = e['wsgi.input'].read()
# t.eq(body, "HELLO")
#
#
# @t.http_request("009.http")
# def test_009(req):
# e = req.read()
# t.eq(e['REQUEST_METHOD'], 'POST')
# t.eq(e['PATH_INFO'], "/post_identity_body_world")
# t.eq(e['QUERY_STRING'], "q=search")
# t.eq(e['CONTENT_LENGTH'], '5')
# body = e['wsgi.input'].read()
# t.eq(body, "World")
#
#
# @t.http_request("010.http")
# def test_010(req):
# e = req.read()
# t.eq(e['REQUEST_METHOD'], 'POST')
# t.eq(e['PATH_INFO'], "/post_chunked_all_your_base")
# t.eq(e['HTTP_TRANSFER_ENCODING'], "chunked")
# t.eq(e['CONTENT_LENGTH'], '30')
# body = e['wsgi.input'].read()
# t.eq(body, "all your base are belong to us")
#
#
# @t.http_request("011.http")
# def test_011(req):
# e = req.read()
# t.eq(e['REQUEST_METHOD'], 'POST')
# t.eq(e['PATH_INFO'], "/two_chunks_mult_zero_end")
# t.eq(e['HTTP_TRANSFER_ENCODING'], "chunked")
# t.eq(e['CONTENT_LENGTH'], '11')
# body = e['wsgi.input'].read()
# t.eq(body, "hello world")
#
# @t.http_request("017.http")
# def test_017(req):
# e = req.read()
# t.eq(e['REQUEST_METHOD'], 'GET')
# t.eq(e['PATH_INFO'], "/stuff/here")
# t.eq(e["HTTP_IF_MATCH"], "bazinga!, large-sound")
# t.eq(e["wsgi.input"].read(), "")
#
# @t.http_request("017.http")
# def test_018(req):
# os.environ['SCRIPT_NAME'] = "/stuff"
# e = req.read()
# t.eq(e['REQUEST_METHOD'], 'GET')
# t.eq(e['SCRIPT_NAME'], "/stuff")
# t.eq(e['PATH_INFO'], "/here")
# t.eq(e["wsgi.input"].read(), "")
#

4
tests/003-test-config.py
View File

@ -19,7 +19,7 @@ dirname = os.path.dirname(__file__)
def cfg_file():
return os.path.join(dirname, "config", "test_cfg.py")
def paster_ini():
return os.path.join(dirname, "..", "examples", "pylonstest", "nose.ini")
return os.path.join(dirname, "..", "examples", "frameworks", "pylonstest", "nose.ini")
def PasterApp():
try:
@ -145,7 +145,7 @@ def test_cmd_line():
with AltArgs(["prog_name", "-w", "3"]):
app = NoConfigApp()
t.eq(app.cfg.workers, 3)
with AltArgs(["prog_name", "-d"]):
with AltArgs(["prog_name", "--debug"]):
app = NoConfigApp()
t.eq(app.cfg.debug, True)

11
tests/requests/001.py Normal file
View File

@ -0,0 +1,11 @@
request = {
"method": "PUT",
"uri": uri("/stuff/here?foo=bar"),
"version": (1, 0),
"headers": [
("SERVER", "http://127.0.0.1:5984"),
("CONTENT-TYPE", "application/json"),
("CONTENT-LENGTH", "14")
],
"body": '{"nom": "nom"}'
}

2
tests/requests/002.http
View File

@ -2,4 +2,4 @@ GET /test HTTP/1.1\r\n
User-Agent: curl/7.18.0 (i486-pc-linux-gnu) libcurl/7.18.0 OpenSSL/0.9.8g zlib/1.2.3.3 libidn/1.1\r\n
Host: 0.0.0.0=5000\r\n
Accept: */*\r\n
\r\n
\r\n

11
tests/requests/002.py Normal file
View File

@ -0,0 +1,11 @@
request = {
"method": "GET",
"uri": uri("/test"),
"version": (1, 1),
"headers": [
("USER-AGENT", "curl/7.18.0 (i486-pc-linux-gnu) libcurl/7.18.0 OpenSSL/0.9.8g zlib/1.2.3.3 libidn/1.1"),
("HOST", "0.0.0.0=5000"),
("ACCEPT", "*/*")
],
"body": ""
}

2
tests/requests/003.http
View File

@ -7,4 +7,4 @@ Accept-Encoding: gzip,deflate\r\n
Accept-Charset: ISO-8859-1,utf-8;q=0.7,*;q=0.7\r\n
Keep-Alive: 300\r\n
Connection: keep-alive\r\n
\r\n
\r\n

16
tests/requests/003.py Normal file
View File

@ -0,0 +1,16 @@
request = {
"method": "GET",
"uri": uri("/favicon.ico"),
"version": (1, 1),
"headers": [
("HOST", "0.0.0.0=5000"),
("USER-AGENT", "Mozilla/5.0 (X11; U; Linux i686; en-US; rv:1.9) Gecko/2008061015 Firefox/3.0"),
("ACCEPT", "text/html,application/xhtml+xml,application/xml;q=0.9,*/*;q=0.8"),
("ACCEPT-LANGUAGE", "en-us,en;q=0.5"),
("ACCEPT-ENCODING", "gzip,deflate"),
("ACCEPT-CHARSET", "ISO-8859-1,utf-8;q=0.7,*;q=0.7"),
("KEEP-ALIVE", "300"),
("CONNECTION", "keep-alive")
],
"body": ""
}

4
tests/requests/004.http
View File

@ -1,3 +1,3 @@
GET /dumbfuck HTTP/1.1\r\n
GET /silly HTTP/1.1\r\n
aaaaaaaaaaaaa:++++++++++\r\n
\r\n
\r\n

9
tests/requests/004.py Normal file
View File

@ -0,0 +1,9 @@
request = {
"method": "GET",
"uri": uri("/silly"),
"version": (1, 1),
"headers": [
("AAAAAAAAAAAAA", "++++++++++")
],
"body": ""
}

2
tests/requests/005.http
View File

@ -1,2 +1,2 @@
GET /forums/1/topics/2375?page=1#posts-17408 HTTP/1.1\r\n
\r\n
\r\n

7
tests/requests/005.py Normal file
View File

@ -0,0 +1,7 @@
request = {
"method": "GET",
"uri": uri("/forums/1/topics/2375?page=1#posts-17408"),
"version": (1, 1),
"headers": [],
"body": ""
}

2
tests/requests/006.http
View File

@ -1,2 +1,2 @@
GET /get_no_headers_no_body/world HTTP/1.1\r\n
\r\n
\r\n

7
tests/requests/006.py Normal file
View File

@ -0,0 +1,7 @@
request = {
"method": "GET",
"uri": uri("/get_no_headers_no_body/world"),
"version": (1, 1),
"headers": [],
"body": ""
}

2
tests/requests/007.http
View File

@ -1,3 +1,3 @@
GET /get_one_header_no_body HTTP/1.1\r\n
Accept: */*\r\n
\r\n
\r\n

9
tests/requests/007.py Normal file
View File

@ -0,0 +1,9 @@
request = {
"method": "GET",
"uri": uri("/get_one_header_no_body"),
"version": (1, 1),
"headers": [
("ACCEPT", "*/*")
],
"body": ""
}

2
tests/requests/008.http
View File

@ -1,4 +1,4 @@
GET /get_funky_content_length_body_hello HTTP/1.0\r\n
GET /unusual_content_length HTTP/1.0\r\n
conTENT-Length: 5\r\n
\r\n
HELLO

9
tests/requests/008.py Normal file
View File

@ -0,0 +1,9 @@
request = {
"method": "GET",
"uri": uri("/unusual_content_length"),
"version": (1, 0),
"headers": [
("CONTENT-LENGTH", "5")
],
"body": "HELLO"
}

11
tests/requests/009.py Normal file
View File

@ -0,0 +1,11 @@
request = {
"method": "POST",
"uri": uri("/post_identity_body_world?q=search#hey"),
"version": (1, 1),
"headers": [
("ACCEPT", "*/*"),
("TRANSFER-ENCODING", "identity"),
("CONTENT-LENGTH", "5")
],
"body": "World"
}

2
tests/requests/010.http
View File

@ -4,4 +4,4 @@ Transfer-Encoding: chunked\r\n
1e\r\n
all your base are belong to us\r\n
0\r\n
\r\n

9
tests/requests/010.py Normal file
View File

@ -0,0 +1,9 @@
request = {
"method": "POST",
"uri": uri("/post_chunked_all_your_base"),
"version": (1, 1),
"headers": [
("TRANSFER-ENCODING", "chunked"),
],
"body": "all your base are belong to us"
}

9
tests/requests/011.py Normal file
View File

@ -0,0 +1,9 @@
request = {
"method": "POST",
"uri": uri("/two_chunks_mult_zero_end"),
"version": (1, 1),
"headers": [
("TRANSFER-ENCODING", "chunked")
],
"body": "hello world"
}

13
tests/requests/012.py Normal file
View File

@ -0,0 +1,13 @@
request = {
"method": "POST",
"uri": uri("/chunked_w_trailing_headers"),
"version": (1, 1),
"headers": [
("TRANSFER-ENCODING", "chunked")
],
"body": "hello world",
"trailers": [
("VARY", "*"),
("CONTENT-TYPE", "text/plain")
]
}

17
tests/requests/013.http
View File

@ -1,8 +1,9 @@
POST /chunked_w_bullshit_after_length HTTP/1.1
Transfer-Encoding: chunked
5; some; parameters=stuff
hello
6; blahblah; blah
world
0
POST /chunked_w_extensions HTTP/1.1\r\n
Transfer-Encoding: chunked\r\n
\r\n
5; some; parameters=stuff\r\n
hello\r\n
6; blahblah; blah\r\n
world\r\n
0\r\n
\r\n

9
tests/requests/013.py Normal file
View File

@ -0,0 +1,9 @@
request = {
"method": "POST",
"uri": uri("/chunked_w_extensions"),
"version": (1, 1),
"headers": [
("TRANSFER-ENCODING", "chunked")
],
"body": "hello world"
}

4
tests/requests/014.http
View File

@ -1,2 +1,2 @@
GET /with_"stupid"_quotes?foo="bar" HTTP/1.1
GET /with_"quotes"?foo="bar" HTTP/1.1\r\n
\r\n

7
tests/requests/014.py Normal file
View File

@ -0,0 +1,7 @@
request = {
"method": "GET",
"uri": uri('/with_"quotes"?foo="bar"'),
"version": (1, 1),
"headers": [],
"body": ""
}

10
tests/requests/015.http
View File

@ -1,5 +1,5 @@
GET /test HTTP/1.0
Host: 0.0.0.0:5000
User-Agent: ApacheBench/2.3
Accept: */*
GET /test HTTP/1.0\r\n
Host: 0.0.0.0:5000\r\n
User-Agent: ApacheBench/2.3\r\n
Accept: */*\r\n
\r\n

11
tests/requests/015.py Normal file
View File

@ -0,0 +1,11 @@
request = {
"method": "GET",
"uri": uri("/test"),
"version": (1, 0),
"headers": [
("HOST", "0.0.0.0:5000"),
("USER-AGENT", "ApacheBench/2.3"),
("ACCEPT", "*/*")
],
"body": ""
}

67
tests/requests/016.http
View File

@ -1,33 +1,34 @@
GET / HTTP/1.1
X-SSL-Bullshit: -----BEGIN CERTIFICATE-----
MIIFbTCCBFWgAwIBAgICH4cwDQYJKoZIhvcNAQEFBQAwcDELMAkGA1UEBhMCVUsx
ETAPBgNVBAoTCGVTY2llbmNlMRIwEAYDVQQLEwlBdXRob3JpdHkxCzAJBgNVBAMT
AkNBMS0wKwYJKoZIhvcNAQkBFh5jYS1vcGVyYXRvckBncmlkLXN1cHBvcnQuYWMu
dWswHhcNMDYwNzI3MTQxMzI4WhcNMDcwNzI3MTQxMzI4WjBbMQswCQYDVQQGEwJV
SzERMA8GA1UEChMIZVNjaWVuY2UxEzARBgNVBAsTCk1hbmNoZXN0ZXIxCzAJBgNV
BAcTmrsogriqMWLAk1DMRcwFQYDVQQDEw5taWNoYWVsIHBhcmQYJKoZIhvcNAQEB
BQADggEPADCCAQoCggEBANPEQBgl1IaKdSS1TbhF3hEXSl72G9J+WC/1R64fAcEF
W51rEyFYiIeZGx/BVzwXbeBoNUK41OK65sxGuflMo5gLflbwJtHBRIEKAfVVp3YR
gW7cMA/s/XKgL1GEC7rQw8lIZT8RApukCGqOVHSi/F1SiFlPDxuDfmdiNzL31+sL
0iwHDdNkGjy5pyBSB8Y79dsSJtCW/iaLB0/n8Sj7HgvvZJ7x0fr+RQjYOUUfrePP
u2MSpFyf+9BbC/aXgaZuiCvSR+8Snv3xApQY+fULK/xY8h8Ua51iXoQ5jrgu2SqR
wgA7BUi3G8LFzMBl8FRCDYGUDy7M6QaHXx1ZWIPWNKsCAwEAAaOCAiQwggIgMAwG
1UdEwEB/wQCMAAwEQYJYIZIAYb4QgHTTPAQDAgWgMA4GA1UdDwEB/wQEAwID6DAs
BglghkgBhvhCAQ0EHxYdVUsgZS1TY2llbmNlIFVzZXIgQ2VydGlmaWNhdGUwHQYD
VR0OBBYEFDTt/sf9PeMaZDHkUIldrDYMNTBZMIGaBgNVHSMEgZIwgY+AFAI4qxGj
loCLDdMVKwiljjDastqooXSkcjBwMQswCQYDVQQGEwJVSzERMA8GA1UEChMIZVNj
aWVuY2UxEjAQBgNVBAsTCUF1dGhvcml0eTELMAkGA1UEAxMCQ0ExLTArBgkqhkiG
9w0BCQEWHmNhLW9wZXJhdG9yQGdyaWQtc3VwcG9ydC5hYy51a4IBADApBgNVHRIE
IjAggR5jYS1vcGVyYXRvckBncmlkLXN1cHBvcnQuYWMudWswGQYDVR0gBBIwEDAO
BgwrBgEEAdkvAQEBAQYwPQYJYIZIAYb4QgEEBDAWLmh0dHA6Ly9jYS5ncmlkLXN1
cHBvcnQuYWMudmT4sopwqlBWsvcHViL2NybC9jYWNybC5jcmwwPQYJYIZIAYb4Qg
EDBDAWLmh0dHA6Ly9jYS5ncmlkLXN1cHBvcnQuYWMudWsvcHViL2NybC9jYWNybC
5jcmwwPwYDVR0fBDgwNjA0oDKgMIYuaHR0cDovL2NhLmdyaWQt5hYy51ay9wdWIv
Y3JsL2NhY3JsLmNybDANBgkqhkiG9w0BAQUFAAOCAQEAS/U4iiooBENGW/Hwmmd3
XCy6Zrt08YjKCzGNjorT98g8uGsqYjSxv/hmi0qlnlHs+k/3Iobc3LjS5AMYr5L8
UO7OSkgFFlLHQyC9JzPfmLCAugvzEbyv4Olnsr8hbxF1MbKZoQxUZtMVu29wjfXk
hTeApBv7eaKCWpSp7MCbvgzm74izKhu3vlDk9w6qVrxePfGgpKPqfHiOoGhFnbTK
wTC6o2xq5y0qZ03JonF7OJspEd3I5zKY3E+ov7/ZhW6DqT8UFvsAdjvQbXyhV8Eu
Yhixw1aKEPzNjNowuIseVogKOLXxWI5vAi5HgXdS0/ES5gDGsABo4fqovUKlgop3
RA==
-----END CERTIFICATE-----
GET / HTTP/1.1\r\n
X-SSL-Cert: -----BEGIN CERTIFICATE-----\r\n
MIIFbTCCBFWgAwIBAgICH4cwDQYJKoZIhvcNAQEFBQAwcDELMAkGA1UEBhMCVUsx\r\n
ETAPBgNVBAoTCGVTY2llbmNlMRIwEAYDVQQLEwlBdXRob3JpdHkxCzAJBgNVBAMT\r\n
AkNBMS0wKwYJKoZIhvcNAQkBFh5jYS1vcGVyYXRvckBncmlkLXN1cHBvcnQuYWMu\r\n
dWswHhcNMDYwNzI3MTQxMzI4WhcNMDcwNzI3MTQxMzI4WjBbMQswCQYDVQQGEwJV\r\n
SzERMA8GA1UEChMIZVNjaWVuY2UxEzARBgNVBAsTCk1hbmNoZXN0ZXIxCzAJBgNV\r\n
BAcTmrsogriqMWLAk1DMRcwFQYDVQQDEw5taWNoYWVsIHBhcmQYJKoZIhvcNAQEB\r\n
BQADggEPADCCAQoCggEBANPEQBgl1IaKdSS1TbhF3hEXSl72G9J+WC/1R64fAcEF\r\n
W51rEyFYiIeZGx/BVzwXbeBoNUK41OK65sxGuflMo5gLflbwJtHBRIEKAfVVp3YR\r\n
gW7cMA/s/XKgL1GEC7rQw8lIZT8RApukCGqOVHSi/F1SiFlPDxuDfmdiNzL31+sL\r\n
0iwHDdNkGjy5pyBSB8Y79dsSJtCW/iaLB0/n8Sj7HgvvZJ7x0fr+RQjYOUUfrePP\r\n
u2MSpFyf+9BbC/aXgaZuiCvSR+8Snv3xApQY+fULK/xY8h8Ua51iXoQ5jrgu2SqR\r\n
wgA7BUi3G8LFzMBl8FRCDYGUDy7M6QaHXx1ZWIPWNKsCAwEAAaOCAiQwggIgMAwG\r\n
1UdEwEB/wQCMAAwEQYJYIZIAYb4QgHTTPAQDAgWgMA4GA1UdDwEB/wQEAwID6DAs\r\n
BglghkgBhvhCAQ0EHxYdVUsgZS1TY2llbmNlIFVzZXIgQ2VydGlmaWNhdGUwHQYD\r\n
VR0OBBYEFDTt/sf9PeMaZDHkUIldrDYMNTBZMIGaBgNVHSMEgZIwgY+AFAI4qxGj\r\n
loCLDdMVKwiljjDastqooXSkcjBwMQswCQYDVQQGEwJVSzERMA8GA1UEChMIZVNj\r\n
aWVuY2UxEjAQBgNVBAsTCUF1dGhvcml0eTELMAkGA1UEAxMCQ0ExLTArBgkqhkiG\r\n
9w0BCQEWHmNhLW9wZXJhdG9yQGdyaWQtc3VwcG9ydC5hYy51a4IBADApBgNVHRIE\r\n
IjAggR5jYS1vcGVyYXRvckBncmlkLXN1cHBvcnQuYWMudWswGQYDVR0gBBIwEDAO\r\n
BgwrBgEEAdkvAQEBAQYwPQYJYIZIAYb4QgEEBDAWLmh0dHA6Ly9jYS5ncmlkLXN1\r\n
cHBvcnQuYWMudmT4sopwqlBWsvcHViL2NybC9jYWNybC5jcmwwPQYJYIZIAYb4Qg\r\n
EDBDAWLmh0dHA6Ly9jYS5ncmlkLXN1cHBvcnQuYWMudWsvcHViL2NybC9jYWNybC\r\n
5jcmwwPwYDVR0fBDgwNjA0oDKgMIYuaHR0cDovL2NhLmdyaWQt5hYy51ay9wdWIv\r\n
Y3JsL2NhY3JsLmNybDANBgkqhkiG9w0BAQUFAAOCAQEAS/U4iiooBENGW/Hwmmd3\r\n
XCy6Zrt08YjKCzGNjorT98g8uGsqYjSxv/hmi0qlnlHs+k/3Iobc3LjS5AMYr5L8\r\n
UO7OSkgFFlLHQyC9JzPfmLCAugvzEbyv4Olnsr8hbxF1MbKZoQxUZtMVu29wjfXk\r\n
hTeApBv7eaKCWpSp7MCbvgzm74izKhu3vlDk9w6qVrxePfGgpKPqfHiOoGhFnbTK\r\n
wTC6o2xq5y0qZ03JonF7OJspEd3I5zKY3E+ov7/ZhW6DqT8UFvsAdjvQbXyhV8Eu\r\n
Yhixw1aKEPzNjNowuIseVogKOLXxWI5vAi5HgXdS0/ES5gDGsABo4fqovUKlgop3\r\n
RA==\r\n
-----END CERTIFICATE-----\r\n
\r\n

40
tests/requests/016.py Normal file
View File

@ -0,0 +1,40 @@
certificate = """-----BEGIN CERTIFICATE-----\r\n
MIIFbTCCBFWgAwIBAgICH4cwDQYJKoZIhvcNAQEFBQAwcDELMAkGA1UEBhMCVUsx\r\n
ETAPBgNVBAoTCGVTY2llbmNlMRIwEAYDVQQLEwlBdXRob3JpdHkxCzAJBgNVBAMT\r\n
AkNBMS0wKwYJKoZIhvcNAQkBFh5jYS1vcGVyYXRvckBncmlkLXN1cHBvcnQuYWMu\r\n
dWswHhcNMDYwNzI3MTQxMzI4WhcNMDcwNzI3MTQxMzI4WjBbMQswCQYDVQQGEwJV\r\n
SzERMA8GA1UEChMIZVNjaWVuY2UxEzARBgNVBAsTCk1hbmNoZXN0ZXIxCzAJBgNV\r\n
BAcTmrsogriqMWLAk1DMRcwFQYDVQQDEw5taWNoYWVsIHBhcmQYJKoZIhvcNAQEB\r\n
BQADggEPADCCAQoCggEBANPEQBgl1IaKdSS1TbhF3hEXSl72G9J+WC/1R64fAcEF\r\n
W51rEyFYiIeZGx/BVzwXbeBoNUK41OK65sxGuflMo5gLflbwJtHBRIEKAfVVp3YR\r\n
gW7cMA/s/XKgL1GEC7rQw8lIZT8RApukCGqOVHSi/F1SiFlPDxuDfmdiNzL31+sL\r\n
0iwHDdNkGjy5pyBSB8Y79dsSJtCW/iaLB0/n8Sj7HgvvZJ7x0fr+RQjYOUUfrePP\r\n
u2MSpFyf+9BbC/aXgaZuiCvSR+8Snv3xApQY+fULK/xY8h8Ua51iXoQ5jrgu2SqR\r\n
wgA7BUi3G8LFzMBl8FRCDYGUDy7M6QaHXx1ZWIPWNKsCAwEAAaOCAiQwggIgMAwG\r\n
1UdEwEB/wQCMAAwEQYJYIZIAYb4QgHTTPAQDAgWgMA4GA1UdDwEB/wQEAwID6DAs\r\n
BglghkgBhvhCAQ0EHxYdVUsgZS1TY2llbmNlIFVzZXIgQ2VydGlmaWNhdGUwHQYD\r\n
VR0OBBYEFDTt/sf9PeMaZDHkUIldrDYMNTBZMIGaBgNVHSMEgZIwgY+AFAI4qxGj\r\n
loCLDdMVKwiljjDastqooXSkcjBwMQswCQYDVQQGEwJVSzERMA8GA1UEChMIZVNj\r\n
aWVuY2UxEjAQBgNVBAsTCUF1dGhvcml0eTELMAkGA1UEAxMCQ0ExLTArBgkqhkiG\r\n
9w0BCQEWHmNhLW9wZXJhdG9yQGdyaWQtc3VwcG9ydC5hYy51a4IBADApBgNVHRIE\r\n
IjAggR5jYS1vcGVyYXRvckBncmlkLXN1cHBvcnQuYWMudWswGQYDVR0gBBIwEDAO\r\n
BgwrBgEEAdkvAQEBAQYwPQYJYIZIAYb4QgEEBDAWLmh0dHA6Ly9jYS5ncmlkLXN1\r\n
cHBvcnQuYWMudmT4sopwqlBWsvcHViL2NybC9jYWNybC5jcmwwPQYJYIZIAYb4Qg\r\n
EDBDAWLmh0dHA6Ly9jYS5ncmlkLXN1cHBvcnQuYWMudWsvcHViL2NybC9jYWNybC\r\n
5jcmwwPwYDVR0fBDgwNjA0oDKgMIYuaHR0cDovL2NhLmdyaWQt5hYy51ay9wdWIv\r\n
Y3JsL2NhY3JsLmNybDANBgkqhkiG9w0BAQUFAAOCAQEAS/U4iiooBENGW/Hwmmd3\r\n
XCy6Zrt08YjKCzGNjorT98g8uGsqYjSxv/hmi0qlnlHs+k/3Iobc3LjS5AMYr5L8\r\n
UO7OSkgFFlLHQyC9JzPfmLCAugvzEbyv4Olnsr8hbxF1MbKZoQxUZtMVu29wjfXk\r\n
hTeApBv7eaKCWpSp7MCbvgzm74izKhu3vlDk9w6qVrxePfGgpKPqfHiOoGhFnbTK\r\n
wTC6o2xq5y0qZ03JonF7OJspEd3I5zKY3E+ov7/ZhW6DqT8UFvsAdjvQbXyhV8Eu\r\n
Yhixw1aKEPzNjNowuIseVogKOLXxWI5vAi5HgXdS0/ES5gDGsABo4fqovUKlgop3\r\n
RA==\r\n
-----END CERTIFICATE-----""".replace("\n\n", "\n")
request = {
"method": "GET",
"uri": uri("/"),
"version": (1, 1),
"headers": [("X-SSL-CERT", certificate)],
"body": ""
}

10
tests/requests/017.py Normal file
View File

@ -0,0 +1,10 @@
request = {
"method": "GET",
"uri": uri("/stuff/here?foo=bar"),
"version": (1, 0),
"headers": [
("IF-MATCH", "bazinga!"),
("IF-MATCH", "large-sound")
],
"body": ""
}

2
tests/requests/018.http Normal file
View File

@ -0,0 +1,2 @@
GET /foo/bar HTTP/1.0\r\n
baz

2
tests/requests/018.py Normal file
View File

@ -0,0 +1,2 @@
from gunicorn.http.errors import NoMoreData
request = NoMoreData

2
tests/requests/019.http Normal file
View File

@ -0,0 +1,2 @@
GET HTTP/1.1\r\n
\r\n

2
tests/requests/019.py Normal file
View File

@ -0,0 +1,2 @@
from gunicorn.http.errors import InvalidRequestLine
request = InvalidRequestLine

2
tests/requests/020.http Normal file
View File

@ -0,0 +1,2 @@
-blargh /foo HTTP/1.1\r\n
\r\n

2
tests/requests/020.py Normal file
View File

@ -0,0 +1,2 @@
from gunicorn.http.errors import InvalidRequestMethod
request = InvalidRequestMethod

2
tests/requests/021.http Normal file
View File

@ -0,0 +1,2 @@
GET /foo FTP/1.1\r\n
\r\n

2
tests/requests/021.py Normal file
View File

@ -0,0 +1,2 @@
from gunicorn.http.errors import InvalidHTTPVersion
request = InvalidHTTPVersion

3
tests/requests/022.http Normal file
View File

@ -0,0 +1,3 @@
GET /foo HTTP/1.1\r\n
ba\0z: bar\r\n
\r\n

2
tests/requests/022.py Normal file
View File

@ -0,0 +1,2 @@
from gunicorn.http.errors import InvalidHeaderName
request = InvalidHeaderName

7
tests/t.py
View File

@ -11,7 +11,7 @@ import tempfile
dirname = os.path.dirname(__file__)
from gunicorn.http.parser import Parser
from gunicorn.http.parser import RequestParser
from gunicorn.http.request import Request
from gunicorn.config import Config
@ -30,7 +30,7 @@ class request(object):
def __call__(self, func):
def run():
src = data_source(self.fname)
func(src, Parser())
func(src, RequestParser(src))
run.func_name = func.func_name
return run
@ -104,6 +104,9 @@ def has(a, b):
def hasnot(a, b):
assert not hasattr(a, b), "%r has an attribute %r" % (a, b)
def istype(a, b):
assert isinstance(a, b), "%r is not an instance of %r" % (a, b)
def raises(exctype, func, *args, **kwargs):
try:
func(*args, **kwargs)

274
tests/treq.py Normal file
View File

@ -0,0 +1,274 @@
# Copyright 2009 Paul J. Davis <paul.joseph.davis@gmail.com>
#
# This file is part of the pywebmachine package released
# under the MIT license.
import t
import inspect
import os
import random
import urlparse
from gunicorn.http.errors import ParseException
from gunicorn.http.parser import RequestParser
dirname = os.path.dirname(__file__)
random.seed()
def uri(data):
ret = {"raw": data}
parts = urlparse.urlparse(data)
ret["scheme"] = parts.scheme or None
ret["host"] = parts.netloc.rsplit(":", 1)[0] or None
ret["port"] = parts.port or 80
if parts.path and parts.params:
ret["path"] = ";".join([parts.path, parts.params])
elif parts.path:
ret["path"] = parts.path
elif parts.params:
# Don't think this can happen
ret["path"] = ";" + parts.path
else:
ret["path"] = None
ret["query"] = parts.query or None
ret["fragment"] = parts.fragment or None
return ret
class request(object):
def __init__(self, fname, expect):
self.fname = fname
self.name = os.path.basename(fname)
self.expect = expect
if not isinstance(self.expect, list):
self.expect = [self.expect]
with open(self.fname) as handle:
self.data = handle.read()
self.data = self.data.replace("\n", "").replace("\\r\\n", "\r\n")
self.data = self.data.replace("\\0", "\000")
# Functions for sending data to the parser.
# These functions mock out reading from a
# socket or other data source that might
# be used in real life.
def send_all(self):
yield self.data
def send_lines(self):
lines = self.data
pos = lines.find("\r\n")
while pos > 0:
yield lines[:pos+2]
lines = lines[pos+2:]
pos = lines.find("\r\n")
if len(lines):
yield lines
def send_bytes(self):
for d in self.data:
yield d
def send_random(self):
maxs = len(self.data) / 10
read = 0
while read < len(self.data):
chunk = random.randint(1, maxs)
yield self.data[read:read+chunk]
read += chunk
# These functions define the sizes that the
# read functions will read with.
def size_all(self):
return -1
def size_bytes(self):
return 1
def size_small_random(self):
return random.randint(0, 2)
def size_random(self):
return random.randint(1, 4096)
# Match a body against various ways of reading
# a message. Pass in the request, expected body
# and one of the size functions.
def szread(self, func, sizes):
sz = sizes()
data = func(sz)
if sz >= 0 and len(data) > sz:
raise AssertionError("Read more than %d bytes: %s" % (sz, data))
return data
def match_read(self, req, body, sizes):
data = self.szread(req.body.read, sizes)
count = 1000
while len(body):
if body[:len(data)] != data:
raise AssertionError("Invalid body data read: %r != %r" % (
data, body[:len(data)]))
body = body[len(data):]
data = self.szread(req.body.read, sizes)
if not data:
count -= 1
if count <= 0:
raise AssertionError("Unexpected apparent EOF")
if len(body):
raise AssertionError("Failed to read entire body: %r" % body)
elif len(data):
raise AssertionError("Read beyond expected body: %r" % data)
data = req.body.read(sizes())
if data:
raise AssertionError("Read after body finished: %r" % data)
def match_readline(self, req, body, sizes):
data = self.szread(req.body.readline, sizes)
count = 1000
while len(body):
if body[:len(data)] != data:
raise AssertionError("Invalid data read: %r" % data)
if '\n' in data[:-1]:
raise AssertionError("Embedded new line: %r" % data)
body = body[len(data):]
data = self.szread(req.body.readline, sizes)
if not data:
count -= 1
if count <= 0:
raise AssertionError("Apparent unexpected EOF")
if len(body):
raise AssertionError("Failed to read entire body: %r" % body)
elif len(data):
raise AssertionError("Read beyond expected body: %r" % data)
data = req.body.readline(sizes())
if data:
raise AssertionError("Read data after body finished: %r" % data)
def match_readlines(self, req, body, sizes):
"""\
This skips the sizes checks as we don't implement it.
"""
data = req.body.readlines()
for line in data:
if '\n' in line[:-1]:
raise AssertionError("Embedded new line: %r" % line)
if line != body[:len(line)]:
raise AssertionError("Invalid body data read: %r != %r" % (
line, body[:len(line)]))
body = body[len(line):]
if len(body):
raise AssertionError("Failed to read entire body: %r" % body)
data = req.body.readlines(sizes())
if data:
raise AssertionError("Read data after body finished: %r" % data)
def match_iter(self, req, body, sizes):
"""\
This skips sizes because there's its not part of the iter api.
"""
for line in req.body:
if '\n' in line[:-1]:
raise AssertionError("Embedded new line: %r" % line)
if line != body[:len(line)]:
raise AssertionError("Invalid body data read: %r != %r" % (
line, body[:len(line)]))
body = body[len(line):]
if len(body):
raise AssertionError("Failed to read entire body: %r" % body)
try:
data = iter(req.body).next()
raise AssertionError("Read data after body finished: %r" % data)
except StopIteration:
pass
# Construct a series of test cases from the permutations of
# send, size, and match functions.
def gen_cases(self):
def get_funs(p):
return [v for k, v in inspect.getmembers(self) if k.startswith(p)]
senders = get_funs("send_")
sizers = get_funs("size_")
matchers = get_funs("match_")
cfgs = [
(mt, sz, sn)
for mt in matchers
for sz in sizers
for sn in senders
]
# Strip out match_readlines, match_iter for all but one sizer
cfgs = [
(mt, sz, sn)
for (mt, sz, sn) in cfgs
if mt in [self.match_readlines, self.match_iter]
and sz != self.size_all
]
ret = []
for (mt, sz, sn) in cfgs:
mtn = mt.func_name[6:]
szn = sz.func_name[5:]
snn = sn.func_name[5:]
def test_req(sn, sz, mt):
self.check(sn, sz, mt)
desc = "%s: MT: %s SZ: %s SN: %s" % (self.name, mtn, szn, snn)
test_req.description = desc
ret.append((test_req, sn, sz, mt))
return ret
def check(self, sender, sizer, matcher):
cases = self.expect[:]
ended = False
try:
p = RequestParser(sender())
except Exception, e:
if not isinstance(cases[0], Exception):
raise
self.same_error(e, cases[0])
t.eq(len(casese), 1)
while True:
try:
req = p.next()
except StopIteration, e:
t.eq(len(cases), 0)
ended = True
break
except ParseException, e:
if not issubclass(cases[0], Exception):
raise
self.same_error(e, cases.pop(0))
t.eq(len(cases), 0)
return
else:
self.same(req, sizer, matcher, cases.pop(0))
t.eq(len(cases), 0)
t.eq(ended, True)
def same(self, req, sizer, matcher, exp):
if isinstance(req, Exception):
self.same_error(req, exp)
else:
self.same_obj(req, sizer, matcher, exp)
def same_error(self, req, exp):
t.istype(req, Exception)
t.istype(req, exp)
def same_obj(self, req, sizer, matcher, exp):
t.eq(req.method, exp["method"])
t.eq(req.uri, exp["uri"]["raw"])
t.eq(req.scheme, exp["uri"]["scheme"])
t.eq(req.host, exp["uri"]["host"])
t.eq(req.port, exp["uri"]["port"])
t.eq(req.path, exp["uri"]["path"])
t.eq(req.query, exp["uri"]["query"])
t.eq(req.fragment, exp["uri"]["fragment"])
t.eq(req.version, exp["version"])
t.eq(req.headers, exp["headers"])
matcher(req, exp["body"], sizer)
t.eq(req.trailers, exp.get("trailers", []))