Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_create_server_ipv4_bind_works(mock_sock):
cfile = create_config(("listen=127.0.0.1:9000",))
Config(cfile).load()
_server = server("127.0.0.1", 9000, socket.AF_INET)
assert mock_sock.called is True
assert mock_sock.call_count == 1
_server["sock"].close()
async def test_worker_ping_pong(unused_tcp_port):
aserver = server("127.0.0.1", unused_tcp_port, socket.AF_INET)
started = time.monotonic()
worker = Worker("1", [aserver])
assert worker._started is True
await asyncio.sleep(35)
worker.stop()
assert worker._started is False
assert worker.ping > started
assert worker.ping_count == 2
aserver["sock"].close()
def test_create_server_ipv4_bind_fails():
cfile = create_config(("listen=127.0.0.1:9000",))
Config(cfile).load()
with mock.patch(
"socket.socket.bind", side_effect=OSError
) as mock_sock, pytest.raises(BlackholeRuntimeException):
server("127.0.0.1", 9000, socket.AF_INET, {})
assert mock_sock.called is True
assert mock_sock.call_count == 1
def test_create_server_ipv6_bind_fails():
cfile = create_config(("listen=:::9000",))
Config(cfile).load()
with mock.patch(
"socket.socket.bind", side_effect=OSError
) as mock_sock, pytest.raises(BlackholeRuntimeException):
_server = server("::", 9000, socket.AF_INET6, {})
_server["sock"].close()
assert mock_sock.called is True
assert mock_sock.call_count == 1
def test_create_server_ipv6_bind_works():
cfile = create_config(("listen=:::9000",))
Config(cfile).load()
with mock.patch("socket.socket.bind") as mock_sock:
_server = server("::", 9000, socket.AF_INET6)
_server["sock"].close()
assert mock_sock.called is True
assert mock_sock.call_count == 1
def test_create_server_ipv4_tls_bind_fails():
cfile = create_config(("tls_listen=127.0.0.1:9000",))
Config(cfile).load()
with mock.patch(
"socket.socket.bind", side_effect=OSError
) as mock_sock, pytest.raises(BlackholeRuntimeException):
server("127.0.0.1", 9000, socket.AF_INET)
assert mock_sock.called is True
assert mock_sock.call_count == 1
def test_create_server_tls_ipv4_bind_works():
cfile = create_config(("listen=127.0.0.1:25", "tls_listen=127.0.0.1:9000"))
conf = Config(cfile).load()
conf.args = Args((("less_secure", False),))
with mock.patch("socket.socket.bind") as mock_sock, mock.patch(
"ssl.create_default_context"
) as mock_ssl:
_server = server("127.0.0.1", 9000, socket.AF_INET, use_tls=True)
_server["sock"].close()
assert mock_sock.called is True
assert mock_sock.call_count == 1
assert mock_ssl.called is True
assert mock_ssl.call_count == 1
def test_create_server_ipv6_tls_bind_fails():
cfile = create_config(("tls_listen=:::9000",))
Config(cfile).load()
with mock.patch(
"socket.socket.bind", side_effect=OSError
) as mock_sock, pytest.raises(BlackholeRuntimeException):
server("::", 9000, socket.AF_INET6)
assert mock_sock.called is True
assert mock_sock.call_count == 1
def create_socket(self, listeners, use_tls=False):
"""Create supervisor socket."""
_tls = ""
if use_tls:
_tls = " (TLS)"
for host, port, family, flags in listeners:
aserver = server(host, port, family, use_tls=use_tls)
self.socks.append(aserver)
logger.debug(f"Attaching {host}:{port}{_tls} with flags {flags}")