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

adding some test

Browse source
This commit is contained in:
AleaJactaEst 2018-03-07 22:41:35 +01:00
parent b8d5c1dca1
commit 336b77db09
5 changed files with 238 additions and 7 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
pymanager/manager.py
View file

@ -357,7 +357,7 @@ class ManageHttpRequest(http.server.SimpleHTTPRequestHandler):
logging.error("Authentification with unknown user (%s)" % account) logging.error("Authentification with unknown user (%s)" % account)
return False return False
hashed_password = self.server.users[account] hashed_password = self.server.users[account]
if bcrypt.checkpw(password, hashed_password): if bcrypt.checkpw(password.encode('utf-8'), hashed_password):
return True return True
else: else:
logging.error("Authentification with wrong password for user (%s)" % account) 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: except (ValueError, IndexError, AttributeError) as e:
logging.error("Error detected %s" % e) logging.error("Error detected %s" % e)
return False return False
return True
def do_GET(self): def do_GET(self):
""" """

6
setup.py
View file

@ -132,9 +132,9 @@ setup(
entry_points={ entry_points={
'console_scripts': [ 'console_scripts': [
'opennel_certificate=pymanager.certificate:main', 'opennel_certificate=pymanager.certificate:main',
'opennel_manager=pymanager.manager:main', 'opennel_manager=pymanager.manager:main',
'opennel_client=pymanager.client:main', 'opennel_client=pymanager.client:main',
'opennel_password=pymanager.password:main', 'opennel_password=pymanager.password:main',
], ],
}, },
) )

193
tests/test_client.py
View file

@ -38,7 +38,7 @@ 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
class TestManager(unittest.TestCase): class TestClient(unittest.TestCase):
def setUp(self): def setUp(self):
pass pass
@ -186,7 +186,7 @@ class TestManager(unittest.TestCase):
https = Client.HTTPSConnectionCertificate(None, None, 'ca') https = Client.HTTPSConnectionCertificate(None, None, 'ca')
https.connect() https.connect()
def test_client_send_json(self): def test_client_load_config(self):
#workdir = tempfile.mkdtemp(prefix='test_client_send_json') #workdir = tempfile.mkdtemp(prefix='test_client_send_json')
workdir_cert_root = tempfile.mkdtemp(prefix='pymanager-certificate-root-') workdir_cert_root = tempfile.mkdtemp(prefix='pymanager-certificate-root-')
workdir_cert_appli = tempfile.mkdtemp(prefix='pymanager-certificate-application-') workdir_cert_appli = tempfile.mkdtemp(prefix='pymanager-certificate-application-')
@ -214,6 +214,195 @@ class TestManager(unittest.TestCase):
except: except:
self.fail('Error detected on load config') 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__': if __name__ == '__main__':
unittest.main() unittest.main()

34
tests/test_manager.py
View file

@ -1158,6 +1158,40 @@ class TestManager(unittest.TestCase):
res = manage.check_authentication() res = manage.check_authentication()
self.assertEqual(False, res) self.assertEqual(False, res)
logging.error.assert_called_with("Authentification with unknown user (test)") 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__') @patch.object(http.server.SimpleHTTPRequestHandler, '__init__')
def test_run_manage_do_GET_2(self, init): def test_run_manage_do_GET_2(self, init):

9
tests/test_password.py
View file

@ -121,6 +121,15 @@ class TestPassword(unittest.TestCase):
password = Password.PasswordFile(pwdfile.name, False, True, 'username', 'MyPassword') password = Password.PasswordFile(pwdfile.name, False, True, 'username', 'MyPassword')
password.delete() 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): def test_run_update(self):
pwdfile = tempfile.NamedTemporaryFile(suffix="passwordfile.tmp", mode='w+t') pwdfile = tempfile.NamedTemporaryFile(suffix="passwordfile.tmp", mode='w+t')
pwdfile.write('username:$2a$12$2C97xW0KC/vFp3YyjlOgU.fWXJ3EiGT2Ihb0SWN9Mw0XI4WngiUqS\n\n') pwdfile.write('username:$2a$12$2C97xW0KC/vFp3YyjlOgU.fWXJ3EiGT2Ihb0SWN9Mw0XI4WngiUqS\n\n')