#!/usr/bin/python3 # -*- coding: utf-8 -*- # # create certificate (use for test) # Copyright (C) 2017 AleaJactaEst # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU Affero General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU Affero General Public License for more details. # # You should have received a copy of the GNU Affero General Public License # along with this program. If not, see . import unittest import tempfile import os import configparser import multiprocessing import time import re import queue import signal import http.server from unittest.mock import patch from unittest.mock import MagicMock try: import pymanager.manager as Manager except ImportError: import sys sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) import pymanager.manager as Manager #try: # import pymanager.certificate as cert #except ImportError: # import sys # import os # sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) # import pymanager.certificate as cert def handler(signum, frame): print("TimeOut !") raise Exception("end of time") class TestManager(unittest.TestCase): def setUp(self): self.openssl = '/usr/bin/openssl' self.size_root = 4096 self.size_appli = 4096 self.size_child = 2048 self.passroot = 'BadPasswordRoot' self.passappli = 'BadPasswordApplication' self.country_name = 'FR' self.state_or_province_name = 'France' self.locality_name = 'Paris' self.organization_name = 'khanat' self.common_name = 'khanat' self.path = os.path.dirname(os.path.abspath(__file__)) self.program = os.path.join(self.path, 'simulate_program.py') self.badprogram = os.path.join(self.path, 'test.cfg') signal.signal(signal.SIGALRM, handler) def test_load_config(self): config = configparser.ConfigParser() config.add_section('config:server') config.set('config:server', 'port', '8000') config.set('config:server', 'keyfile', '/home/gameserver/ca/appli/private/serverkey.pem') config.set('config:server', 'certfile', '/home/gameserver/ca/appli/certs/servercert.pem') config.set('config:server', 'ca_cert', '/home/gameserver/ca/appli/certs/cachaincert.pem') config.set('config:server', 'address', '') config.set('config:server', 'authentification', 'yes') config.add_section('config:client') config.set('config:client', 'port', '8000') config.set('config:client', 'keyfile', '/home/gameserver/ca/appli/private/clientkey.pem') config.set('config:client', 'certfile', '/home/gameserver/ca/appli/certs/clientcert.pem') config.set('config:client', 'ca_cert', '/home/gameserver/ca/appli/certs/cachaincert.pem') config.set('config:client', 'address', '127.0.0.1') config.add_section('command:test') config.set('command:test', 'path', '/home/gameserver') config.set('command:test', 'command', '/bin/sleep 10') config.set('command:test', 'logsize', '10') config.set('command:test', 'bufsize', '10') config.add_section('config:user') config.set('config:user', 'usename', 'filter_all, filter_admin') try: manager = Manager.Manager(False) manager._load_config(config) self.assertTrue(True) except: self.fail('Error detected on load config') def test_load_config2(self): config = configparser.ConfigParser() config.add_section('config:server') config.set('config:server', 'authentification', 'no') config.add_section('command:test') config.set('command:test', 'path', '/home/gameserver') config.set('command:test', 'command', '/bin/sleep 10') try: manager = Manager.Manager(False) manager._load_config(config) self.assertTrue(True) except: self.fail('Error detected on load config') def test_load_config_bad_param_logsize(self): config = configparser.ConfigParser() config.add_section('command:test') config.set('command:test', 'command', '/bin/sleep 10') config.set('command:test', 'logsize', 'bidon') with self.assertRaises(ValueError): manager = Manager.Manager(False) manager._load_config(config) self.assertTrue(True) def test_load_config_bad_param_bufsize(self): config = configparser.ConfigParser() config.add_section('command:test') config.set('command:test', 'command', '/bin/sleep 10') config.set('command:test', 'bufsize', 'bidon') with self.assertRaises(ValueError): manager = Manager.Manager(False) manager._load_config(config) self.assertTrue(True) def test_load_config_empty(self): config = configparser.ConfigParser() config.add_section('config:server') config.add_section('config:client') config.set('config:client', 'port', '8000') config.set('config:client', 'keyfile', '/home/gameserver/ca/appli/private/clientkey.pem') config.set('config:client', 'certfile', '/home/gameserver/ca/appli/certs/clientcert.pem') config.set('config:client', 'ca_cert', '/home/gameserver/ca/appli/certs/cachaincert.pem') config.set('config:client', 'address', '127.0.0.1') config.add_section('config:user') config.add_section('command:test') config.set('command:test', 'command', '/bin/sleep 10') try: manager = Manager.Manager(False) manager._load_config(config) self.assertTrue(True) except: self.fail('Error detected on load config') def test_load_config_file(self): cfgfile = tempfile.NamedTemporaryFile(suffix="config.cfg", mode='w+t') cfgfile.write('#\n[config:server]\nauthentification = No\n') cfgfile.flush() try: manager = Manager.Manager(False) manager.load_config(cfgfile) self.assertTrue(True) except: self.fail('Error detected on load configuration') def test_load_config_file_none(self): with self.assertRaises(ValueError): manager = Manager.Manager(False) manager.load_config(None) def test_load_password_file(self): pwdfile = tempfile.NamedTemporaryFile(suffix="password.cfg", mode='w+t') config = configparser.ConfigParser() config.add_section('config:server') config.set('config:server', 'authentification', 'yes') config.set('config:server', 'passwordfile', pwdfile.name) pwdfile.write('username:$2a$12$2C97xW0KC/vFp3YyjlOgU.fWXJ3EiGT2Ihb0SWN9Mw0XI4WngiUqS\n\nusername2:badhash\n') pwdfile.flush() try: manager = Manager.Manager(False) manager._load_config(config) manager.load_password() self.assertTrue(True) except: self.fail('Error detected on load password') def test_constructor_manager_command(self): try: logsize = 10 bufsize = 10 queueIn = multiprocessing.Queue() queueOut = multiprocessing.Queue() event = multiprocessing.Event() manageCommand = Manager.ManageCommand('test_constructor_manager_command', self.program, self.path, logsize, bufsize, queueIn, queueOut, event) manageCommand.list_thread() self.assertTrue(True) except: self.fail('Error initialize object ManageCommand') def test_execute_manager_command(self): try: logsize = 10 bufsize = 10 queueIn = multiprocessing.Queue() queueOut = multiprocessing.Queue() event = multiprocessing.Event() manageCommand = Manager.ManageCommand('test_execute_manager_command', self.program, self.path, logsize, bufsize, queueIn, queueOut, event) manageCommand.status() manageCommand.start() manageCommand.status() manageCommand.start() foundEnd = re.compile('.*(Started).*') foundY = re.compile('.*(sendToStdinY).*') loop = 10 while loop > 0: time.sleep(1) out = manageCommand.getlog(0) if foundEnd.match(out): break loop -= 1 if not foundEnd.match(out): manageCommand.stop() self.assertTrue(False, 'Missing message in log') manageCommand.list_thread() retA = manageCommand.action("sendToStdinA") self.assertEqual(retA, "ok", 'Error impossible to send to stdin') for i in range(0, 120): retX = manageCommand.action("sendToStdin%d" % i) self.assertEqual(retX, "ok", 'Error impossible to send to stdin') retY = manageCommand.action("sendToStdinY") self.assertEqual(retY, "ok", 'Error impossible to send to stdin') loop = 10 while loop > 0: time.sleep(1) out = manageCommand.getlog(0) if foundY.match(out): break loop -= 1 if not foundY.match(out): manageCommand.stop() self.assertTrue(False, 'Missing message in log') manageCommand.stop() manageCommand.status() retZ = manageCommand.action("sendToStdinZ") manageCommand.stop() self.assertEqual(retZ, "ko", 'Error send to stdin when process is down') self.assertTrue(True) except: self.fail('Error initialize object ManageCommand') def test_execute_crash_manager_command(self): try: logsize = 10 bufsize = 10 queueIn = multiprocessing.Queue() queueOut = multiprocessing.Queue() event = multiprocessing.Event() manageCommand = Manager.ManageCommand('test_execute_crash_manager_command', self.program + ' --no-loop --timeout 1', self.path, logsize, bufsize, queueIn, queueOut, event) manageCommand.start() time.sleep(3) manageCommand.status() manageCommand.list_thread() manageCommand.stop() manageCommand.status() self.assertTrue(True) except: self.fail('Error initialize object ManageCommand') def test_execute_not_kill_manager_command(self): try: logsize = 10 bufsize = 10 queueIn = multiprocessing.Queue() queueOut = multiprocessing.Queue() event = multiprocessing.Event() manageCommand = Manager.ManageCommand('test_execute_not_kill_manager_command', self.program + " --disable-kill", self.path, logsize, bufsize, queueIn, queueOut, event, maxWaitEnd = 2) manageCommand.start() time.sleep(1) manageCommand.status() manageCommand.list_thread() manageCommand.stop() manageCommand.status() self.assertTrue(True) except: self.fail('Error initialize object ManageCommand') def test_execute_command_crashed(self): try: logsize = 10 bufsize = 10 queueIn = multiprocessing.Queue() queueOut = multiprocessing.Queue() event = multiprocessing.Event() manageCommand = Manager.ManageCommand('test_execute_command_crashed', self.program + " --no-loop --timeout=1", self.path, logsize, bufsize, queueIn, queueOut, event, waitDelay = 10) manageCommand.start() time.sleep(5) manageCommand.start() time.sleep(5) manageCommand.stop() self.assertTrue(True) except: self.fail('Error initialize object ManageCommand') def test_execute_command_file_not_found(self): try: logsize = 10 bufsize = 10 queueIn = multiprocessing.Queue() queueOut = multiprocessing.Queue() event = multiprocessing.Event() manageCommand = Manager.ManageCommand('test_execute_command_file_not_found', self.program + "_not_exist", self.path, logsize, bufsize, queueIn, queueOut, event) ret = manageCommand.start() manageCommand.stop() self.assertEqual(ret, "crashed", 'Error object not generate error when program not exist') except: self.fail('Error initialize object ManageCommand') def test_execute_command_permission(self): try: logsize = 10 bufsize = 10 queueIn = multiprocessing.Queue() queueOut = multiprocessing.Queue() event = multiprocessing.Event() manageCommand = Manager.ManageCommand('test_execute_command_permission', self.badprogram, self.path, logsize, bufsize, queueIn, queueOut, event) ret = manageCommand.start() manageCommand.stop() self.assertEqual(ret, "crashed", 'Error object not generate error when bad permission') except: self.fail('Error initialize object ManageCommand') def _runCommand(self, name, command, path, logsize, bufsize, queueIn, queueOut, event): """ Thread to manage khaganat program """ manageCommand = Manager.ManageCommand(name=name, command=command, path=path, logsize=logsize, bufsize=bufsize, queueIn=queueIn, queueOut=queueOut, event=event) manageCommand.run() def test_root_bad_loglevel(self): with self.assertRaises(ValueError): Manager.root(None, None, 'NOTEXIST', False, False) def test_root_bad_configfile(self): logfile = tempfile.NamedTemporaryFile(suffix="password.cfg", mode='w+t') with self.assertRaises(ValueError): Manager.root(None, logfile, 'DEBUG', True, True) def test_main(self): config = tempfile.NamedTemporaryFile(suffix="password.cfg", mode='w+t') config.write('[config:server]\nauthentification=no\n') config.flush() Manager.main(['--conf=' + config.name]) def test_run_manager_command(self): # Enable timeout signal.alarm(10) class MockServerHttp: def append(self, name, queueIn, queueOut, event): pass def terminate(self): pass def join(self): pass config = configparser.ConfigParser() config.add_section('config:server') config.add_section('command:test') config.set('command:test', 'command', self.program) manage = Manager.Manager(False) manage.serverHttp = MockServerHttp() manage._load_config(config) manage.launch_command() queueIn = manage.threadCommand[0]._args[5] queueOut = manage.threadCommand[0]._args[6] event = manage.threadCommand[0]._args[7] queueIn.put("START") event.set() # Enable timeout signal.alarm(10) item = queueOut.get(timeout=4) self.assertEqual(item, "started", 'Error impossible to start program') signal.alarm(0) time.sleep(1) 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() manage.receive_signal(15, 1) manage.wait_children_commands() #Disable timeout signal.alarm(0) self.assertTrue(True) def test_run_manager_command_autostart(self): # Enable timeout signal.alarm(10) class MockServerHttp: def append(self, name, queueIn, queueOut, event): pass def terminate(self): pass def join(self): pass config = configparser.ConfigParser() config.add_section('config:server') config.add_section('command:test') config.set('command:test', 'command', self.program) manage = Manager.Manager(True) manage.serverHttp = MockServerHttp() manage._load_config(config) manage.launch_command() queueIn = manage.threadCommand[0]._args[5] queueOut = manage.threadCommand[0]._args[6] event = manage.threadCommand[0]._args[7] 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) 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) class MockStreamRequestHandler(): def __init__(self): self.value = "{}" self.message = None def define_return(self, value): self.value = value def read(self, numbercar): return bytes(self.value, "utf-8") def write(self, message): self.message = message class MockServer(): def __init__(self): self.listEvent = {} self.listQueueIn = {} self.listQueueOut = {} @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_log_message(self, init): # Enable timeout signal.alarm(10) manage = Manager.ManageHttpRequest(None, None, None) manage.address_string = MagicMock() manage.log_message('example') @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_set_headers(self, init): # Enable timeout signal.alarm(10) manage = Manager.ManageHttpRequest(None, None, None) manage.log_message = MagicMock() manage.send_response = MagicMock() manage.send_header = MagicMock() manage.send_error = MagicMock() manage.end_headers = MagicMock() manage._set_headers() manage.send_response.assert_called_with(200) manage.send_header.assert_called_with('Content-type', 'application/json') @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_command_log_1(self, init): # Enable timeout signal.alarm(10) manage = Manager.ManageHttpRequest(None, None, None) manage.log_message = MagicMock() manage.send_response = MagicMock() manage.send_header = MagicMock() manage.send_error = MagicMock() manage.end_headers = MagicMock() manage.headers = {} manage._command_log() manage.send_error.assert_called_with(400, "bad content-type") @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_command_log_2(self, init): # Enable timeout signal.alarm(10) manage = Manager.ManageHttpRequest(None, None, None) manage.log_message = MagicMock() manage.send_response = MagicMock() manage.send_header = MagicMock() manage.send_error = MagicMock() manage.end_headers = MagicMock() manage.headers = {'content-type' : 'application/json'} manage._command_log() manage.send_error.assert_called_with(400, "bad content-length") @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_command_log_3(self, init): # Enable timeout signal.alarm(10) manage = Manager.ManageHttpRequest(None, None, None) manage.log_message = MagicMock() manage.send_response = MagicMock() manage.send_header = MagicMock() manage.send_error = MagicMock() manage.end_headers = MagicMock() manage.headers = {'content-length' : '1000', 'content-type' : 'application/json'} manage.rfile = TestManager.MockStreamRequestHandler() manage._command_log() manage.send_error.assert_called_with(400, 'Missing param name') @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_command_log_4(self, init): # Enable timeout signal.alarm(10) manage = Manager.ManageHttpRequest(None, None, None) manage.log_message = MagicMock() manage.send_response = MagicMock() manage.send_header = MagicMock() manage.send_error = MagicMock() manage.end_headers = MagicMock() manage.rfile = TestManager.MockStreamRequestHandler() manage.server = TestManager.MockServer() manage.rfile.define_return( '{"name": "test"}' ) manage.headers = {'content-length' : '1000', 'content-type' : 'application/json'} manage._command_log() manage.send_error.assert_called_with(400, 'Name unknown') @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_command_log_5(self, init): # Enable timeout signal.alarm(10) manage = Manager.ManageHttpRequest(None, None, None) manage.log_message = MagicMock() manage.send_response = MagicMock() manage.send_header = MagicMock() manage.send_error = MagicMock() manage.end_headers = MagicMock() manage.rfile = TestManager.MockStreamRequestHandler() manage.server = TestManager.MockServer() manage.server.listQueueIn = {'test': ''} manage.rfile.define_return( '{"name": "test"}' ) manage.headers = {'content-length' : '1000', 'content-type' : 'application/json'} manage._command_log() manage.send_error.assert_called_with(400, 'Missing param first-line') @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_command_log_6(self, init): # Enable timeout signal.alarm(10) manage = Manager.ManageHttpRequest(None, None, None) manage.log_message = MagicMock() manage.send_response = MagicMock() manage.send_header = MagicMock() manage.send_error = MagicMock() manage.end_headers = MagicMock() manage.rfile = TestManager.MockStreamRequestHandler() manage.server = TestManager.MockServer() manage.server.listQueueIn = {'test': ''} manage.rfile.define_return( '{"name": "test", "first-line" : "test"}' ) manage.headers = {'content-length' : '1000', 'content-type' : 'application/json'} manage._command_log() manage.send_error.assert_called_with(400, 'Impossible to read first-line') @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_command_log_7(self, init): # Enable timeout signal.alarm(10) manage = Manager.ManageHttpRequest(None, None, None) manage.log_message = MagicMock() manage.send_response = MagicMock() manage.send_header = MagicMock() manage.send_error = MagicMock() manage.end_headers = MagicMock() manage.rfile = TestManager.MockStreamRequestHandler() manage.wfile = TestManager.MockStreamRequestHandler() manage.server = TestManager.MockServer() manage.server.listEvent = {'test': multiprocessing.Event() } manage.server.listQueueIn = {'test': multiprocessing.Queue() } manage.server.listQueueOut = {'test': multiprocessing.Queue()} #manage.server.listQueueOut['test'].put("empty") manage.rfile.define_return( '{"name": "test", "first-line" : "1"}' ) manage.headers = {'content-length' : '1000', 'content-type' : 'application/json'} manage._command_log() #self.assertEqual(b'empty',manage.wfile.message) @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_command_log_8(self, init): # Enable timeout signal.alarm(10) manage = Manager.ManageHttpRequest(None, None, None) manage.log_message = MagicMock() manage.send_response = MagicMock() manage.send_header = MagicMock() manage.send_error = MagicMock() manage.end_headers = MagicMock() manage.rfile = TestManager.MockStreamRequestHandler() manage.wfile = TestManager.MockStreamRequestHandler() manage.server = TestManager.MockServer() manage.server.listEvent = {'test': multiprocessing.Event() } manage.server.listQueueIn = {'test': multiprocessing.Queue() } manage.server.listQueueOut = {'test': multiprocessing.Queue()} manage.server.listQueueOut['test'].put("empty") manage.rfile.define_return( '{"name": "test", "first-line" : "1"}' ) manage.headers = {'content-length' : '1000', 'content-type' : 'application/json'} manage._command_log() self.assertEqual(b'empty',manage.wfile.message) @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_send_list(self, init): # Enable timeout signal.alarm(10) manage = Manager.ManageHttpRequest(None, None, None) manage.log_message = MagicMock() manage.send_response = MagicMock() manage.send_header = MagicMock() manage.send_error = MagicMock() manage.end_headers = MagicMock() manage.rfile = TestManager.MockStreamRequestHandler() manage.wfile = TestManager.MockStreamRequestHandler() manage.server = TestManager.MockServer() manage.server.listEvent = {'test': multiprocessing.Event() } manage.server.listQueueIn = {'test': multiprocessing.Queue() } manage.server.listQueueOut = {'test': multiprocessing.Queue()} manage._send_list() self.assertEqual(b'{"0": "test"}',manage.wfile.message) @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_send_shutdown(self, init): # Enable timeout signal.alarm(10) manage = Manager.ManageHttpRequest(None, None, None) manage.log_message = MagicMock() manage.send_response = MagicMock() manage.send_header = MagicMock() manage.send_error = MagicMock() manage.end_headers = MagicMock() manage.rfile = TestManager.MockStreamRequestHandler() manage.wfile = TestManager.MockStreamRequestHandler() manage.server = TestManager.MockServer() manage.server.listEvent = {'test': multiprocessing.Event() } manage.server.listQueueIn = {'test': multiprocessing.Queue() } manage.server.listQueueOut = {'test': multiprocessing.Queue()} manage._send_shutdown() self.assertEqual(b'{"shutdown": "ok"}',manage.wfile.message) @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_send_command_all(self, init): # Enable timeout signal.alarm(10) manage = Manager.ManageHttpRequest(None, None, None) manage.log_message = MagicMock() manage.send_response = MagicMock() manage.send_header = MagicMock() manage.send_error = MagicMock() manage.end_headers = MagicMock() manage.rfile = TestManager.MockStreamRequestHandler() manage.wfile = TestManager.MockStreamRequestHandler() manage.server = TestManager.MockServer() manage.server.listEvent = {'test': multiprocessing.Event() } manage.server.listQueueIn = {'test': multiprocessing.Queue() } manage.server.listQueueOut = {'test': multiprocessing.Queue()} manage.server.listQueueOut['test'].put("empty") manage._send_command_all("test") self.assertEqual(b'{"test": "empty"}',manage.wfile.message) @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_send_command_all_timeout(self, init): # Enable timeout signal.alarm(10) manage = Manager.ManageHttpRequest(None, None, None) manage.log_message = MagicMock() manage.send_response = MagicMock() manage.send_header = MagicMock() manage.send_error = MagicMock() manage.end_headers = MagicMock() manage.rfile = TestManager.MockStreamRequestHandler() manage.wfile = TestManager.MockStreamRequestHandler() manage.server = TestManager.MockServer() manage.server.listEvent = {'test': multiprocessing.Event() } manage.server.listQueueIn = {'test': multiprocessing.Queue() } manage.server.listQueueOut = {'test': multiprocessing.Queue()} manage._send_command_all("test") @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_send_action_1(self, init): # Enable timeout signal.alarm(10) manage = Manager.ManageHttpRequest(None, None, None) manage.log_message = MagicMock() manage.send_response = MagicMock() manage.send_header = MagicMock() manage.send_error = MagicMock() manage.end_headers = MagicMock() manage.rfile = TestManager.MockStreamRequestHandler() manage.wfile = TestManager.MockStreamRequestHandler() manage.server = TestManager.MockServer() manage.server.listEvent = {'test': multiprocessing.Event() } manage.server.listQueueIn = {'test': multiprocessing.Queue() } manage.server.listQueueOut = {'test': multiprocessing.Queue()} manage.headers = {'content-length' : '1000', 'content-type' : 'application/json'} manage._send_action() manage.send_error.assert_called_with(400, 'Missing param name') @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_send_action_2(self, init): # Enable timeout signal.alarm(10) manage = Manager.ManageHttpRequest(None, None, None) manage.log_message = MagicMock() manage.send_response = MagicMock() manage.send_header = MagicMock() manage.send_error = MagicMock() manage.end_headers = MagicMock() manage.rfile = TestManager.MockStreamRequestHandler() manage.wfile = TestManager.MockStreamRequestHandler() manage.server = TestManager.MockServer() manage.server.listEvent = {'test': multiprocessing.Event() } manage.server.listQueueIn = {'test': multiprocessing.Queue() } manage.server.listQueueOut = {'test': multiprocessing.Queue()} manage.rfile.define_return( '{"name": "testnew"}' ) manage.headers = {'content-length' : '1000', 'content-type' : 'application/json'} manage._send_action() manage.send_error.assert_called_with(400, 'Name unknown') @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_send_action_3(self, init): # Enable timeout signal.alarm(10) manage = Manager.ManageHttpRequest(None, None, None) manage.log_message = MagicMock() manage.send_response = MagicMock() manage.send_header = MagicMock() manage.send_error = MagicMock() manage.end_headers = MagicMock() manage.rfile = TestManager.MockStreamRequestHandler() manage.wfile = TestManager.MockStreamRequestHandler() manage.server = TestManager.MockServer() manage.server.listEvent = {'test': multiprocessing.Event() } manage.server.listQueueIn = {'test': multiprocessing.Queue() } manage.server.listQueueOut = {'test': multiprocessing.Queue()} manage.rfile.define_return( '{"name": "test"}' ) manage.headers = {'content-length' : '1000', 'content-type' : 'application/json'} manage._send_action() manage.send_error.assert_called_with(400, 'Missing param action') @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_send_action_4(self, init): # Enable timeout signal.alarm(10) manage = Manager.ManageHttpRequest(None, None, None) manage.log_message = MagicMock() manage.send_response = MagicMock() manage.send_header = MagicMock() manage.send_error = MagicMock() manage.end_headers = MagicMock() manage.rfile = TestManager.MockStreamRequestHandler() manage.wfile = TestManager.MockStreamRequestHandler() manage.server = TestManager.MockServer() manage.server.listEvent = {'test': multiprocessing.Event() } manage.server.listQueueIn = {'test': multiprocessing.Queue() } manage.server.listQueueOut = {'test': multiprocessing.Queue()} manage.rfile.define_return( '{"name": "test", "action": "example"}' ) manage.headers = {'content-length' : '1000', 'content-type' : 'application/json'} manage._send_action() self.assertEqual(None,manage.wfile.message) @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_send_action_5(self, init): # Enable timeout signal.alarm(10) manage = Manager.ManageHttpRequest(None, None, None) manage.log_message = MagicMock() manage.send_response = MagicMock() manage.send_header = MagicMock() manage.send_error = MagicMock() manage.end_headers = MagicMock() manage.rfile = TestManager.MockStreamRequestHandler() manage.wfile = TestManager.MockStreamRequestHandler() manage.server = TestManager.MockServer() manage.server.listEvent = {'test': multiprocessing.Event() } manage.server.listQueueIn = {'test': multiprocessing.Queue() } manage.server.listQueueOut = {'test': multiprocessing.Queue()} manage.server.listQueueOut['test'].put("empty") manage.rfile.define_return( '{"name": "test", "action": "example"}' ) manage.headers = {'content-length' : '1000', 'content-type' : 'application/json'} manage._send_action() self.assertEqual(b'{"state": "empty"}',manage.wfile.message) @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_send_action_6(self, init): # Enable timeout signal.alarm(10) manage = Manager.ManageHttpRequest(None, None, None) manage.log_message = MagicMock() manage.send_response = MagicMock() manage.send_header = MagicMock() manage.send_error = MagicMock() manage.end_headers = MagicMock() manage.headers = {} manage._send_action() @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_send_command_1(self, init): # Enable timeout signal.alarm(10) manage = Manager.ManageHttpRequest(None, None, None) manage.log_message = MagicMock() manage.send_response = MagicMock() manage.send_header = MagicMock() manage.send_error = MagicMock() manage.end_headers = MagicMock() manage.headers = {} manage._send_command("STATUS") @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_send_command_2(self, init): # Enable timeout signal.alarm(10) manage = Manager.ManageHttpRequest(None, None, None) manage.log_message = MagicMock() manage.send_response = MagicMock() manage.send_header = MagicMock() manage.send_error = MagicMock() manage.end_headers = MagicMock() manage.rfile = TestManager.MockStreamRequestHandler() manage.wfile = TestManager.MockStreamRequestHandler() manage.server = TestManager.MockServer() manage.server.listEvent = {'test': multiprocessing.Event() } manage.server.listQueueIn = {'test': multiprocessing.Queue() } manage.server.listQueueOut = {'test': multiprocessing.Queue()} manage.headers = {'content-length' : '1000', 'content-type' : 'application/json'} manage._send_command("STATUS") manage.send_error.assert_called_with(400, 'Missing param name') @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_send_command_3(self, init): # Enable timeout signal.alarm(10) manage = Manager.ManageHttpRequest(None, None, None) manage.log_message = MagicMock() manage.send_response = MagicMock() manage.send_header = MagicMock() manage.send_error = MagicMock() manage.end_headers = MagicMock() manage.rfile = TestManager.MockStreamRequestHandler() manage.wfile = TestManager.MockStreamRequestHandler() manage.server = TestManager.MockServer() manage.server.listEvent = {'test': multiprocessing.Event() } manage.server.listQueueIn = {'test': multiprocessing.Queue() } manage.server.listQueueOut = {'test': multiprocessing.Queue()} manage.headers = {'content-length' : '1000', 'content-type' : 'application/json'} manage.server.listQueueOut['test'].put("empty") manage.rfile.define_return( '{"name": "test"}' ) manage._send_command("STATUS") self.assertEqual(b'{"state": "empty"}',manage.wfile.message) @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_send_command_4(self, init): # Enable timeout signal.alarm(10) manage = Manager.ManageHttpRequest(None, None, None) manage.log_message = MagicMock() manage.send_response = MagicMock() manage.send_header = MagicMock() manage.send_error = MagicMock() manage.end_headers = MagicMock() manage.rfile = TestManager.MockStreamRequestHandler() manage.wfile = TestManager.MockStreamRequestHandler() manage.server = TestManager.MockServer() manage.server.listEvent = {'test': multiprocessing.Event() } manage.server.listQueueIn = {'test': multiprocessing.Queue() } manage.server.listQueueOut = {'test': multiprocessing.Queue()} manage.headers = {'content-length' : '1000', 'content-type' : 'application/json'} manage.rfile.define_return( '{"name": "test"}' ) manage._send_command("STATUS") manage.send_error.assert_called_with(500, 'Missing return') @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_send_command_5(self, init): # Enable timeout signal.alarm(10) manage = Manager.ManageHttpRequest(None, None, None) manage.log_message = MagicMock() manage.send_response = MagicMock() manage.send_header = MagicMock() manage.send_error = MagicMock() manage.end_headers = MagicMock() manage.rfile = TestManager.MockStreamRequestHandler() manage.wfile = TestManager.MockStreamRequestHandler() manage.server = TestManager.MockServer() manage.server.listEvent = {'test': multiprocessing.Event() } manage.server.listQueueIn = {'test': multiprocessing.Queue() } manage.server.listQueueOut = {'test': multiprocessing.Queue()} manage.headers = {'content-length' : '1000', 'content-type' : 'application/json'} manage.server.listQueueOut['test'].put("empty") manage.rfile.define_return( '{"name": "testnew"}' ) manage._send_command("STATUS") manage.send_error.assert_called_with(400, 'Name unknown') if __name__ == '__main__': unittest.main()