1"""py.test config""" 2 3import os 4 5import pytest 6 7from circuits.net.sockets import close 8from circuits.web import Server, Static 9from circuits import handler, Component, Debugger 10from circuits.web.client import Client, request 11 12 13DOCROOT = os.path.join(os.path.dirname(__file__), "static") 14 15 16class WebApp(Component): 17 18 channel = "web" 19 20 def init(self): 21 self.closed = False 22 23 self.server = Server(0).register(self) 24 Static("/static", DOCROOT, dirlisting=True).register(self) 25 26 27class WebClient(Client): 28 29 def init(self, *args, **kwargs): 30 self.closed = False 31 32 def __call__(self, method, path, body=None, headers={}): 33 waiter = pytest.WaitEvent(self, "response", channel=self.channel) 34 self.fire(request(method, path, body, headers)) 35 assert waiter.wait() 36 37 return self.response 38 39 @handler("closed", channel="*", priority=1.0) 40 def _on_closed(self): 41 self.closed = True 42 43 44@pytest.fixture 45def webapp(request, manager, watcher): 46 webapp = WebApp().register(manager) 47 assert watcher.wait("ready") 48 49 if hasattr(request.module, "application"): 50 from circuits.web.wsgi import Gateway 51 application = getattr(request.module, "application") 52 Gateway({"/": application}).register(webapp) 53 assert watcher.wait("registered") 54 55 Root = getattr(request.module, "Root", None) 56 if Root is not None: 57 Root().register(webapp) 58 assert watcher.wait("registered") 59 60 if request.config.option.verbose: 61 Debugger().register(webapp) 62 assert watcher.wait("registered") 63 64 def finalizer(): 65 webapp.fire(close()) 66 assert watcher.wait("closed") 67 68 webapp.unregister() 69 assert watcher.wait("unregistered") 70 71 request.addfinalizer(finalizer) 72 73 return webapp 74