magicbus / magicbus / test / test_servers.py

Robert Brewer 41f5dfb 




































from magicbus import bus
from magicbus.wspbus import ChannelFailures
from magicbus.plugins import loggers, servers
from magicbus.test import assertEqual, WebService, WebHandler

#loggers.StdoutLogger(bus).subscribe()


class Handler(WebHandler):

    bus = bus

    def do_GET(self):
        if self.path == '/':
            self.respond("Hello World")
        elif self.path == '/ctrlc':
            self.respond("okey-doke")
            raise KeyboardInterrupt
        elif self.path == '/exit':
            self.respond("ok")
            self.bus.exit()
        else:
            self.respond(status=404)
service = WebService(handler_class=Handler)
adapter = servers.ServerPlugin(bus, service, service.address)
adapter.subscribe()


class TestServers(object):

    def test_keyboard_interrupt(self):
        # Raise a keyboard interrupt in the HTTP server's main thread.
        bus.start()
        resp = service.do_GET("/ctrlc")
        assertEqual(resp.status, 200)
        bus.block()
        assertEqual(bus.state, bus.states.EXITING)
Tip: Filter by directory path e.g. /media app.js to search for public/media/app.js.
Tip: Use camelCasing e.g. ProjME to search for ProjectModifiedEvent.java.
Tip: Filter by extension type e.g. /repo .js to search for all .js files in the /repo directory.
Tip: Separate your search with spaces e.g. /ssh pom.xml to search for src/ssh/pom.xml.
Tip: Use ↑ and ↓ arrow keys to navigate and return to view the file.
Tip: You can also navigate files with Ctrl+j (next) and Ctrl+k (previous) and view the file with Ctrl+o.
Tip: You can also navigate files with Alt+j (next) and Alt+k (previous) and view the file with Alt+o.