Bitmessage/PyBitmessage-2024-12-20
Archived
2
0
Fork 0
Code Issues 168 Pull Requests 52 Actions Packages Projects Releases 5 Wiki Activity

implemented process killing feature for test cases

Browse Source
This commit is contained in:
navjot 2021-03-11 17:38:09 +05:30
parent 9ff2e4a605
commit 50a258f412
No known key found for this signature in database
GPG Key ID: 9EE70AFD71357F1C
7 changed files with 8 additions and 275 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

39
src/bitmessagekivy/tests/telenium_process.py
View File

@ -1,9 +1,8 @@
import os, signal
import psutil
import os
import shutil
import tempfile
from telenium.tests import TeleniumTestCase
from threads import addressGenerator, sqlThread
_files = (
@ -22,15 +21,13 @@ def cleanup(home=None, files=_files):
home = tempfile.gettempdir()
for pfile in files:
try:
# import pdb;pdb.set_trace()
os.remove(os.path.join(home, pfile))
print(__file__,'.........................................(clean)', pfile)
except OSError:
# print('error............................................')
pass
def set_temp_data():
"""Set temp data in tmp directory"""
for file in tmp_db_file:
old_source_file = os.path.join(
os.path.abspath(os.path.dirname(__file__)), 'sampleData', file)
@ -44,7 +41,7 @@ class TeleniumTestProcess(TeleniumTestCase):
@classmethod
def setUpClass(cls):
print('XXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXXX', __file__)
"""Setupclass is for setting temp environment"""
os.environ["BITMESSAGE_HOME"] = tempfile.gettempdir()
set_temp_data()
super(TeleniumTestProcess, cls).setUpClass()
@ -52,34 +49,14 @@ class TeleniumTestProcess(TeleniumTestCase):
@classmethod
def tearDownClass(cls):
"""Ensures that pybitmessage stopped and removes files"""
# super(TeleniumTestProcess, cls).tearDownClass()
print('tearDownClass.........................................(pass)', os.getpid())
cleanup()
pid = os.getpid()
os.kill(int(pid), signal.SIGKILL)
# import pdb;pdb.set_trace()
# # os.kill()
# # import psutil
# # cnt = 0
# # # print('total count=========================================', len(psutil.process_iter()))
# # for proc in psutil.process_iter():
# # print('line...................................62', proc.name(), proc.pid)
# # cnt = cnt +1
# # if proc.name() == 'python':
# # print('line........................69', proc.pid)
# # # os.kill(int(pid), signal.SIGKILL)
# # print('total cnt.............................', cnt)
# # # cleanup()
# # proc.kill()
# # # os.kill(int(proc.pid), signal.SIGKILL)
cls.cli.app_quit()
cls.process.kill()
@classmethod
def setUp(self):
# self.widget = Widget('The widget')
print('&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&7setup')
pass
@classmethod
def tearDown(self):
print('###############################################tearDown')
# self.widget.dispose()
pass

13
src/bitmessagekivy/tests/test_draft_message.py
View File

@ -91,16 +91,3 @@ class DraftMessage(TeleniumTestProcess):
time.sleep(5)
self.cli.click_on('//MDToolbar/BoxLayout[2]/MDIconButton[1]')
time.sleep(2)
# def test_all_draft_method(self):
# """Calling All The Methods Draft Class"""
# self.test_select_draft_message()
# self.test_edit_draft_messgae()
# self.test_delete_draft_message()
# if __name__ == '__main__':
# """Start Application"""
# obj = DraftMessage()
# obj.setUpClass()
# obj.test_all_draft_method()

7
src/bitmessagekivy/tests/test_network_screen.py
View File

@ -17,10 +17,3 @@ class NetwrokStatusScreen(TeleniumTestProcess):
time.sleep(4)
self.cli.click_on('//NetworkStat/MDTabs[0]/MDTabsBar[0]/MDTabsScrollView[0]/MDGridLayout[0]/MDTabsLabel[1]')
time.sleep(4)
# if __name__ == '__main__':
# """Start Application"""
# obj = NetwrokStatusScreen()
# obj.setUpClass()
# obj.test_network_status()

7
src/bitmessagekivy/tests/test_payment_subscription.py
View File

@ -24,10 +24,3 @@ class PaymentScreen(TeleniumTestProcess):
time.sleep(2)
self.cli.click_on('//MDRaisedButton[3]')
time.sleep(2)
# if __name__ == '__main__':
# """Start Application"""
# obj = PaymentScreen()
# obj.setUpClass()
# obj.test_select_subscripton()

203
src/bitmessagekivy/tests/test_process_data.py
View File

@ -1,203 +0,0 @@
"""
Common reusable code for tests and tests for pybitmessage process.
"""
import os
import signal
import subprocess # nosec
import sys
import state
import tempfile
import time
import unittest
import psutil
from .common_data import cleanup, put_signal_file, skip_python3
skip_python3()
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_min = 15
_threads_count_max = 16
_threads_names = [
'PyBitmessage',
'addressGenerato',
'singleWorker',
'SQL',
'objectProcessor',
'singleCleaner',
'singleAPI',
'Asyncore',
'ReceiveQueue_0',
'ReceiveQueue_1',
'ReceiveQueue_2',
'Announcer',
'InvBroadcaster',
'AddrBroadcaster',
'Downloader',
'Uploader'
]
_files = (
'keys.dat', 'debug.log', 'messages.dat', 'knownnodes.dat',
'.api_started', 'unittest.lock'
)
home = None
@classmethod
def setUpClass(cls):
"""Setup environment and start pybitmessage"""
print('setUpClass.........................................(pass)', state.appdata)
# print('os.environ["BITMESSAGE_HOME"](test_process file)57...........................', os.environ['BITMESSAGE_HOME'])
print(__file__, 'cls.home..................................', cls.home)
cls.flag = False
# if os.environ.get('BITMESSAGE_HOME'):
# cls._cleanup_files()
# cls.home = os.environ.get('BITMESSAGE_HOME')
# else:
# if not cls.home:
# cls.home = tempfile.gettempdir()
# cls._cleanup_files()
# os.environ['BITMESSAGE_HOME'] = cls.home
cls.flag = False
if not cls.home:
cls.home = tempfile.gettempdir()
cls._cleanup_files()
os.environ['BITMESSAGE_HOME'] = cls.home
print('os.environ["BITMESSAGE_HOME"](test_process file)65...........................', os.environ['BITMESSAGE_HOME'])
put_signal_file(cls.home, 'unittest.lock')
starttime = int(time.time()) - 0.5
cls.process = psutil.Popen(
cls._process_cmd, stderr=subprocess.STDOUT) # nosec
pidfile = os.path.join(cls.home, 'singleton.lock')
for _ in range(10):
time.sleep(1)
try:
pstat = os.stat(pidfile)
if starttime <= pstat.st_mtime and pstat.st_size > 0:
break # the pidfile is suitable
except OSError:
continue
try:
pid = int(cls._get_readline('singleton.lock'))
cls.process = psutil.Process(pid)
time.sleep(5)
except (psutil.NoSuchProcess, TypeError):
cls.flag = True
def setUp(self):
print('setUp.........................................(pass)')
if self.flag:
self.fail("%s is not started ):" % self._process_cmd)
@classmethod
def _get_readline(cls, pfile):
print('_get_readline.........................................(pass)')
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):
print('_stop_process.........................................(pass)')
cls.process.send_signal(signal.SIGTERM)
try:
cls.process.wait(timeout)
except psutil.TimeoutExpired:
return False
return True
@classmethod
def _kill_process(cls, timeout=5):
print('_kill_process.........................................')
try:
cls.process.send_signal(signal.SIGKILL)
cls.process.wait(timeout)
# Windows or already dead
except (AttributeError, psutil.NoSuchProcess):
return True
# except psutil.TimeoutExpired propagates, it means something is very
# wrong
return True
@classmethod
def _cleanup_files(cls):
print('_cleanup_files.........................................(pass)')
cleanup(cls.home, cls._files)
@classmethod
def tearDownClass(cls):
"""Ensures that pybitmessage stopped and removes files"""
print('tearDownClass.........................................(pass)')
try:
if not cls._stop_process(10):
processes = cls.process.children(recursive=True)
processes.append(cls.process)
for p in processes:
try:
p.kill()
except psutil.NoSuchProcess:
pass
except psutil.NoSuchProcess:
pass
finally:
cls._cleanup_files()
def _test_threads(self):
"""Test number and names of threads"""
# pylint: disable=invalid-name
print('_test_threads.........................................(pass)')
self.longMessage = True
try:
# using ps for posix platforms
# because of https://github.com/giampaolo/psutil/issues/613
thread_names = subprocess.check_output([
"ps", "-L", "-o", "comm=", "--pid",
str(self.process.pid)
]).split()
except: # pylint: disable=bare-except
thread_names = []
running_threads = len(thread_names)
if 0 < running_threads < 30: # adequacy check
extra_threads = []
missing_threads = []
for thread_name in thread_names:
if thread_name not in self._threads_names:
extra_threads.append(thread_name)
for thread_name in self._threads_names:
if thread_name not in thread_names:
missing_threads.append(thread_name)
msg = "Missing threads: {}, Extra threads: {}".format(
",".join(missing_threads), ",".join(extra_threads))
else:
running_threads = self.process.num_threads()
if sys.platform.startswith('win'):
running_threads -= 1 # one extra thread on Windows!
msg = "Unexpected running thread count"
self.assertGreaterEqual(
running_threads,
self._threads_count_min,
msg)
self.assertLessEqual(
running_threads,
self._threads_count_max,
msg)
def remove_temp_data(self):
cleanup()

7
src/bitmessagekivy/tests/test_setting_screen.py
View File

@ -19,10 +19,3 @@ class SettingScreen(TeleniumTestProcess):
time.sleep(3)
self.cli.click_on('//NavigationItem[7]')
time.sleep(2)
# if __name__ == '__main__':
# """Start Application"""
# obj = SettingScreen()
# obj.setUpClass()
# obj.test_setting_screen()
# obj.tearDownClass()

7
src/bitmessagekivy/tests/test_trash_message.py
View File

@ -20,10 +20,3 @@ class TrashMessage(TeleniumTestProcess):
time.sleep(2)
self.cli.click_on('//MDDialog/MDCard[0]/AnchorLayout[0]/MDBoxLayout[0]/MDFlatButton[0]')
time.sleep(4)
# if __name__ == '__main__':
# """Start Application"""
# obj = TrashMessage()
# obj.setUpClass()
# obj.test_delete_trash_message()