diff options
| -rw-r--r-- | libpathod/pathod.py | 31 | ||||
| -rw-r--r-- | libpathod/test.py | 8 | ||||
| -rw-r--r-- | test/test_pathod.py | 6 | ||||
| -rw-r--r-- | test/test_test.py | 12 |
4 files changed, 45 insertions, 12 deletions
diff --git a/libpathod/pathod.py b/libpathod/pathod.py index ef37d5ad..1491072c 100644 --- a/libpathod/pathod.py +++ b/libpathod/pathod.py @@ -1,4 +1,4 @@ -import urllib, threading +import urllib, threading, re from netlib import tcp, http, odict, wsgi import version, app, rparse @@ -23,16 +23,23 @@ class PathodHandler(tcp.BaseHandler): self.rfile, self.wfile, headers, httpversion, None ) - if path.startswith(self.server.prefix): + crafted = None + for i in self.server.anchors: + if i[0].match(path): + crafted = i[1] + + if not crafted and path.startswith(self.server.prefix): spec = urllib.unquote(path)[len(self.server.prefix):] try: - presp = rparse.parse(self.server.request_settings, spec) + crafted = rparse.parse(self.server.request_settings, spec) except rparse.ParseException, v: - presp = rparse.InternalResponse( + crafted = rparse.InternalResponse( 800, "Error parsing response spec: %s\n"%v.msg + v.marked() ) - ret = presp.serve(self.wfile) + + if crafted: + ret = crafted.serve(self.wfile) if ret["disconnect"]: self.finish() ret["request"] = dict( @@ -62,6 +69,14 @@ class PathodHandler(tcp.BaseHandler): class Pathod(tcp.TCPServer): LOGBUF = 500 def __init__(self, addr, ssloptions=None, prefix="/p/", staticdir=None, anchors=None): + """ + addr: (address, port) tuple. If port is 0, a free port will be + automatically chosen. + ssloptions: a dictionary containing certfile and keyfile specifications. + prefix: string specifying the prefix at which to anchor response generation. + staticdir: path to a directory of static resources, or None. + anchors: A list of (regex, spec) tuples, or None. + """ tcp.TCPServer.__init__(self, addr) self.ssloptions = ssloptions self.staticdir = staticdir @@ -70,6 +85,12 @@ class Pathod(tcp.TCPServer): self.app.config["pathod"] = self self.log = [] self.logid = 0 + self.anchors = [] + if anchors: + for i in anchors: + arex = re.compile(i[0]) + aresp = rparse.parse(self.request_settings, i[1]) + self.anchors.append((arex, aresp)) @property def request_settings(self): diff --git a/libpathod/test.py b/libpathod/test.py index cb1b9745..00e03823 100644 --- a/libpathod/test.py +++ b/libpathod/test.py @@ -8,7 +8,7 @@ IFACE = "127.0.0.1" class Daemon: def __init__(self, staticdir=None, anchors=(), ssl=None): self.q = Queue.Queue() - self.thread = PaThread(self.q, ssl, staticdir) + self.thread = PaThread(self.q, staticdir, anchors, ssl) self.thread.start() self.port = self.q.get(True, 5) self.urlbase = "%s://%s:%s"%("https" if ssl else "http", IFACE, self.port) @@ -43,10 +43,9 @@ class Daemon: class PaThread(threading.Thread): - def __init__(self, q, ssl, staticdir): + def __init__(self, q, staticdir, anchors, ssl): threading.Thread.__init__(self) - self.q, self.ssl, self.staticdir = q, ssl, staticdir - self.port = None + self.q, self.staticdir, self.anchors, self.ssl = q, staticdir, anchors, ssl def run(self): if self.ssl is True: @@ -59,6 +58,7 @@ class PaThread(threading.Thread): self.server = pathod.Pathod( (IFACE, 0), ssloptions = ssloptions, + anchors = self.anchors, staticdir = self.staticdir ) self.q.put(self.server.port) diff --git a/test/test_pathod.py b/test/test_pathod.py index 966ae12e..e00694cd 100644 --- a/test/test_pathod.py +++ b/test/test_pathod.py @@ -15,7 +15,11 @@ class _TestApplication: class TestPathod: def test_instantiation(self): - p = pathod.Pathod(("127.0.0.1", 0)) + p = pathod.Pathod( + ("127.0.0.1", 0), + anchors = [(".*", "200")] + ) + assert p.anchors def test_logging(self): p = pathod.Pathod(("127.0.0.1", 0)) diff --git a/test/test_test.py b/test/test_test.py index b9a9cfac..dcf980bc 100644 --- a/test/test_test.py +++ b/test/test_test.py @@ -38,7 +38,10 @@ class TestDaemonManual: class TestDaemon: @classmethod def setUpAll(self): - self.d = test.Daemon(staticdir=tutils.test_data.path("data")) + self.d = test.Daemon( + staticdir=tutils.test_data.path("data"), + anchors=[("/anchor/.*", "202")] + ) @classmethod def tearDownAll(self): @@ -47,6 +50,9 @@ class TestDaemon: def setUp(self): self.d.clear_log() + def getpath(self, path): + return requests.get("http://localhost:%s/%s"%(self.d.port, path)) + def get(self, spec): return requests.get("http://localhost:%s/p/%s"%(self.d.port, spec)) @@ -72,5 +78,7 @@ class TestDaemon: assert rsp.status_code == 200 assert rsp.content.strip() == "testfile" - + def test_anchor(self): + rsp = self.getpath("anchor/foo") + assert rsp.status_code == 202 |