How to use the waitress.wasyncore.dispatcher function in waitress

To help you get started, we’ve selected a few waitress examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github Pylons / waitress / tests / test_wasyncore.py View on Github external
def _makeOne(self, sock=None, map=None):
        from waitress.wasyncore import dispatcher

        return dispatcher(sock=sock, map=map)
github Pylons / waitress / tests / test_wasyncore.py View on Github external
def __init__(self, sock=None):
        asyncore.dispatcher.__init__(self, sock)
        self.flag = False
github Pylons / waitress / tests / test_wasyncore.py View on Github external
raise Exception("handle_accepted not supposed to be called")

    def handle_connect(self):
        raise Exception("handle_connect not supposed to be called")

    def handle_expt(self):
        raise Exception("handle_expt not supposed to be called")

    def handle_close(self):
        raise Exception("handle_close not supposed to be called")

    def handle_error(self):
        raise


class BaseServer(asyncore.dispatcher):
    """A server which listens on an address and dispatches the
    connection to a handler.
    """

    def __init__(self, family, addr, handler=BaseTestHandler):
        asyncore.dispatcher.__init__(self)
        self.create_socket(family)
        self.set_reuse_addr()
        bind_af_aware(self.socket, addr)
        self.listen(5)
        self.handler = handler

    @property
    def address(self):
        return self.socket.getsockname()
github Pylons / waitress / tests / test_wasyncore.py View on Github external
def test_repr(self):
        d = asyncore.dispatcher()
        self.assertEqual(repr(d), "" % id(d))
github Pylons / waitress / tests / test_wasyncore.py View on Github external
def test_close_twice(self):
        fd = os.open(TESTFN, os.O_RDONLY)
        f = asyncore.file_wrapper(fd)
        os.close(fd)

        os.close(f.fd)  # file_wrapper dupped fd
        with self.assertRaises(OSError):
            f.close()

        self.assertEqual(f.fd, -1)
        # calling close twice should not fail
        f.close()


class BaseTestHandler(asyncore.dispatcher):  # pragma: no cover
    def __init__(self, sock=None):
        asyncore.dispatcher.__init__(self, sock)
        self.flag = False

    def handle_accept(self):
        raise Exception("handle_accept not supposed to be called")

    def handle_accepted(self):
        raise Exception("handle_accepted not supposed to be called")

    def handle_connect(self):
        raise Exception("handle_connect not supposed to be called")

    def handle_expt(self):
        raise Exception("handle_expt not supposed to be called")
github Pylons / waitress / tests / test_wasyncore.py View on Github external
def handle_accept(self):
                asyncore.dispatcher.handle_accept(self)
github Pylons / waitress / tests / test_wasyncore.py View on Github external
def test_unhandled(self):
        import logging

        inst = asyncore.dispatcher()
        logger = DummyLogger()
        inst.logger = logger

        inst.handle_expt()
        inst.handle_read()
        inst.handle_write()
        inst.handle_connect()

        expected = [
            (logging.WARN, "unhandled incoming priority event"),
            (logging.WARN, "unhandled read event"),
            (logging.WARN, "unhandled write event"),
            (logging.WARN, "unhandled connect event"),
        ]
        self.assertEqual(logger.messages, expected)
github Pylons / waitress / tests / test_wasyncore.py View on Github external
def test_set_reuse_addr(self):  # pragma: no cover
        if HAS_UNIX_SOCKETS and self.family == socket.AF_UNIX:
            self.skipTest("Not applicable to AF_UNIX sockets.")

        with closewrapper(socket.socket(self.family)) as sock:
            try:
                sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
            except OSError:
                unittest.skip("SO_REUSEADDR not supported on this platform")
            else:
                # if SO_REUSEADDR succeeded for sock we expect asyncore
                # to do the same
                s = asyncore.dispatcher(socket.socket(self.family))
                self.assertFalse(
                    s.socket.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR)
                )
                s.socket.close()
                s.create_socket(self.family)
                s.set_reuse_addr()
                self.assertTrue(
                    s.socket.getsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR)
                )
github Pylons / waitress / tests / test_wasyncore.py View on Github external
def test_log_info(self):
        import logging

        inst = asyncore.dispatcher(map={})
        logger = DummyLogger()
        inst.logger = logger
        inst.log_info("message", "warning")
        self.assertEqual(logger.messages, [(logging.WARN, "message")])