PyBitmessage-2021-04-27/src/tests/test_process.py
Dmitri Bogomolov 7b9824afc9
Separate method for stopping the bitmessage process and new test case
for settings in the generated config, which particularly checks that
extralowdifficulty settings are not applied to daemon.
2019-04-22 12:19:09 +03:00

135 lines
3.9 KiB
Python

"""
Common reusable code for tests and tests for pybitmessage process.
"""
import os
import signal
import subprocess # nosec
import tempfile
import time
import unittest
import psutil
def put_signal_file(path, filename):
"""Creates file, presence of which is a signal about some event."""
with open(os.path.join(path, filename), 'wb') as outfile:
outfile.write(str(time.time()))
class TestProcessProto(unittest.TestCase):
"""Test case implementing common logic for external testing:
it starts pybitmessage in setUpClass() and stops it in tearDownClass()
"""
_process_cmd = ['pybitmessage', '-d']
_threads_count = 15
_files = (
'keys.dat', 'debug.log', 'messages.dat', 'knownnodes.dat',
'.api_started', 'unittest.lock'
)
@classmethod
def setUpClass(cls):
"""Setup environment and start pybitmessage"""
cls.home = os.environ['BITMESSAGE_HOME'] = tempfile.gettempdir()
put_signal_file(cls.home, 'unittest.lock')
subprocess.call(cls._process_cmd) # nosec
time.sleep(5)
cls.pid = int(cls._get_readline('singleton.lock'))
cls.process = psutil.Process(cls.pid)
@classmethod
def _get_readline(cls, pfile):
pfile = os.path.join(cls.home, pfile)
try:
return open(pfile, 'rb').readline().strip()
except (OSError, IOError):
pass
@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 _cleanup_files(cls):
for pfile in cls._files:
try:
os.remove(os.path.join(cls.home, pfile))
except OSError:
pass
@classmethod
def tearDownClass(cls):
"""Ensures that pybitmessage stopped and removes files"""
try:
if not cls._stop_process():
print(open(os.path.join(cls.home, 'debug.log'), 'rb').read())
cls.process.kill()
except psutil.NoSuchProcess:
pass
finally:
cls._cleanup_files()
def _test_threads(self):
# only count for now
# because of https://github.com/giampaolo/psutil/issues/613
# PyBitmessage
# - addressGenerator
# - singleWorker
# - SQL
# - objectProcessor
# - singleCleaner
# - singleAPI
# - Asyncore
# - ReceiveQueue_0
# - ReceiveQueue_1
# - ReceiveQueue_2
# - Announcer
# - InvBroadcaster
# - AddrBroadcaster
# - Downloader
self.assertEqual(
len(self.process.threads()), self._threads_count)
class TestProcessShutdown(TestProcessProto):
"""Separate test case for SIGTERM"""
def test_shutdown(self):
"""Send to pybitmessage SIGTERM and ensure it stopped"""
# longer wait time because it's not a benchmark
self.assertTrue(
self._stop_process(20),
'%s has not stopped in 20 sec' % ' '.join(self._process_cmd))
@classmethod
def tearDownClass(cls):
"""Special teardown because pybitmessage is already stopped"""
cls._cleanup_files()
class TestProcess(TestProcessProto):
"""A test case for pybitmessage process"""
def test_process_name(self):
"""Check PyBitmessage process name"""
self.assertEqual(self.process.name(), 'PyBitmessage')
def test_files(self):
"""Check existence of PyBitmessage files"""
for pfile in self._files:
if pfile.startswith('.'):
continue
self.assertIsNot(
self._get_readline(pfile), None,
'Failed to read file %s' % pfile
)
def test_threads(self):
"""Testing PyBitmessage threads"""
self._test_threads()