Dmitri Bogomolov
7b9824afc9
for settings in the generated config, which particularly checks that extralowdifficulty settings are not applied to daemon.
135 lines
3.9 KiB
Python
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()
|