From 336b77db09c9a4f6179d21ec7ed4a36c3c906858 Mon Sep 17 00:00:00 2001 From: AleaJactaEst Date: Wed, 7 Mar 2018 22:41:35 +0100 Subject: [PATCH] adding some test --- pymanager/manager.py | 3 +- setup.py | 6 +- tests/test_client.py | 193 ++++++++++++++++++++++++++++++++++++++++- tests/test_manager.py | 34 ++++++++ tests/test_password.py | 9 ++ 5 files changed, 238 insertions(+), 7 deletions(-) diff --git a/pymanager/manager.py b/pymanager/manager.py index 3392f8e..1a38a2a 100755 --- a/pymanager/manager.py +++ b/pymanager/manager.py @@ -357,7 +357,7 @@ class ManageHttpRequest(http.server.SimpleHTTPRequestHandler): logging.error("Authentification with unknown user (%s)" % account) return False hashed_password = self.server.users[account] - if bcrypt.checkpw(password, hashed_password): + if bcrypt.checkpw(password.encode('utf-8'), hashed_password): return True else: logging.error("Authentification with wrong password for user (%s)" % account) @@ -365,7 +365,6 @@ class ManageHttpRequest(http.server.SimpleHTTPRequestHandler): except (ValueError, IndexError, AttributeError) as e: logging.error("Error detected %s" % e) return False - return True def do_GET(self): """ diff --git a/setup.py b/setup.py index e4e1d91..29dc308 100755 --- a/setup.py +++ b/setup.py @@ -132,9 +132,9 @@ setup( entry_points={ 'console_scripts': [ 'opennel_certificate=pymanager.certificate:main', - 'opennel_manager=pymanager.manager:main', - 'opennel_client=pymanager.client:main', - 'opennel_password=pymanager.password:main', + 'opennel_manager=pymanager.manager:main', + 'opennel_client=pymanager.client:main', + 'opennel_password=pymanager.password:main', ], }, ) diff --git a/tests/test_client.py b/tests/test_client.py index 7b4137d..19fbc96 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -38,7 +38,7 @@ except ImportError: sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) import pymanager.certificate as cert -class TestManager(unittest.TestCase): +class TestClient(unittest.TestCase): def setUp(self): pass @@ -186,7 +186,7 @@ class TestManager(unittest.TestCase): https = Client.HTTPSConnectionCertificate(None, None, 'ca') https.connect() - def test_client_send_json(self): + def test_client_load_config(self): #workdir = tempfile.mkdtemp(prefix='test_client_send_json') workdir_cert_root = tempfile.mkdtemp(prefix='pymanager-certificate-root-') workdir_cert_appli = tempfile.mkdtemp(prefix='pymanager-certificate-application-') @@ -214,6 +214,195 @@ class TestManager(unittest.TestCase): except: self.fail('Error detected on load config') + @patch.object(Client.Client, 'send_json') + def test_client_root(self, send_json): + cfgfile = tempfile.NamedTemporaryFile(suffix="config.cfg", mode='w+t') + cfgfile.write('#\n[config:client]\port = 8000\n') + cfgfile.flush() + + Client.root(command='STATUS', + program='ALL', + stdin="", + firstline="", + fileLog="", + logLevel="INFO", + username=None, + password_comand_line=False, + password=None, + show_log_console=False, + raw_data=False, + remove_color=False, + filecfg=cfgfile) + + @patch.object(Client.Client, 'send_json') + def test_client_root_2(self, send_json): + cfgfile = tempfile.NamedTemporaryFile(suffix="config.cfg", mode='w+t') + cfgfile.write('#\n[config:client]\port = 8000\n') + cfgfile.flush() + + Client.root(command='START', + program='ALL', + stdin="", + firstline="", + fileLog="", + logLevel="INFO", + username=None, + password_comand_line=False, + password=None, + show_log_console=False, + raw_data=False, + remove_color=False, + filecfg=cfgfile) + + @patch.object(Client.Client, 'send_json') + def test_client_root_3(self, send_json): + cfgfile = tempfile.NamedTemporaryFile(suffix="config.cfg", mode='w+t') + cfgfile.write('#\n[config:client]\port = 8000\n') + cfgfile.flush() + + Client.root(command='STDIN', + program='ALL', + stdin="", + firstline="", + fileLog="", + logLevel="INFO", + username=None, + password_comand_line=False, + password=None, + show_log_console=False, + raw_data=False, + remove_color=False, + filecfg=cfgfile) + + @patch.object(Client.Client, 'send_json') + def test_client_root_4(self, send_json): + cfgfile = tempfile.NamedTemporaryFile(suffix="config.cfg", mode='w+t') + cfgfile.write('#\n[config:client]\port = 8000\n') + cfgfile.flush() + + Client.root(command='STDOUT', + program='ALL', + stdin="", + firstline="", + fileLog="", + logLevel="INFO", + username=None, + password_comand_line=False, + password=None, + show_log_console=False, + raw_data=False, + remove_color=False, + filecfg=cfgfile) + + @patch.object(Client.Client, 'send_json') + def test_client_root_5(self, send_json): + cfgfile = tempfile.NamedTemporaryFile(suffix="config.cfg", mode='w+t') + cfgfile.write('#\n[config:client]\port = 8000\n') + cfgfile.flush() + + Client.root(command='LIST', + program='ALL', + stdin="", + firstline="", + fileLog="", + logLevel="INFO", + username=None, + password_comand_line=False, + password=None, + show_log_console=False, + raw_data=False, + remove_color=False, + filecfg=cfgfile) + + @patch.object(Client.Client, 'send_json') + def test_client_root_6(self, send_json): + cfgfile = tempfile.NamedTemporaryFile(suffix="config.cfg", mode='w+t') + cfgfile.write('#\n[config:client]\port = 8000\n') + cfgfile.flush() + + Client.root(command='STATUSALL', + program='ALL', + stdin="", + firstline="", + fileLog="", + logLevel="INFO", + username=None, + password_comand_line=False, + password=None, + show_log_console=False, + raw_data=False, + remove_color=False, + filecfg=cfgfile) + + @patch.object(Client.Client, 'send_json') + def test_client_root_7(self, send_json): + cfgfile = tempfile.NamedTemporaryFile(suffix="config.cfg", mode='w+t') + cfgfile.write('#\n[config:client]\port = 8000\n') + cfgfile.flush() + + Client.root(command='SHUTDOWN', + program='ALL', + stdin="", + firstline="", + fileLog="", + logLevel="INFO", + username=None, + password_comand_line=False, + password=None, + show_log_console=False, + raw_data=False, + remove_color=False, + filecfg=cfgfile) + + @patch.object(Client.Client, 'send_json') + def test_client_root_8(self, send_json): + cfgfile = tempfile.NamedTemporaryFile(suffix="config.cfg", mode='w+t') + logfile = tempfile.NamedTemporaryFile(suffix="logfile", mode='w+t') + cfgfile.write('#\n[config:client]\port = 8000\n') + cfgfile.flush() + logfile.flush() + + Client.root(command='SHUTDOWN', + program='ALL', + stdin="", + firstline="", + fileLog=logfile, + logLevel="INFO", + username=None, + password_comand_line=False, + password=None, + show_log_console=True, + raw_data=False, + remove_color=False, + filecfg=cfgfile) + + @patch.object(Client.Client, 'send_json') + def test_client_root_9(self, send_json): + cfgfile = tempfile.NamedTemporaryFile(suffix="config.cfg", mode='w+t') + cfgfile.write('#\n[config:client]\port = 8000\n') + cfgfile.flush() + + with self.assertRaises(ValueError): + Client.root(command='SHUTDOWN', + program='ALL', + stdin="", + firstline="", + fileLog="", + logLevel="BADLEVEL", + username=None, + password_comand_line=False, + password=None, + show_log_console=False, + raw_data=False, + remove_color=False, + filecfg=cfgfile) + + def test_main(self): + config = tempfile.NamedTemporaryFile(suffix="password.cfg", mode='w+t') + config.write('[config:server]\nauthentification=no\n') + config.flush() + Client.main(['--conf=' + config.name]) + if __name__ == '__main__': unittest.main() diff --git a/tests/test_manager.py b/tests/test_manager.py index 8ac1e46..98bccc6 100644 --- a/tests/test_manager.py +++ b/tests/test_manager.py @@ -1158,6 +1158,40 @@ class TestManager(unittest.TestCase): res = manage.check_authentication() self.assertEqual(False, res) logging.error.assert_called_with("Authentification with unknown user (test)") + signal.alarm(0) + + @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') + def test_run_manage_check_authentication_7(self, init): + # Enable timeout + signal.alarm(10) + manage = Manager.ManageHttpRequest(None, None, None) + manage.server = TestManager.MockServer() + #manage.server.users = {"username": bytes("$2b$12$f5kbmfXvUq3LDequGHcoqu06AueJ35TpIU2MIf5L8c9Y1PE/6Rz4i", "utf-8")} + #manage.server.users = {"username": b"$2b$12$f5kbmfXvUq3LDequGHcoqu06AueJ35TpIU2MIf5L8c9Y1PE/6Rz4i"} + manage.server.users = {"username": "$2b$12$f5kbmfXvUq3LDequGHcoqu06AueJ35TpIU2MIf5L8c9Y1PE/6Rz4i".encode('utf-8')} + logging.error = MagicMock() + manage.server.authentification = True + #manage.headers = {'Authorization': bytes('Basic dXNlcm5hbWU6cGFzc3dvcmQ=', 'UTF-8')} + manage.headers = {'Authorization': 'Basic dXNlcm5hbWU6cGFzc3dvcmQ='} + res = manage.check_authentication() + self.assertEqual(True, res) + #logging.error.assert_called_with("Authentification with user (username)") + signal.alarm(0) + + @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') + def test_run_manage_check_authentication_8(self, init): + # Enable timeout + signal.alarm(10) + manage = Manager.ManageHttpRequest(None, None, None) + manage.server = TestManager.MockServer() + manage.server.users = {"username": "$2b$12$/rfBBlTy3E9CgB8ZGeWFBOo54UN5Ogj3PuEOHVJcXyQ2hL6kYVwWW".encode('utf-8')} + logging.error = MagicMock() + manage.server.authentification = True + manage.headers = {'Authorization': 'Basic dXNlcm5hbWU6cGFzc3dvcmQ='} + res = manage.check_authentication() + self.assertEqual(False, res) + logging.error.assert_called_with("Authentification with wrong password for user (username)") + signal.alarm(0) @patch.object(http.server.SimpleHTTPRequestHandler, '__init__') def test_run_manage_do_GET_2(self, init): diff --git a/tests/test_password.py b/tests/test_password.py index cdfdad9..71323e6 100644 --- a/tests/test_password.py +++ b/tests/test_password.py @@ -121,6 +121,15 @@ class TestPassword(unittest.TestCase): password = Password.PasswordFile(pwdfile.name, False, True, 'username', 'MyPassword') password.delete() + def test_delete_2(self): + pwdfile = tempfile.NamedTemporaryFile(suffix="passwordfile.tmp", mode='w+t') + pwdfile.write('username:$2a$12$2C97xW0KC/vFp3YyjlOgU.fWXJ3EiGT2Ihb0SWN9Mw0XI4WngiUqS\n\n') + pwdfile.write('username2:$2a$12$2C97xW0KC/vFp3YyjlOgU.fWXJ3EiGT2Ihb0SWN9Mw0XI4WngiUqS\n\n') + pwdfile.flush() + + password = Password.PasswordFile(pwdfile.name, False, True, 'username3', 'MyPassword') + password.delete() + def test_run_update(self): pwdfile = tempfile.NamedTemporaryFile(suffix="passwordfile.tmp", mode='w+t') pwdfile.write('username:$2a$12$2C97xW0KC/vFp3YyjlOgU.fWXJ3EiGT2Ihb0SWN9Mw0XI4WngiUqS\n\n')