From df412191e3ed27e4dbff9e58bde9c82a466949f1 Mon Sep 17 00:00:00 2001 From: Andrew Wang Date: Fri, 10 Sep 2021 01:39:46 -0400 Subject: [PATCH] add tests --- ceo/cli/mysql.py | 36 ++++++++----- ceo/cli/postgresql.py | 36 ++++++++----- ceod/api/database.py | 4 +- ceod/db/MySQLService.py | 2 + ceod/db/PostgreSQLService.py | 2 + tests/ceo/cli/test_db_mysql.py | 77 ++++++++++++++++++++++++++++ tests/ceo/cli/test_db_postgresql.py | 79 +++++++++++++++++++++++++++++ 7 files changed, 210 insertions(+), 26 deletions(-) create mode 100644 tests/ceo/cli/test_db_mysql.py create mode 100644 tests/ceo/cli/test_db_postgresql.py diff --git a/ceo/cli/mysql.py b/ceo/cli/mysql.py index 30786c3..4098ea1 100644 --- a/ceo/cli/mysql.py +++ b/ceo/cli/mysql.py @@ -4,11 +4,11 @@ import os from zope import component from ceo_common.interfaces import IConfig -from ..utils import http_post, http_get -from .utils import handle_sync_response, check_file_path +from ..utils import http_post, http_get, http_delete +from .utils import handle_sync_response, check_file_path, check_if_in_development -def mysql_create_info_file(file, username, password): +def mysql_cli_response(file, username, password): cfg_srv = component.getUtility(IConfig) mysql_host = cfg_srv.get('mysql_host') info = f"""MySQL Database Information for {username} @@ -34,6 +34,17 @@ def mysql_create_info_file(file, username, password): os.chown(file, username, username) os.chmod(file, 0o640) + click.echo(f"""MySQL database created + + Connection Information: + + Database: {username} + Username: {username} + Password: {password} + Host: {mysql_host} + + Settings and more info has been written to {file}""") + @click.group(short_help='Perform operations on MySQL') def mysql(): @@ -53,11 +64,7 @@ def create(username): result = handle_sync_response(resp) password = result['password'] - mysql_create_info_file(info_file_path, username, password) - - click.echo(f""" - MySQL database {username} with password {password} has been created - This password and more details have been written to {info_file_path}""") + mysql_cli_response(info_file_path, username, password) @mysql.command(short_help='Reset the password of a MySQL user') @@ -73,8 +80,13 @@ def pwreset(username): result = handle_sync_response(resp) password = result['password'] - mysql_create_info_file(info_file_path, username, password) + mysql_cli_response(info_file_path, username, password) - click.echo(f""" - MySQL database {username} now has the password {password} - This password and more details have been written to {info_file_path}""") + +@mysql.command(short_help="Delete the database of a MySQL user") +@click.argument('username') +def delete(username): + check_if_in_development() + click.confirm(f"Are you sure?", abort=True) + resp = http_delete(f'/api/db/mysql/{username}') + handle_sync_response(resp) diff --git a/ceo/cli/postgresql.py b/ceo/cli/postgresql.py index 7127ac8..4807d7e 100644 --- a/ceo/cli/postgresql.py +++ b/ceo/cli/postgresql.py @@ -4,11 +4,11 @@ import os from zope import component from ceo_common.interfaces import IConfig -from ..utils import http_post, http_get -from .utils import handle_sync_response, check_file_path +from ..utils import http_post, http_get, http_delete +from .utils import handle_sync_response, check_file_path, check_if_in_development -def psql_create_info_file(file, username, password): +def psql_cli_response(file, username, password): cfg_srv = component.getUtility(IConfig) psql_host = cfg_srv.get('postgresql_host') info = f"""PostgreSQL Database Information for {username} @@ -34,6 +34,17 @@ def psql_create_info_file(file, username, password): os.chown(file, username, username) os.chmod(file, 0o640) + click.echo(f"""PostgreSQL database created + + Connection Information: + + Database: {username} + Username: {username} + Password: {password} + Host: {psql_host} + + Settings and more info has been written to {file}""") + @click.group(short_help='Perform operations on PostgreSQL') def postgresql(): @@ -53,11 +64,7 @@ def create(username): result = handle_sync_response(resp) password = result['password'] - psql_create_info_file(info_file_path, username, password) - - click.echo(f""" - PostgreSQL database {username} with password {password} has been created - This password and more details have been written to {info_file_path}""") + psql_cli_response(info_file_path, username, password) @postgresql.command(short_help='Reset the password of a PostgreSQL user') @@ -73,8 +80,13 @@ def pwreset(username): result = handle_sync_response(resp) password = result['password'] - psql_create_info_file(info_file_path, username, password) + psql_cli_response(info_file_path, username, password) - click.echo(f""" - PostgreSQL database {username} now has the password {password} - This password and more details have been written to {info_file_path}""") + +@postgresql.command(short_help="Delete the database of a PostgreSQL user") +@click.argument('username') +def delete(username): + check_if_in_development() + click.confirm(f"Are you sure?", abort=True) + resp = http_delete(f'/api/db/postgresql/{username}') + handle_sync_response(resp) diff --git a/ceod/api/database.py b/ceod/api/database.py index ca81367..94e9dc0 100644 --- a/ceod/api/database.py +++ b/ceod/api/database.py @@ -31,9 +31,9 @@ def db_exception_handler(func): except InvalidUsernameError: return {'error': 'username contains invalid characters'}, 400 except DatabaseConnectionError: - return {'error': 'unable to connect or authenticate to sql server'}, 500 + return {'error': 'unable to connect to sql server'}, 500 except DatabasePermissionError: - return {'error': 'unable to perform action due to permissions'}, 500 + return {'error': 'unable to connect or action failed due to permissions'}, 500 return function diff --git a/ceod/db/MySQLService.py b/ceod/db/MySQLService.py index 629af96..10889c9 100644 --- a/ceod/db/MySQLService.py +++ b/ceod/db/MySQLService.py @@ -35,9 +35,11 @@ class MySQLService: password=self.auth_password, ) as con: yield con + # unable to connect except OperationalError as e: logger.error(e) raise DatabaseConnectionError() + # invalid credentials / user does not exist / invalid permissions for action except ProgrammingError as e: logger.error(e) raise DatabasePermissionError() diff --git a/ceod/db/PostgreSQLService.py b/ceod/db/PostgreSQLService.py index 3f3cbb8..e01e3e1 100644 --- a/ceod/db/PostgreSQLService.py +++ b/ceod/db/PostgreSQLService.py @@ -39,9 +39,11 @@ class PostgreSQLService: ) con.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT) yield con + # unable to connect / invalid credentials / user does not exist except OperationalError as e: logger.error(e) raise DatabaseConnectionError() + # invalid permissions for action except ProgrammingError as e: logger.error(e) raise DatabasePermissionError() diff --git a/tests/ceo/cli/test_db_mysql.py b/tests/ceo/cli/test_db_mysql.py new file mode 100644 index 0000000..758e9c9 --- /dev/null +++ b/tests/ceo/cli/test_db_mysql.py @@ -0,0 +1,77 @@ +import pytest, os + +from click.testing import CliRunner +from ceo.cli import cli + +from mysql.connector import connect +from mysql.connector.errors import ProgrammingError + + +def mysql_attempt_connection(host, username, password): + with connect( + host=host, + user=username, + password=password, + ) as con, con.cursor() as cur: + cur.execute("SHOW DATABASES") + response = cur.fetchall() + assert len(response) == 2 + + with pytest.raises(ProgrammingError): + cur.execute("CREATE DATABASE new_db") + + +def test_mysql(cli_setup, cfg, ldap_user): + runner = CliRunner() + + username = ldap_user.uid + host = cfg.get("mysql_host") + info_file_path = os.path.join(ldap_user.home_directory, "ceo-mysql-info") + assert not os.path.isfile(info_file_path) + + # create database for user + result = runner.invoke(cli, ['mysql', 'create', username]) + assert result.exit_code == 0 + assert os.path.isfile(info_file_path) + + response_arr = result.output.split() + passwd = response_arr[response_arr.index("Password:") + 1] + with open(info_file_path, 'r') as file: + old_info = file.read() + + expected = f"""MySQL database created + + Connection Information: + + Database: {username} + Username: {username} + Password: {passwd} + Host: {host} + + Settings and more info has been written to {info_file_path}""" + + assert result.output == expected + mysql_attempt_connection(host, username, passwd) + + # perform password reset for user + result = runner.invoke(cli, ['mysql', 'pwreset', username], input="y\n") + assert result.exit_code == 0 + + response_arr = result.output.split() + new_passwd = response_arr[response_arr.index("Password:") + 1] + with open(info_file_path, 'r') as file: + new_info = file.read() + + assert new_passwd != passwd + assert old_info != new_info + mysql_attempt_connection(host, username, new_passwd) + + # delete database and file + result = runner.invoke(cli, ['mysql', 'delete', username], input="y\n") + assert result.exit_code == 0 + + # user should be deleted + with pytest.raises(ProgrammingError): + mysql_attempt_connection(host, username, passwd) + + os.remove(info_file_path) diff --git a/tests/ceo/cli/test_db_postgresql.py b/tests/ceo/cli/test_db_postgresql.py new file mode 100644 index 0000000..06f47c2 --- /dev/null +++ b/tests/ceo/cli/test_db_postgresql.py @@ -0,0 +1,79 @@ +import pytest, os + +from click.testing import CliRunner +from ceo.cli import cli + +from psycopg2 import connect, OperationalError, ProgrammingError + + +def psql_attempt_connection(host, username, password): + con = connect( + host=host, + user=username, + password=password, + ) + con.autocommit = True + with con.cursor() as cur: + cur.execute("SELECT datname FROM pg_database") + response = cur.fetchall() + # 3 of the 4 are postgres, template0, template1 + assert len(response) == 4 + with pytest.raises(ProgrammingError): + cur.execute("CREATE DATABASE new_db") + con.close() + + +def test_postgresql(cli_setup, cfg, ldap_user): + runner = CliRunner() + + username = ldap_user.uid + host = cfg.get("postgresql_host") + info_file_path = os.path.join(ldap_user.home_directory, "ceo-psql-info") + assert not os.path.isfile(info_file_path) + + # create database for user + result = runner.invoke(cli, ['postgresql', 'create', username]) + assert result.exit_code == 0 + assert os.path.isfile(info_file_path) + + response_arr = result.output.split() + passwd = response_arr[response_arr.index("Password:") + 1] + with open(info_file_path, 'r') as file: + old_info = file.read() + + expected = f"""PostgreSQL database created + + Connection Information: + + Database: {username} + Username: {username} + Password: {passwd} + Host: {host} + + Settings and more info has been written to {info_file_path}""" + + assert result.output == expected + psql_attempt_connection(host, username, passwd) + + # perform password reset for user + result = runner.invoke(cli, ['postgresql', 'pwreset', username], input="y\n") + assert result.exit_code == 0 + + response_arr = result.output.split() + new_passwd = response_arr[response_arr.index("Password:") + 1] + with open(info_file_path, 'r') as file: + new_info = file.read() + + assert new_passwd != passwd + assert old_info != new_info + psql_attempt_connection(host, username, new_passwd) + + # delete database and file + result = runner.invoke(cli, ['postgresql', 'delete', username], input="y\n") + assert result.exit_code == 0 + + # user should be deleted + with pytest.raises(OperationalError): + psql_attempt_connection(host, username, passwd) + + os.remove(info_file_path)