Grasscutters/mitmproxy
3
0
Fork 0
mirror of https://github.com/Grasscutters/mitmproxy.git synced 2024-11-27 02:24:18 +00:00
Code Issues Projects Releases Wiki Activity

SSL support for test struss.

Browse Source 
Also, rewrite requests usage for latest version.
This commit is contained in:
Aldo Cortesi 2012-06-07 17:02:17 +12:00
parent 81fc990314
commit a29ebe31dc
2 changed files with 41 additions and 16 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
libpathod/test.py
View File

@ -1,36 +1,41 @@
import json, threading, Queue import json, threading, Queue
import requests import requests
import pathod import pathod, utils
IFACE = "127.0.0.1" IFACE = "127.0.0.1"
class Daemon: class Daemon:
def __init__(self, staticdir=None, anchors=()): def __init__(self, staticdir=None, anchors=(), ssl=None):
self.app = pathod.make_app(staticdir=staticdir, anchors=anchors) self.app = pathod.make_app(staticdir=staticdir, anchors=anchors)
self.q = Queue.Queue() self.q = Queue.Queue()
self.thread = PaThread(self.q, self.app) self.thread = PaThread(self.q, self.app, ssl)
self.thread.start() self.thread.start()
self.port = self.q.get(True, 5) self.port = self.q.get(True, 5)
self.urlbase = "http://%s:%s"%(IFACE, self.port) self.urlbase = "%s://%s:%s"%("https" if ssl else "http", IFACE, self.port)
def info(self): def info(self):
resp = requests.get("%s/api/info"%self.urlbase) resp = requests.get("%s/api/info"%self.urlbase, verify=False)
if resp.ok: return resp.json
return json.loads(resp.read())
def shutdown(self): def shutdown(self):
requests.post("%s/api/shutdown"%self.urlbase) requests.post("%s/api/shutdown"%self.urlbase, verify=False)
class PaThread(threading.Thread): class PaThread(threading.Thread):
def __init__(self, q, app): def __init__(self, q, app, ssl):
threading.Thread.__init__(self) threading.Thread.__init__(self)
self.q = q self.q, self.app, self.ssl = q, app, ssl
self.app = app
self.port = None self.port = None
# begin nocover # begin nocover
def run(self): def run(self):
self.server, self.port = pathod.make_server(self.app, 0, IFACE, None) if self.ssl is True:
ssloptions = dict(
keyfile = utils.data.path("resources/server.key"),
certfile = utils.data.path("resources/server.crt"),
)
else:
ssloptions = self.ssl
self.server, self.port = pathod.make_server(self.app, 0, IFACE, ssloptions)
self.q.put(self.port) self.q.put(self.port)
pathod.run(self.server) pathod.run(self.server)

28
test/test_test.py
View File

@ -1,8 +1,9 @@
import time import time, logging
import libpry import libpry
import requests import requests
from libpathod import test, version from libpathod import test, version, utils
logging.disable(logging.CRITICAL)
class uDaemonManual(libpry.AutoTree): class uDaemonManual(libpry.AutoTree):
def test_startstop(self): def test_startstop(self):
@ -11,8 +12,27 @@ class uDaemonManual(libpry.AutoTree):
assert rsp.ok assert rsp.ok
assert rsp.status_code == 202 assert rsp.status_code == 202
d.shutdown() d.shutdown()
rsp = requests.get("http://localhost:%s/p/202"%d.port) libpry.raises(requests.ConnectionError, requests.get, "http://localhost:%s/p/202"%d.port)
assert not rsp.ok
def test_startstop_ssl(self):
d = test.Daemon(ssl=True)
rsp = requests.get("https://localhost:%s/p/202"%d.port, verify=False)
assert rsp.ok
assert rsp.status_code == 202
d.shutdown()
libpry.raises(requests.ConnectionError, requests.get, "http://localhost:%s/p/202"%d.port)
def test_startstop_ssl_explicit(self):
ssloptions = dict(
keyfile = utils.data.path("resources/server.key"),
certfile = utils.data.path("resources/server.crt"),
)
d = test.Daemon(ssl=ssloptions)
rsp = requests.get("https://localhost:%s/p/202"%d.port, verify=False)
assert rsp.ok
assert rsp.status_code == 202
d.shutdown()
libpry.raises(requests.ConnectionError, requests.get, "http://localhost:%s/p/202"%d.port)
class uDaemon(libpry.AutoTree): class uDaemon(libpry.AutoTree):