2012-06-13 06:16:47 +00:00
|
|
|
import os, shutil, tempfile
|
2012-06-09 01:42:43 +00:00
|
|
|
from contextlib import contextmanager
|
2013-02-16 03:46:16 +00:00
|
|
|
from libmproxy import flow, utils
|
2012-06-28 02:29:15 +00:00
|
|
|
from netlib import certutils
|
2013-02-16 03:46:16 +00:00
|
|
|
|
2011-03-05 02:58:48 +00:00
|
|
|
|
|
|
|
def treq(conn=None):
|
|
|
|
if not conn:
|
2011-08-03 10:38:23 +00:00
|
|
|
conn = flow.ClientConnect(("address", 22))
|
2012-02-19 22:29:36 +00:00
|
|
|
headers = flow.ODictCaseless()
|
2011-03-05 02:58:48 +00:00
|
|
|
headers["header"] = ["qvalue"]
|
2012-06-09 22:46:22 +00:00
|
|
|
return flow.Request(conn, (1, 1), "host", 80, "http", "GET", "/path", headers, "content")
|
2011-03-05 02:58:48 +00:00
|
|
|
|
|
|
|
|
|
|
|
def tresp(req=None):
|
|
|
|
if not req:
|
|
|
|
req = treq()
|
2012-02-19 22:29:36 +00:00
|
|
|
headers = flow.ODictCaseless()
|
2011-03-05 02:58:48 +00:00
|
|
|
headers["header_response"] = ["svalue"]
|
2012-06-28 02:29:15 +00:00
|
|
|
cert = certutils.SSLCert.from_der(file(test_data.path("data/dercert")).read())
|
|
|
|
return flow.Response(req, (1, 1), 200, "message", headers, "content_response", cert)
|
2011-03-05 02:58:48 +00:00
|
|
|
|
|
|
|
|
|
|
|
def tflow():
|
|
|
|
r = treq()
|
|
|
|
return flow.Flow(r)
|
|
|
|
|
|
|
|
|
|
|
|
def tflow_full():
|
|
|
|
r = treq()
|
|
|
|
f = flow.Flow(r)
|
|
|
|
f.response = tresp(r)
|
|
|
|
return f
|
|
|
|
|
|
|
|
|
2011-05-14 23:22:35 +00:00
|
|
|
def tflow_err():
|
|
|
|
r = treq()
|
|
|
|
f = flow.Flow(r)
|
2011-08-03 10:38:23 +00:00
|
|
|
f.error = flow.Error(r, "error")
|
2011-05-14 23:22:35 +00:00
|
|
|
return f
|
|
|
|
|
|
|
|
|
2012-06-09 01:42:43 +00:00
|
|
|
@contextmanager
|
|
|
|
def tmpdir(*args, **kwargs):
|
|
|
|
orig_workdir = os.getcwd()
|
|
|
|
temp_workdir = tempfile.mkdtemp(*args, **kwargs)
|
|
|
|
os.chdir(temp_workdir)
|
|
|
|
|
|
|
|
yield temp_workdir
|
|
|
|
|
|
|
|
os.chdir(orig_workdir)
|
|
|
|
shutil.rmtree(temp_workdir)
|
|
|
|
|
|
|
|
|
2012-06-08 22:57:00 +00:00
|
|
|
def raises(exc, obj, *args, **kwargs):
|
|
|
|
"""
|
|
|
|
Assert that a callable raises a specified exception.
|
|
|
|
|
|
|
|
:exc An exception class or a string. If a class, assert that an
|
|
|
|
exception of this type is raised. If a string, assert that the string
|
|
|
|
occurs in the string representation of the exception, based on a
|
|
|
|
case-insenstivie match.
|
|
|
|
|
|
|
|
:obj A callable object.
|
|
|
|
|
|
|
|
:args Arguments to be passsed to the callable.
|
|
|
|
|
|
|
|
:kwargs Arguments to be passed to the callable.
|
|
|
|
"""
|
|
|
|
try:
|
|
|
|
apply(obj, args, kwargs)
|
|
|
|
except Exception, v:
|
|
|
|
if isinstance(exc, basestring):
|
|
|
|
if exc.lower() in str(v).lower():
|
|
|
|
return
|
|
|
|
else:
|
|
|
|
raise AssertionError(
|
|
|
|
"Expected %s, but caught %s"%(
|
|
|
|
repr(str(exc)), v
|
|
|
|
)
|
|
|
|
)
|
|
|
|
else:
|
|
|
|
if isinstance(v, exc):
|
|
|
|
return
|
|
|
|
else:
|
|
|
|
raise AssertionError(
|
|
|
|
"Expected %s, but caught %s %s"%(
|
|
|
|
exc.__name__, v.__class__.__name__, str(v)
|
|
|
|
)
|
|
|
|
)
|
|
|
|
raise AssertionError("No exception raised.")
|
2013-01-05 12:16:08 +00:00
|
|
|
|
2012-06-09 01:42:43 +00:00
|
|
|
test_data = utils.Data(__name__)
|