From d9cb0838093a962f636ef6cda17fd652bc2ef65e Mon Sep 17 00:00:00 2001 From: Aldo Cortesi Date: Sat, 5 Mar 2011 15:58:48 +1300 Subject: Reorganize test suite to remove confusion between test utils and libmproxy utils. --- test/test_proxy.py | 98 ++++++------------------------------------------------ 1 file changed, 11 insertions(+), 87 deletions(-) (limited to 'test/test_proxy.py') diff --git a/test/test_proxy.py b/test/test_proxy.py index 8a59cf0b..00857586 100644 --- a/test/test_proxy.py +++ b/test/test_proxy.py @@ -1,114 +1,38 @@ -import threading, urllib, Queue, urllib2, cStringIO +import urllib, urllib2, cStringIO import libpry -import serv, sslserv from libmproxy import proxy, controller, utils, dump, script -import random +import tutils -# Yes, the random ports are horrible. During development, sockets are often not -# properly closed during error conditions, which means you have to wait until -# you can re-bind to the same port. This is a pain in the ass, so we just pick -# a random port and keep moving. -PROXL_PORT = random.randint(10000, 20000) -HTTP_PORT = random.randint(20000, 30000) -HTTPS_PORT = random.randint(30000, 40000) - -class TestMaster(controller.Master): - def __init__(self, port, testq): - serv = proxy.ProxyServer(proxy.Config("data/testkey.pem"), port) - controller.Master.__init__(self, serv) - self.testq = testq - self.log = [] - - def clear(self): - self.log = [] - - def handle(self, m): - self.log.append(m) - m.ack() - - -class ProxyThread(threading.Thread): - def __init__(self, port, testq): - self.tmaster = TestMaster(port, testq) - threading.Thread.__init__(self) - - def run(self): - self.tmaster.run() - - def shutdown(self): - self.tmaster.shutdown() - - -class ServerThread(threading.Thread): - def __init__(self, server): - self.server = server - threading.Thread.__init__(self) - - def run(self): - self.server.serve_forever() - - def shutdown(self): - self.server.shutdown() - - -class _TestServers(libpry.TestContainer): - def setUpAll(self): - self.tqueue = Queue.Queue() - # We don't make any concurrent requests, so we can access - # the attributes on this object safely. - self.proxthread = ProxyThread(PROXL_PORT, self.tqueue) - self.threads = [ - ServerThread(serv.make(HTTP_PORT)), - ServerThread(sslserv.make(HTTPS_PORT)), - self.proxthread - ] - for i in self.threads: - i.start() - - def setUp(self): - self.proxthread.tmaster.clear() - - def tearDownAll(self): - for i in self.threads: - i.shutdown() - - -class _ProxTests(libpry.AutoTree): - def log(self): - pthread = self.findAttr("proxthread") - return pthread.tmaster.log - - -class uSanity(_ProxTests): +class uSanity(tutils.ProxTest): def test_http(self): """ Just check that the HTTP server is running. """ - f = urllib.urlopen("http://127.0.0.1:%s"%HTTP_PORT) + f = urllib.urlopen("http://127.0.0.1:%s"%tutils.HTTP_PORT) assert f.read() def test_https(self): """ Just check that the HTTPS server is running. """ - f = urllib.urlopen("https://127.0.0.1:%s"%HTTPS_PORT) + f = urllib.urlopen("https://127.0.0.1:%s"%tutils.HTTPS_PORT) assert f.read() -class uProxy(_ProxTests): +class uProxy(tutils.ProxTest): HOST = "127.0.0.1" def _get(self, host=HOST): - r = urllib2.Request("http://%s:%s"%(host, HTTP_PORT)) - r.set_proxy("127.0.0.1:%s"%PROXL_PORT, "http") + r = urllib2.Request("http://%s:%s"%(host, tutils.HTTP_PORT)) + r.set_proxy("127.0.0.1:%s"%tutils.PROXL_PORT, "http") return urllib2.urlopen(r) def _sget(self, host=HOST): proxy_support = urllib2.ProxyHandler( - {"https" : "https://127.0.0.1:%s"%PROXL_PORT} + {"https" : "https://127.0.0.1:%s"%tutils.PROXL_PORT} ) opener = urllib2.build_opener(proxy_support) - r = urllib2.Request("https://%s:%s"%(host, HTTPS_PORT)) + r = urllib2.Request("https://%s:%s"%(host, tutils.HTTPS_PORT)) return opener.open(r) def test_http(self): @@ -307,7 +231,7 @@ tests = [ u_parse_url(), uError(), uClientConnect(), - _TestServers(), [ + tutils.TestServers(), [ uSanity(), uProxy(), ], -- cgit v1.2.3