mirror of
https://github.com/Grasscutters/mitmproxy.git
synced 2024-11-23 08:11:00 +00:00
Port test suite over to Nose.
This commit is contained in:
parent
456b11df9a
commit
00c8bef0ff
3
.coveragerc
Normal file
3
.coveragerc
Normal file
@ -0,0 +1,3 @@
|
||||
[report]
|
||||
omit = *contrib*
|
||||
include = *libpathod*
|
1
.gitignore
vendored
1
.gitignore
vendored
@ -6,3 +6,4 @@ MANIFEST
|
||||
# Vim swap files
|
||||
*.swp
|
||||
/doc
|
||||
.coverage
|
||||
|
@ -393,10 +393,8 @@ class Response:
|
||||
def add_timeout(self, s, callback):
|
||||
if TESTING:
|
||||
callback()
|
||||
# begin nocover
|
||||
else:
|
||||
else: # pragma: no cover
|
||||
tornado.ioloop.IOLoop.instance().add_timeout(time.time() + s, callback)
|
||||
# end nocover
|
||||
|
||||
def write_values(self, fp, vals, actions, sofar=0, skip=0, blocksize=BLOCKSIZE):
|
||||
while vals:
|
||||
|
@ -1,6 +1,7 @@
|
||||
import json, threading, Queue
|
||||
import requests
|
||||
import pathod, utils
|
||||
import tutils
|
||||
|
||||
IFACE = "127.0.0.1"
|
||||
|
||||
|
@ -1,8 +1,7 @@
|
||||
import libpry
|
||||
from libpathod import pathod
|
||||
from tornado import httpserver
|
||||
|
||||
class uApplication(libpry.AutoTree):
|
||||
class TestApplication:
|
||||
def test_anchors(self):
|
||||
a = pathod.PathodApp(staticdir=None)
|
||||
a.add_anchor("/foo", "200")
|
||||
@ -31,13 +30,6 @@ class uApplication(libpry.AutoTree):
|
||||
assert not a.log_by_id(0)
|
||||
|
||||
|
||||
class u_make_server(libpry.AutoTree):
|
||||
def test_simple(self):
|
||||
def test_make_server():
|
||||
app = pathod.PathodApp()
|
||||
assert pathod.make_server(app, 0, "127.0.0.1", None)
|
||||
|
||||
|
||||
tests = [
|
||||
uApplication(),
|
||||
u_make_server()
|
||||
]
|
||||
|
@ -1,11 +1,11 @@
|
||||
import os
|
||||
import libpry
|
||||
from libpathod import rparse, utils
|
||||
import tutils
|
||||
|
||||
rparse.TESTING = True
|
||||
|
||||
|
||||
class uMisc(libpry.AutoTree):
|
||||
class TestMisc:
|
||||
def test_generators(self):
|
||||
v = rparse.Value.parseString("'val'")[0]
|
||||
g = v.get_generator({})
|
||||
@ -26,7 +26,7 @@ class uMisc(libpry.AutoTree):
|
||||
assert g[1] == "n"
|
||||
|
||||
def test_filegenerator(self):
|
||||
t = self.tmpdir()
|
||||
with tutils.tmpdir() as t:
|
||||
path = os.path.join(t, "foo")
|
||||
f = open(path, "w")
|
||||
f.write("x"*10000)
|
||||
@ -50,7 +50,7 @@ class uMisc(libpry.AutoTree):
|
||||
v = rparse.Value.parseString("<path")[0]
|
||||
assert v.path == "path"
|
||||
|
||||
t = self.tmpdir()
|
||||
with tutils.tmpdir() as t:
|
||||
p = os.path.join(t, "path")
|
||||
f = open(p, "w")
|
||||
f.write("x"*10000)
|
||||
@ -59,8 +59,8 @@ class uMisc(libpry.AutoTree):
|
||||
assert v.get_generator(dict(staticdir=t))
|
||||
|
||||
v = rparse.Value.parseString("<path2")[0]
|
||||
libpry.raises(rparse.ServerError, v.get_generator, dict(staticdir=t))
|
||||
libpry.raises("no static directory", v.get_generator, dict())
|
||||
tutils.raises(rparse.ServerError, v.get_generator, dict(staticdir=t))
|
||||
tutils.raises("no static directory", v.get_generator, dict())
|
||||
|
||||
def test_generated_value(self):
|
||||
v = rparse.Value.parseString("@10b")[0]
|
||||
@ -136,7 +136,7 @@ class uMisc(libpry.AutoTree):
|
||||
s.serve(d)
|
||||
|
||||
|
||||
class uDisconnects(libpry.AutoTree):
|
||||
class TestDisconnects:
|
||||
def test_parse(self):
|
||||
assert (0, "disconnect") in rparse.parse({}, "400:d0").actions
|
||||
assert ("r", "disconnect") in rparse.parse({}, "400:dr").actions
|
||||
@ -155,13 +155,13 @@ class uDisconnects(libpry.AutoTree):
|
||||
assert v.value == "r"
|
||||
|
||||
|
||||
class uShortcuts(libpry.AutoTree):
|
||||
class TestShortcuts:
|
||||
def test_parse(self):
|
||||
assert rparse.parse({}, "400:c'foo'").headers[0][0][:] == "Content-Type"
|
||||
assert rparse.parse({}, "400:l'foo'").headers[0][0][:] == "Location"
|
||||
|
||||
|
||||
class uPauses(libpry.AutoTree):
|
||||
class TestPauses:
|
||||
def test_parse(self):
|
||||
e = rparse.PauseAt.expr()
|
||||
v = e.parseString("p10,10")[0]
|
||||
@ -182,9 +182,9 @@ class uPauses(libpry.AutoTree):
|
||||
assert r.actions[0] == (10, "pause", 10)
|
||||
|
||||
|
||||
class uparse(libpry.AutoTree):
|
||||
class TestParse:
|
||||
def test_parse_err(self):
|
||||
libpry.raises(rparse.ParseException, rparse.parse, {}, "400:msg,b:")
|
||||
tutils.raises(rparse.ParseException, rparse.parse, {}, "400:msg,b:")
|
||||
try:
|
||||
rparse.parse({}, "400'msg':b:")
|
||||
except rparse.ParseException, v:
|
||||
@ -212,7 +212,7 @@ class uparse(libpry.AutoTree):
|
||||
assert r.length()
|
||||
|
||||
|
||||
class uResponse(libpry.AutoTree):
|
||||
class TestResponse:
|
||||
def dummy_response(self):
|
||||
return rparse.parse({}, "400'msg'")
|
||||
|
||||
@ -291,13 +291,3 @@ class uResponse(libpry.AutoTree):
|
||||
testlen(rparse.parse({}, "400'msg'"))
|
||||
testlen(rparse.parse({}, "400'msg':h'foo'='bar'"))
|
||||
testlen(rparse.parse({}, "400'msg':h'foo'='bar':b@100b"))
|
||||
|
||||
|
||||
tests = [
|
||||
uResponse(),
|
||||
uPauses(),
|
||||
uDisconnects(),
|
||||
uMisc(),
|
||||
uparse(),
|
||||
uShortcuts()
|
||||
]
|
||||
|
@ -1,18 +1,18 @@
|
||||
import time, logging
|
||||
import libpry
|
||||
import requests
|
||||
from libpathod import test, version, utils
|
||||
import tutils
|
||||
|
||||
logging.disable(logging.CRITICAL)
|
||||
|
||||
class uDaemonManual(libpry.AutoTree):
|
||||
class TestDaemonManual:
|
||||
def test_startstop(self):
|
||||
d = test.Daemon()
|
||||
rsp = requests.get("http://localhost:%s/p/202"%d.port)
|
||||
assert rsp.ok
|
||||
assert rsp.status_code == 202
|
||||
d.shutdown()
|
||||
libpry.raises(requests.ConnectionError, requests.get, "http://localhost:%s/p/202"%d.port)
|
||||
tutils.raises(requests.ConnectionError, requests.get, "http://localhost:%s/p/202"%d.port)
|
||||
|
||||
def test_startstop_ssl(self):
|
||||
d = test.Daemon(ssl=True)
|
||||
@ -20,7 +20,7 @@ class uDaemonManual(libpry.AutoTree):
|
||||
assert rsp.ok
|
||||
assert rsp.status_code == 202
|
||||
d.shutdown()
|
||||
libpry.raises(requests.ConnectionError, requests.get, "http://localhost:%s/p/202"%d.port)
|
||||
tutils.raises(requests.ConnectionError, requests.get, "http://localhost:%s/p/202"%d.port)
|
||||
|
||||
def test_startstop_ssl_explicit(self):
|
||||
ssloptions = dict(
|
||||
@ -32,22 +32,18 @@ class uDaemonManual(libpry.AutoTree):
|
||||
assert rsp.ok
|
||||
assert rsp.status_code == 202
|
||||
d.shutdown()
|
||||
libpry.raises(requests.ConnectionError, requests.get, "http://localhost:%s/p/202"%d.port)
|
||||
tutils.raises(requests.ConnectionError, requests.get, "http://localhost:%s/p/202"%d.port)
|
||||
|
||||
|
||||
class uDaemon(libpry.AutoTree):
|
||||
class TestDaemon:
|
||||
@classmethod
|
||||
def setUpAll(self):
|
||||
self.d = test.Daemon()
|
||||
|
||||
@classmethod
|
||||
def tearDownAll(self):
|
||||
self.d.shutdown()
|
||||
|
||||
def test_info(self):
|
||||
assert tuple(self.d.info()["version"]) == version.IVERSION
|
||||
|
||||
|
||||
|
||||
tests = [
|
||||
uDaemonManual(),
|
||||
uDaemon()
|
||||
]
|
||||
|
@ -1,21 +1,13 @@
|
||||
import libpry
|
||||
from libpathod import utils
|
||||
import tutils
|
||||
|
||||
|
||||
class uparse_anchor_spec(libpry.AutoTree):
|
||||
def test_simple(self):
|
||||
def test_parse_anchor_spec():
|
||||
assert utils.parse_anchor_spec("foo=200", {}) == ("foo", "200")
|
||||
libpry.raises(utils.AnchorError, utils.parse_anchor_spec, "foobar", {})
|
||||
libpry.raises(utils.AnchorError, utils.parse_anchor_spec, "*=200", {})
|
||||
libpry.raises(utils.AnchorError, utils.parse_anchor_spec, "foo=bar", {})
|
||||
tutils.raises(utils.AnchorError, utils.parse_anchor_spec, "foobar", {})
|
||||
tutils.raises(utils.AnchorError, utils.parse_anchor_spec, "*=200", {})
|
||||
tutils.raises(utils.AnchorError, utils.parse_anchor_spec, "foo=bar", {})
|
||||
|
||||
|
||||
class udata_path(libpry.AutoTree):
|
||||
def test_simple(self):
|
||||
libpry.raises(ValueError, utils.data.path, "nonexistent")
|
||||
|
||||
|
||||
tests = [
|
||||
udata_path(),
|
||||
uparse_anchor_spec()
|
||||
]
|
||||
def test_data_path():
|
||||
tutils.raises(ValueError, utils.data.path, "nonexistent")
|
||||
|
57
test/tutils.py
Normal file
57
test/tutils.py
Normal file
@ -0,0 +1,57 @@
|
||||
import tempfile, os, shutil
|
||||
from contextlib import contextmanager
|
||||
from libpathod import utils
|
||||
|
||||
|
||||
|
||||
@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)
|
||||
|
||||
|
||||
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.")
|
||||
|
||||
test_data = utils.Data(__name__)
|
Loading…
Reference in New Issue
Block a user