Khaganat-NeL/opennel-pymanager
Khaganat-NeL/opennel-pymanager
2
0
Fork 0
Code Issues 3 Pull requests Projects Releases Packages Wiki Activity

adding multiprocess check

Browse source
This commit is contained in:
AleaJactaEst 2018-02-09 23:29:36 +01:00
parent 4a5bd254e7
commit e5fe6e5388
1 changed files with 135 additions and 71 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

204
tests/test_manager.py
View file

@ -25,6 +25,7 @@ import multiprocessing
import time import time
import re import re
import queue import queue
import signal
from unittest.mock import patch from unittest.mock import patch
try: try:
@ -41,6 +42,12 @@ except ImportError:
# sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) # sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
# import pymanager.certificate as cert # import pymanager.certificate as cert
def handler(signum, frame):
print("TimeOut !")
raise Exception("end of time")
class TestManager(unittest.TestCase): class TestManager(unittest.TestCase):
def setUp(self): def setUp(self):
self.openssl = '/usr/bin/openssl' self.openssl = '/usr/bin/openssl'
@ -57,6 +64,7 @@ class TestManager(unittest.TestCase):
self.path = os.path.dirname(os.path.abspath(__file__)) self.path = os.path.dirname(os.path.abspath(__file__))
self.program = os.path.join(self.path, 'simulate_program.py') self.program = os.path.join(self.path, 'simulate_program.py')
self.badprogram = os.path.join(self.path, 'test.cfg') self.badprogram = os.path.join(self.path, 'test.cfg')
signal.signal(signal.SIGALRM, handler)
def test_load_config(self): def test_load_config(self):
config = configparser.ConfigParser() config = configparser.ConfigParser()
@ -407,77 +415,133 @@ class TestManager(unittest.TestCase):
config.flush() config.flush()
Manager.main(['--conf=' + config.name]) Manager.main(['--conf=' + config.name])
# def test_run_manager_command(self): def test_run_manager_command(self):
# # Doesn't work (we need enable --concurrency=multiprocessing on coverage command but we need coverage 4.0) # Enable timeout
# logsize = 10 signal.alarm(10)
# bufsize = 10 class MockServerHttp:
# queueIn = multiprocessing.Queue() def append(self, name, queueIn, queueOut, event):
# queueOut = multiprocessing.Queue() pass
# event = multiprocessing.Event() def terminate(self):
# threadCommand = multiprocessing.Process(target=self._runCommand, pass
# args=('test_run_manager_command', def join(self):
# self.program, pass
# self.path, config = configparser.ConfigParser()
# logsize, config.add_section('config:server')
# bufsize, config.add_section('command:test')
# queueIn, config.set('command:test', 'command', self.program)
# queueOut,
# event)) manage = Manager.Manager(False)
# threadCommand.start() manage.serverHttp = MockServerHttp()
# event.set() manage._load_config(config)
# queueIn.put("START") manage.launch_command()
# item = queueOut.get(timeout=4)
# self.assertEqual(item, "started", 'Error impossible to start program') queueIn = manage.threadCommand[0]._args[5]
# time.sleep(1) queueOut = manage.threadCommand[0]._args[6]
# event.set() event = manage.threadCommand[0]._args[7]
# queueIn.put("STATUS")
# item = queueOut.get(timeout=4) queueIn.put("START")
# self.assertEqual(item, "started", 'Error impossible to start program') event.set()
# time.sleep(1) # Enable timeout
# print("-" * 80, "shutdown" ) signal.alarm(10)
# event.set() item = queueOut.get(timeout=4)
# queueIn.put("SHUTDOWN") self.assertEqual(item, "started", 'Error impossible to start program')
# with self.assertRaises(queue.Empty): signal.alarm(0)
# item = queueOut.get(timeout=4) time.sleep(1)
# print("-" * 80, "wait thread" ) signal.alarm(10)
event.set()
queueIn.put("STATUS")
item = queueOut.get(timeout=4)
self.assertEqual(item, "started", 'Error impossible to read status')
time.sleep(1)
event.set()
queueIn.put("STDIN arg")
item = queueOut.get(timeout=4)
self.assertEqual(item, "ok", 'Error when send STDIN')
signal.alarm(0)
time.sleep(1)
signal.alarm(10)
event.set()
queueIn.put("STDOUT 4")
item = queueOut.get(timeout=4)
signal.alarm(0)
self.assertRegex(item,
'^[{](.*)("first-line": 4)(.*)[}]$',
'Error when read STDOUT (Missing first-line)')
self.assertRegex(item,
'^[{](.*)("last-line": 4)(.*)[}]$',
'Error when read STDOUT (Missing last-line)')
self.assertRegex(item,
'^[{](.*)(4 arg")(.*)[}]$',
'Error when read STDOUT (bad record)')
time.sleep(1)
signal.alarm(10)
event.set()
queueIn.put("BADCOMMAND")
item = queueOut.get(timeout=4)
self.assertEqual(item, "error : command unknown", 'Error impossible to read status')
signal.alarm(0)
time.sleep(1)
signal.alarm(10)
event.set()
queueIn.put("STOP")
item = queueOut.get(timeout=4)
self.assertEqual(item, "stopped", 'Error impossible to read status')
signal.alarm(0)
time.sleep(1)
signal.alarm(10)
event.set()
queueIn.put("SHUTDOWN")
with self.assertRaises(queue.Empty):
item = queueOut.get(timeout=4)
#threadCommand.join() #threadCommand.join()
# self.assertTrue(True) manage.receive_signal(15, 1)
# manage.wait_children_commands()
# def test_run_manager_command_2(self): #Disable timeout
# manage = Manager.Manager(True) signal.alarm(0)
# logsize = 10 self.assertTrue(True)
# bufsize = 10
# queueIn = multiprocessing.Queue() def test_run_manager_command_autostart(self):
# queueOut = multiprocessing.Queue() # Enable timeout
# event = multiprocessing.Event() signal.alarm(10)
# manage.runCommand('test_run_manager_command', class MockServerHttp:
# self.program, def append(self, name, queueIn, queueOut, event):
# self.path, pass
# logsize, def terminate(self):
# bufsize, pass
# queueIn, def join(self):
# queueOut, pass
# event) config = configparser.ConfigParser()
# event.set() config.add_section('config:server')
# queueIn.put("START") config.add_section('command:test')
# item = queueOut.get(timeout=4) config.set('command:test', 'command', self.program)
# self.assertEqual(item, "started", 'Error impossible to start program')
# time.sleep(1) manage = Manager.Manager(True)
# event.set() manage.serverHttp = MockServerHttp()
# queueIn.put("STATUS") manage._load_config(config)
# item = queueOut.get(timeout=4) manage.launch_command()
# self.assertEqual(item, "started", 'Error impossible to start program')
# time.sleep(1) queueIn = manage.threadCommand[0]._args[5]
# print("-" * 80, "shutdown" ) queueOut = manage.threadCommand[0]._args[6]
# event.set() event = manage.threadCommand[0]._args[7]
# queueIn.put("SHUTDOWN")
# with self.assertRaises(queue.Empty): signal.alarm(10)
# item = queueOut.get(timeout=4) event.set()
# print("-" * 80, "wait thread" ) queueIn.put("STATUS")
# #threadCommand.join() item = queueOut.get(timeout=4)
# manage.receive_signal(15, 1) self.assertEqual(item, "started", 'Error impossible to read status')
# manage.wait_children_commands() time.sleep(1)
# self.assertTrue(True) signal.alarm(10)
event.set()
queueIn.put("SHUTDOWN")
with self.assertRaises(queue.Empty):
item = queueOut.get(timeout=4)
#threadCommand.join()
manage.receive_signal(15, 1)
manage.wait_children_commands()
#Disable timeout
signal.alarm(0)
self.assertTrue(True)
if __name__ == '__main__': if __name__ == '__main__':
unittest.main() unittest.main()