#!/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 import logging import json 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: signal.alarm(30) 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) res = manageCommand.start() self.assertEqual(res, 'started') wait = True while wait: time.sleep(1) res = manageCommand.status() if res == 'stopped': wait = False 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: signal.alarm(30) 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) res = manageCommand.start() self.assertEqual(res, 'started') wait = True while wait: time.sleep(1) res = manageCommand.status() self.assertEqual(res, 'started') res = manageCommand.getlog(0) try: resjson = json.loads(res) if 'last-line' in res: if resjson['last-line'] == 3: wait = False except: pass manageCommand.list_thread() manageCommand.stop() res = manageCommand.status() self.assertEqual(res, 'stopped') 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() wait = True while wait: time.sleep(1) res = manageCommand.status() if res == 'stopped': wait = False res = manageCommand.status() self.assertEqual(res, 'stopped') manageCommand.start() wait = True while wait: time.sleep(1) res = manageCommand.status() if res == 'stopped': wait = False res = manageCommand.status() self.assertEqual(res, 'stopped') 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 = {} self.authentification = False @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') @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_do_GET_1(self, init): # Enable timeout signal.alarm(10) manage = Manager.ManageHttpRequest(None, None, None) manage.server = TestManager.MockServer() manage._command_log = MagicMock() manage.path = "/STDOUT" manage.do_GET() manage._command_log.assert_called_with() @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_check_authentication_1(self, init): # Enable timeout signal.alarm(10) manage = Manager.ManageHttpRequest(None, None, None) manage.server = TestManager.MockServer() res = manage.check_authentication() self.assertEqual(True, res) @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_check_authentication_2(self, init): # Enable timeout signal.alarm(10) manage = Manager.ManageHttpRequest(None, None, None) manage.server = TestManager.MockServer() manage.server.authentification = True res = manage.check_authentication() self.assertEqual(False, res) @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_check_authentication_3(self, init): # Enable timeout signal.alarm(10) manage = Manager.ManageHttpRequest(None, None, None) manage.server = TestManager.MockServer() logging.error = MagicMock() manage.server.authentification = True manage.headers = {'Authorization' : 'Other'} res = manage.check_authentication() self.assertEqual(False, res) logging.error.assert_called_with("Authentification with Bad method (Other)") @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_check_authentication_4(self, init): # Enable timeout signal.alarm(10) manage = Manager.ManageHttpRequest(None, None, None) manage.server = TestManager.MockServer() logging.error = MagicMock() manage.server.authentification = True manage.headers = {'Authorization' : 'Basic '} res = manage.check_authentication() self.assertEqual(False, res) logging.error.assert_called_with("Error detected list index out of range") @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_check_authentication_5(self, init): # Enable timeout signal.alarm(10) manage = Manager.ManageHttpRequest(None, None, None) manage.server = TestManager.MockServer() logging.error = MagicMock() manage.server.authentification = True manage.headers = {'Authorization' : 'Basic test:$2b$12$AoRGhrAYsuSlROUDx.0AZ.3SPBVYd8wD.Vyz7jCRXZdwm7ArEN3Oi'} res = manage.check_authentication() self.assertEqual(False, res) logging.error.assert_called_with("Error detected Incorrect padding") @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_check_authentication_6(self, init): # Enable timeout signal.alarm(10) manage = Manager.ManageHttpRequest(None, None, None) manage.server = TestManager.MockServer() manage.server.users = {"user1": ""} logging.error = MagicMock() manage.server.authentification = True manage.headers = {'Authorization' : 'Basic dGVzdDokMmIkMTIkQW9SR2hyQVlzdVNsUk9VRHguMEFaLjNTUEJWWWQ4d0QuVnl6N2pDUlhaZHdtN0FyRU4zT2k='} res = manage.check_authentication() self.assertEqual(False, res) logging.error.assert_called_with("Authentification with unknown user (test)") @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_do_GET_2(self, init): # Enable timeout signal.alarm(10) manage = Manager.ManageHttpRequest(None, None, None) manage.server = TestManager.MockServer() manage._send_command = MagicMock() manage.path = "/STATUS" manage.do_GET() manage._send_command.assert_called_with("STATUS") @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_do_GET_3(self, init): # Enable timeout signal.alarm(10) manage = Manager.ManageHttpRequest(None, None, None) manage.server = TestManager.MockServer() manage._send_list = MagicMock() manage.path = "/LIST" manage.do_GET() manage._send_list.assert_called_with() @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_do_GET_4(self, init): # Enable timeout signal.alarm(10) manage = Manager.ManageHttpRequest(None, None, None) manage.server = TestManager.MockServer() manage._send_command_all = MagicMock() manage.path = "/STATUSALL" manage.do_GET() manage._send_command_all.assert_called_with("STATUS") @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_do_GET_5(self, init): # Enable timeout signal.alarm(10) manage = Manager.ManageHttpRequest(None, None, None) manage.server = TestManager.MockServer() manage.send_error = MagicMock() manage.path = "/BADPATH" manage.do_GET() manage.send_error.assert_called_with(400, 'Path unknown') @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_do_GET_6(self, init): # Enable timeout signal.alarm(10) manage = Manager.ManageHttpRequest(None, None, None) manage.server = TestManager.MockServer() manage.server.authentification = True manage.send_error = MagicMock() manage.path = "/BADPATH" manage.do_GET() manage.send_error.assert_called_with(403, 'Wrong authentication') @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_do_POST_1(self, init): # Enable timeout signal.alarm(10) manage = Manager.ManageHttpRequest(None, None, None) manage.server = TestManager.MockServer() manage._send_command = MagicMock() manage.path = "/START" manage.do_POST() manage._send_command.assert_called_with("START") @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_do_POST_2(self, init): # Enable timeout signal.alarm(10) manage = Manager.ManageHttpRequest(None, None, None) manage.server = TestManager.MockServer() manage._send_command = MagicMock() manage.path = "/STOP" manage.do_POST() manage._send_command.assert_called_with("STOP") @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_do_POST_3(self, init): # Enable timeout signal.alarm(10) manage = Manager.ManageHttpRequest(None, None, None) manage.server = TestManager.MockServer() manage._send_action = MagicMock() manage.path = "/STDIN" manage.do_POST() manage._send_action.assert_called_with() @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_do_POST_4(self, init): # Enable timeout signal.alarm(10) manage = Manager.ManageHttpRequest(None, None, None) manage.server = TestManager.MockServer() manage._send_shutdown = MagicMock() manage.path = "/SHUTDOWN" manage.do_POST() manage._send_shutdown.assert_called_with() @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_do_POST_5(self, init): # Enable timeout signal.alarm(10) manage = Manager.ManageHttpRequest(None, None, None) manage.server = TestManager.MockServer() manage._send_command_all = MagicMock() manage.path = "/STARTALL" manage.do_POST() manage._send_command_all.assert_called_with("START") @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_do_POST_6(self, init): # Enable timeout signal.alarm(10) manage = Manager.ManageHttpRequest(None, None, None) manage.server = TestManager.MockServer() manage._send_command_all = MagicMock() manage.path = "/STOPALL" manage.do_POST() manage._send_command_all.assert_called_with("STOP") @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_do_POST_7(self, init): # Enable timeout signal.alarm(10) manage = Manager.ManageHttpRequest(None, None, None) manage.server = TestManager.MockServer() manage.log_message = MagicMock() manage.send_response = MagicMock() manage.send_header = MagicMock() manage.send_error = MagicMock() manage.end_headers = MagicMock() manage._send_command = MagicMock() manage._send_action = MagicMock() manage._send_shutdown = MagicMock() manage._send_command_all = MagicMock() manage.path = "/BADPATH" manage.do_POST() manage.send_error.assert_called_with(400, 'Path unknown') @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_do_POST_8(self, init): # Enable timeout signal.alarm(10) manage = Manager.ManageHttpRequest(None, None, None) manage.server = TestManager.MockServer() manage.server.authentification = True manage.send_error = MagicMock() manage.path = "/BADPATH" manage.do_POST() manage.send_error.assert_called_with(403, 'Wrong authentication') @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_do_HEAD(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.path = "test" manage.do_HEAD() manage.send_error.assert_called_with(404, 'File Not Found: test') @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_do_PUT(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.path = "test" manage.do_PUT() manage.send_error.assert_called_with(404, 'File Not Found: test') @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_do_PATCH(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.path = "test" manage.do_PATCH() manage.send_error.assert_called_with(404, 'File Not Found: test') @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_do_DELETE(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.path = "test" manage.do_DELETE() manage.send_error.assert_called_with(404, 'File Not Found: test') if __name__ == '__main__': unittest.main()