Start building a Pathod unit testing truss.

- Add test.py, which will house the testing API.
- Extend API with a shutdown method, used to terminate the test daemon.
- Refactor to allow clean shutdown.
This commit is contained in:
Aldo Cortesi 2012-06-07 11:23:23 +12:00
parent b5a74a26ee
commit 14b2a69d21
5 changed files with 100 additions and 18 deletions

View File

@ -18,6 +18,12 @@ class APILogClear(tornado.web.RequestHandler):
self.write("OK")
class APIShutdown(tornado.web.RequestHandler):
def post(self):
tornado.ioloop.IOLoop.instance().stop()
self.write("OK")
class _Page(tornado.web.RequestHandler):
def render(self, name, **kwargs):
tornado.web.RequestHandler.render(self, name + ".html", **kwargs)
@ -134,6 +140,7 @@ class PathodApp(tornado.web.Application):
(r"/log/([0-9]+)", OneLog),
(r"/help", Help),
(r"/preview", Preview),
(r"/api/shutdown", APIShutdown),
(r"/api/log", APILog),
(r"/api/log/clear", APILogClear),
(r"/p/.*", RequestPathod, settings),
@ -204,12 +211,28 @@ class PathodApp(tornado.web.Application):
return self.log
# begin nocover
def make_app(staticdir=None, anchors=()):
"""
staticdir: A directory for static assets referenced in response patterns.
anchors: A sequence of strings of the form "pattern=pagespec"
"""
settings = dict(
staticdir=staticdir
)
application = PathodApp(**settings)
for i in anchors:
rex, spec = utils.parse_anchor_spec(i, settings)
application.add_anchor(rex, spec)
return application
def make_server(application, port, address, ssl_options):
"""
Returns the bound port. This will match the passed port, unless the
passed port was 0. In that case, an arbitrary empty port will be bound
to, and this new port will be returned.
Returns a (server, port) tuple.
The returned port will match the passed port, unless the passed port
was 0. In that case, an arbitrary empty port will be bound to, and this
new port will be returned.
"""
http_server = tornado.httpserver.HTTPServer(
application,
@ -221,8 +244,12 @@ def make_server(application, port, address, ssl_options):
sn = i.getsockname()
if sn[0] == address:
port = sn[1]
return port
return http_server, port
def run():
# begin nocover
def run(server):
tornado.ioloop.IOLoop.instance().start()
server.stop()
tornado.ioloop.IOLoop.instance().close()

31
libpathod/test.py Normal file
View File

@ -0,0 +1,31 @@
import threading
import requests
import Queue
import pathod
class PaThread(threading.Thread):
def __init__(self, q, app):
threading.Thread.__init__(self)
self.q = q
self.app = app
self.port = None
def run(self):
self.server, self.port = pathod.make_server(self.app, 0, "127.0.0.1", None)
self.q.put(self.port)
pathod.run(self.server)
class Daemon:
def __init__(self, staticdir=None, anchors=()):
self.app = pathod.make_app(staticdir=staticdir, anchors=anchors)
self.q = Queue.Queue()
self.thread = PaThread(self.q, self.app)
self.thread.start()
self.port = self.q.get(True, 5)
def clear(self):
pass
def shutdown(self):
requests.post("http://localhost:%s/api/shutdown"%self.port)

13
pathod
View File

@ -31,16 +31,11 @@ if __name__ == "__main__":
help='SSL cert file. If not specified, a default cert is used.'
)
args = parser.parse_args()
settings = dict(
staticdir=args.staticdir
)
application = pathod.PathodApp(**settings)
for i in args.anchors:
try:
rex, spec = utils.parse_anchor_spec(i, settings)
app = pathod.make_app(staticdir=args.staticdir, anchors=args.anchors)
except utils.AnchorError, v:
parser.error(str(v))
application.add_anchor(rex, spec)
if args.ssl:
ssl = dict(
@ -50,8 +45,8 @@ if __name__ == "__main__":
else:
ssl = None
try:
port = pathod.make_server(application, args.port, args.address, ssl)
server, port = pathod.make_server(app, args.port, args.address, ssl)
print "%s listening on port %s"%(version.NAMEVERSION, port)
pathod.run()
pathod.run(server)
except KeyboardInterrupt:
pass

View File

@ -57,7 +57,15 @@ class uPages(libpry.AutoTree):
assert "".join(page._write_buffer)
class u_make_server(libpry.AutoTree):
def test_simple(self):
app = pathod.PathodApp()
assert pathod.make_server(app, 0, "127.0.0.1", None)
tests = [
uApplication(),
#uPages(),
u_make_server()
]

21
test/test_test.py Normal file
View File

@ -0,0 +1,21 @@
import time
import libpry
import requests
from libpathod import test
class uDaemon(libpry.AutoTree):
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()
rsp = requests.get("http://localhost:%s/p/202"%d.port)
assert not rsp.ok
tests = [
uDaemon()
]