add tests

pull/15/head
Andrew Wang 1 year ago
parent 300688cb9a
commit df412191e3
  1. 36
      ceo/cli/mysql.py
  2. 36
      ceo/cli/postgresql.py
  3. 4
      ceod/api/database.py
  4. 2
      ceod/db/MySQLService.py
  5. 2
      ceod/db/PostgreSQLService.py
  6. 77
      tests/ceo/cli/test_db_mysql.py
  7. 79
      tests/ceo/cli/test_db_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 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)

@ -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)

@ -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

@ -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()

@ -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()

@ -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)

@ -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)
Loading…
Cancel
Save