From 1ccb2c5dea9530682aae83d489f1738d9286fa4e Mon Sep 17 00:00:00 2001
From: Aldo Cortesi <aldo@nullcube.com>
Date: Sat, 16 Feb 2013 16:46:16 +1300
Subject: Test WSGI app calling.

- Factor out test servers into a separate file
- Adjust docs to note new Flask dependency
---
 test/tservers.py | 183 +++++++++++++++++++++++++++++++++++++++++++++++++++++++
 1 file changed, 183 insertions(+)
 create mode 100644 test/tservers.py

(limited to 'test/tservers.py')

diff --git a/test/tservers.py b/test/tservers.py
new file mode 100644
index 00000000..2966a436
--- /dev/null
+++ b/test/tservers.py
@@ -0,0 +1,183 @@
+import threading, Queue
+import flask
+import human_curl as hurl
+import libpathod.test, libpathod.pathoc
+from libmproxy import proxy, flow, controller
+import tutils
+
+testapp = flask.Flask(__name__)
+
+@testapp.route("/")
+def hello():
+    return "testapp"
+
+@testapp.route("/error")
+def error():
+    raise ValueError("An exception...")
+
+
+def errapp(environ, start_response):
+    raise ValueError("errapp")
+
+
+class TestMaster(flow.FlowMaster):
+    def __init__(self, testq, config):
+        s = proxy.ProxyServer(config, 0)
+        s.apps.add(testapp, "testapp", 80)
+        s.apps.add(errapp, "errapp", 80)
+        state = flow.State()
+        flow.FlowMaster.__init__(self, s, state)
+        self.testq = testq
+
+    def handle(self, m):
+        flow.FlowMaster.handle(self, m)
+        m._ack()
+
+
+class ProxyThread(threading.Thread):
+    def __init__(self, testq, config):
+        self.tmaster = TestMaster(testq, config)
+        controller.should_exit = False
+        threading.Thread.__init__(self)
+
+    @property
+    def port(self):
+        return self.tmaster.server.port
+
+    def run(self):
+        self.tmaster.run()
+
+    def shutdown(self):
+        self.tmaster.shutdown()
+
+
+class ProxTestBase:
+    @classmethod
+    def setupAll(cls):
+        cls.tqueue = Queue.Queue()
+        cls.server = libpathod.test.Daemon(ssl=cls.ssl)
+        pconf = cls.get_proxy_config()
+        config = proxy.ProxyConfig(
+            certfile=tutils.test_data.path("data/testkey.pem"),
+            **pconf
+        )
+        cls.proxy = ProxyThread(cls.tqueue, config)
+        cls.proxy.start()
+
+    @property
+    def master(cls):
+        return cls.proxy.tmaster
+
+    @classmethod
+    def teardownAll(cls):
+        cls.proxy.shutdown()
+        cls.server.shutdown()
+
+    def setUp(self):
+        self.master.state.clear()
+
+    @property
+    def scheme(self):
+        return "https" if self.ssl else "http"
+
+    @property
+    def proxies(self):
+        """
+            The URL base for the server instance.
+        """
+        return (
+            (self.scheme, ("127.0.0.1", self.proxy.port))
+        )
+
+    @property
+    def urlbase(self):
+        """
+            The URL base for the server instance.
+        """
+        return self.server.urlbase
+
+    def last_log(self):
+        return self.server.last_log()
+
+
+class HTTPProxTest(ProxTestBase):
+    ssl = None
+    clientcerts = False
+    @classmethod
+    def get_proxy_config(cls):
+        d = dict()
+        if cls.clientcerts:
+            d["clientcerts"] = tutils.test_data.path("data/clientcert")
+        return d
+
+    def pathoc(self, connect_to = None):
+        p = libpathod.pathoc.Pathoc("localhost", self.proxy.port)
+        p.connect(connect_to)
+        return p
+
+    def pathod(self, spec):
+        """
+            Constructs a pathod request, with the appropriate base and proxy.
+        """
+        return hurl.get(
+            self.urlbase + "/p/" + spec,
+            proxy=self.proxies,
+            validate_cert=False,
+            #debug=hurl.utils.stdout_debug
+        )
+
+
+class TResolver:
+    def __init__(self, port):
+        self.port = port
+
+    def original_addr(self, sock):
+        return ("127.0.0.1", self.port)
+
+
+class TransparentProxTest(ProxTestBase):
+    ssl = None
+    @classmethod
+    def get_proxy_config(cls):
+        return dict(
+                transparent_proxy = dict(
+                    resolver = TResolver(cls.server.port),
+                    sslports = []
+                )
+            )
+
+    def pathod(self, spec):
+        """
+            Constructs a pathod request, with the appropriate base and proxy.
+        """
+        r = hurl.get(
+            "http://127.0.0.1:%s"%self.proxy.port + "/p/" + spec,
+            validate_cert=False,
+            #debug=hurl.utils.stdout_debug
+        )
+        return r
+
+
+class ReverseProxTest(ProxTestBase):
+    ssl = None
+    @classmethod
+    def get_proxy_config(cls):
+        return dict(
+            reverse_proxy = (
+                "https" if cls.ssl else "http",
+                "127.0.0.1",
+                cls.server.port
+            )
+        )
+
+    def pathod(self, spec):
+        """
+            Constructs a pathod request, with the appropriate base and proxy.
+        """
+        r = hurl.get(
+            "http://127.0.0.1:%s"%self.proxy.port + "/p/" + spec,
+            validate_cert=False,
+            #debug=hurl.utils.stdout_debug
+        )
+        return r
+
-- 
cgit v1.2.3