Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _makeOne(self, sock=None, map=None):
from waitress.wasyncore import dispatcher
return dispatcher(sock=sock, map=map)
def __init__(self, sock=None):
asyncore.dispatcher.__init__(self, sock)
self.flag = False
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()
def test_repr(self):
d = asyncore.dispatcher()
self.assertEqual(repr(d), "" % id(d))
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")
def handle_accept(self):
asyncore.dispatcher.handle_accept(self)
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)
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)
)
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")])