94 lines
2.8 KiB
Python
94 lines
2.8 KiB
Python
import os
|
|
import shutil
|
|
import time
|
|
|
|
from click.testing import CliRunner
|
|
from mysql.connector import connect
|
|
from mysql.connector.errors import ProgrammingError
|
|
import pytest
|
|
|
|
from ceo.cli import cli
|
|
|
|
|
|
def mysql_attempt_connection(host, username, password):
|
|
time.sleep(0.5)
|
|
with connect(
|
|
host=host,
|
|
user=username,
|
|
password=password,
|
|
) as con, con.cursor() as cur:
|
|
found = False
|
|
# Sometimes, when running the tests locally, I've observed a race condition
|
|
# where another client can't "see" a database right after we create it.
|
|
# I only observed this after upgrading the containers to bullseye.
|
|
for _ in range(4):
|
|
cur.execute("SHOW DATABASES")
|
|
response = cur.fetchall()
|
|
if len(response) == 2:
|
|
found = True
|
|
break
|
|
time.sleep(0.25)
|
|
assert found
|
|
|
|
with pytest.raises(ProgrammingError):
|
|
cur.execute("CREATE DATABASE new_db")
|
|
|
|
|
|
def test_mysql(cli_setup, cfg, ldap_user):
|
|
runner = CliRunner()
|
|
|
|
username = ldap_user.uid
|
|
os.makedirs(ldap_user.home_directory)
|
|
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], input='y\n')
|
|
#print(result.output) # noqa: E265
|
|
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"""Are you sure you want to create a MySQL database for {username}? [y/N]: y
|
|
MySQL database created.
|
|
Connection Information:
|
|
|
|
Database: {username}
|
|
Username: {username}
|
|
Password: {passwd}
|
|
Host: {host}
|
|
|
|
These settings have 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)
|
|
|
|
shutil.rmtree(ldap_user.home_directory)
|