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_start():
socks = [{"sock": None, "ssl": None}, {"sock": None, "ssl": "abc"}]
child = Child("", "", socks, "1")
with mock.patch("asyncio.Task"), mock.patch(
"blackhole.child.Child.heartbeat"
), mock.patch("{0}.run_forever".format(_LOOP)), mock.patch(
"blackhole.child.Child.stop"
), mock.patch(
"os._exit"
) as mock_exit:
child.start()
assert mock_exit.called is True
async def test_child_heartbeat_started(event_loop):
up_read, up_write = os.pipe()
down_read, down_write = os.pipe()
os.close(up_write)
os.close(down_read)
child = Child(up_read, down_write, [], "1")
child.loop = event_loop
child._started = True
sp = StreamProtocol()
sp.reader = asyncio.StreamReader()
async def _reset():
sp.reader.feed_data(protocols.PING)
child._started = False
reset_task = asyncio.Task(_reset())
with mock.patch(
"blackhole.streams.StreamProtocol", return_value=sp
), mock.patch("asyncio.Task") as mock_task, mock.patch(
"blackhole.child.Child._start"
) as mock_start, mock.patch(
"blackhole.child.Child.stop"
def test_initiation():
Child("", "", [], "1")
def test_stop():
socks = [{"sock": None, "ssl": None}, {"sock": None, "ssl": "abc"}]
child = Child("", "", socks, "1")
child.loop = mock.MagicMock()
child.clients.append(mock.MagicMock())
child.servers.append(mock.MagicMock())
child.heartbeat_task = mock.MagicMock()
child.server_task = mock.MagicMock()
with mock.patch("os._exit") as mock_exit, mock.patch(
"{0}.run_until_complete".format(_LOOP)
):
child.stop()
assert mock_exit.called is True
def test_stop_runtime_exception():
socks = [{"sock": None, "ssl": None}, {"sock": None, "ssl": "abc"}]
child = Child("", "", socks, "1")
child.loop = mock.MagicMock()
child.clients.append(mock.MagicMock())
child.servers.append(mock.MagicMock())
child.heartbeat_task = mock.MagicMock()
child.server_task = mock.MagicMock()
with mock.patch("os._exit") as mock_exit, mock.patch(
"{0}.stop".format(_LOOP), side_effect=RuntimeError
):
child.stop()
assert mock_exit.called is True
async def test_start_child_loop(event_loop):
sock = _socket("127.0.0.1", 0, socket.AF_INET)
socks = ({"sock": sock, "ssl": None},)
child = Child("", "", socks, "1")
child.loop = event_loop
await child._start()
assert len(child.servers) == 1
for server in child.servers:
server.close()
def setup_child(self):
"""Basic setup for the child process and starting it."""
setgid()
setuid()
asyncio.set_event_loop(None)
if setproctitle:
setproctitle.setproctitle("blackhole: worker")
process = Child(self.up_read, self.down_write, self.socks, self.idx)
process.start()