MiNode/minode/tests/test_process.py

157 lines
4.9 KiB
Python
Raw Normal View History

2023-08-01 23:15:00 +02:00
"""Blind tests, starting the minode process"""
import unittest
import signal
import socket
import subprocess
2021-08-02 15:45:25 +02:00
import sys
import tempfile
import time
import psutil
from minode.structure import NetAddrNoPrefix
try:
socket.socket().bind(('127.0.0.1', 7656))
i2p_port_free = True
except (OSError, socket.error):
i2p_port_free = False
class TestProcessProto(unittest.TestCase):
"""Test process attributes, common flow"""
_process_cmd = ['minode']
2022-09-23 03:25:24 +02:00
_connection_limit = 4 if sys.platform.startswith('win') else 10
_listen = False
_listening_port = None
home = None
@classmethod
def setUpClass(cls):
if not cls.home:
cls.home = tempfile.gettempdir()
cmd = cls._process_cmd + [
'--data-dir', cls.home,
'--connection-limit', str(cls._connection_limit)
]
2022-09-23 03:25:24 +02:00
if not cls._listen:
cmd += ['--no-incoming']
2022-09-23 03:25:24 +02:00
elif cls._listening_port:
cmd += ['-p', str(cls._listening_port)]
cls.process = psutil.Popen(cmd, stderr=subprocess.STDOUT) # nosec
@classmethod
def _stop_process(cls, timeout=5):
cls.process.send_signal(signal.SIGTERM)
try:
cls.process.wait(timeout)
except psutil.TimeoutExpired:
return False
return True
@classmethod
def tearDownClass(cls):
2023-08-01 23:15:00 +02:00
"""Ensures that process stopped and removes files"""
try:
if not cls._stop_process(10):
try:
cls.process.kill()
except psutil.NoSuchProcess:
pass
except psutil.NoSuchProcess:
pass
2023-08-02 19:53:36 +02:00
def connections(self):
"""All process' established connections"""
return [
c for c in self.process.connections()
if c.status == 'ESTABLISHED']
class TestProcessShutdown(TestProcessProto):
"""Separate test case for SIGTERM"""
_wait_time = 30
# longer wait time because it's not a benchmark
def test_shutdown(self):
"""Send to minode SIGTERM and ensure it stopped"""
self.assertTrue(
self._stop_process(self._wait_time),
'%s has not stopped in %i sec' % (
' '.join(self._process_cmd), self._wait_time))
class TestProcess(TestProcessProto):
"""The test case for minode process"""
_wait_time = 120
_check_limit = False
def test_connections(self):
"""Check minode process connections"""
_started = time.time()
2022-09-23 03:25:24 +02:00
def continue_check_limit(extra_time):
2023-08-02 04:17:58 +02:00
for _ in range(extra_time * 2):
self.assertLessEqual(
2023-08-02 19:53:36 +02:00
len(self.connections()),
# shared.outgoing_connections, one listening
# TODO: find the cause of one extra
(min(self._connection_limit, 8) if not self._listen
else self._connection_limit) + 1,
'Opened more connections than required'
' by --connection-limit')
time.sleep(1)
2023-08-02 04:17:58 +02:00
for _ in range(self._wait_time * 2):
2023-08-02 19:53:36 +02:00
if len(self.connections()) > self._connection_limit / 2:
_time_to_connect = round(time.time() - _started)
break
if '--i2p' not in self._process_cmd:
groups = []
for c in self.connections():
group = NetAddrNoPrefix.network_group(c.raddr[0])
self.assertNotIn(group, groups)
groups.append(group)
time.sleep(0.5)
else:
self.fail(
'Failed establish at least %i connections in %s sec'
% (int(self._connection_limit / 2), self._wait_time))
if self._check_limit:
continue_check_limit(_time_to_connect)
2022-09-23 03:25:24 +02:00
for c in self.process.connections():
if c.status == 'LISTEN':
if self._listen is False:
self.fail('Listening while started with --no-incoming')
return
self.assertEqual(c.laddr[1], self._listening_port or 8444)
break
2022-09-23 03:25:24 +02:00
else:
if self._listen:
self.fail('No listening connection found')
@unittest.skipIf(i2p_port_free, 'No running i2pd detected')
class TestProcessI2P(TestProcess):
"""Test minode process with --i2p and no IP"""
_process_cmd = ['minode', '--i2p', '--no-ip']
_connection_limit = 4
_listen = True
_listening_port = 8448
2023-08-02 19:53:36 +02:00
def test_connections(self):
"""Ensure all connections are I2P"""
super().test_connections()
for c in self.connections():
self.assertEqual(c.raddr[0], '127.0.0.1')
self.assertEqual(c.raddr[1], 7656)
@unittest.skipUnless(i2p_port_free, 'Detected running i2pd')
class TestProcessNoI2P(TestProcessShutdown):
"""Test minode process shutdown with --i2p and no IP"""
_process_cmd = ['minode', '--i2p', '--no-ip']