This repository has been archived on 2024-12-17. You can view files and clone it, but cannot push or open issues or pull requests.
PyBitmessage-2024-12-17/src/tests/test_process.py
Peter Surda ca567acab3
Put uploads into a separate thread
- instead of being processed in the ReceiveQueue thread, uploads are now done
  in a dedicated thread. Only the parsing is done in ReceiveQueue thread.
- the UploadThread is modelled based on the DownloadThred, but simpler.
- it checks for intersection attack, eliminates duplicates and restricts the
  write buffer size to 2MB (may still grow slightly higher if too many big
  objects are requested, but the absolute limit appears to be about 4.5MB in the
  worst case scenario).
- the restriction of the write buffer may cause some upload throttling (to
  about 2MB per second per connection), but can be optimised later
- fixes #1414
2018-12-20 20:34:01 +01:00

111 lines
3.2 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 _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"""
cls.process.send_signal(signal.SIGTERM)
try:
cls.process.wait(5)
except psutil.TimeoutExpired:
print(open(os.path.join(cls.home, 'debug.log'), 'rb').read())
cls.process.kill()
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 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()