check db config on startup

ceod/db/MySQLService.py

@@ -21,6 +21,26 @@ class MySQLService:
         config = component.getUtility(IConfig)
         self.auth_username = config.get('mysql_username')
         self.auth_password = config.get('mysql_password')
+        try:
+            test_user = "test_user_64559"
+            test_perms = f"""
+            CREATE USER '{test_user}'@'localhost';
+            CREATE DATABASE {test_user};
+            GRANT ALL PRIVILEGES ON {test_user}.* TO '{test_user}'@'localhost';
+            DROP DATABASE {test_user};
+            DROP USER '{test_user}'@'localhost';
+            """
+            with connect(
+                    host='localhost',
+                    user=self.auth_username,
+                    password=self.auth_password,
+            ) as con:
+                with con.cursor() as cursor:
+                    cursor.execute(test_perms)
+        except InterfaceError:
+            raise Exception('unable to connect or authenticate to sql server')
+        except ProgrammingError:
+            raise Exception('insufficient permissions to create users and databases')
 
     @contextmanager
     def mysql_connection(self):

ceod/db/PostgreSQLService.py

@@ -21,6 +21,26 @@ class PostgreSQLService:
         config = component.getUtility(IConfig)
         self.auth_username = config.get('postgresql_username')
         self.auth_password = config.get('postgresql_password')
+        try:
+            test_user = "test_user_64559"
+            test_perms = f"""
+            CREATE USER {test_user};
+            CREATE DATABASE {test_user} OWNER {test_user};
+            REVOKE ALL ON DATABASE {test_user} FROM PUBLIC;
+            DROP DATABASE {test_user};
+            DROP USER {test_user};
+            """
+            with connect(
+                    host='localhost',
+                    user=self.auth_username,
+                    password=self.auth_password,
+            ) as con:
+                with con.cursor() as cursor:
+                    cursor.execute(test_perms)
+        except OperationalError:
+            raise Exception('unable to connect or authenticate to sql server')
+        except ProgrammingError:
+            raise Exception('insufficient permissions to create users and databases')
 
     @contextmanager
     def psql_connection(self):

tests/ceod/api/test_db_mysql.py

@@ -5,8 +5,9 @@ from mysql.connector import connect
 from mysql.connector.errors import InterfaceError, ProgrammingError
 
 
-def test_api_create_mysql_db(cfg, client, g_admin_ctx, create_user_result):
-    uid = create_user_result['uid']
+def test_api_create_mysql_db(cfg, client, g_admin_ctx, ldap_user):
+    uid = ldap_user.uid
+
     with g_admin_ctx():
         user = User(uid='someone_else', cn='Some Name', terms=['s2021'])
         user.add_to_ldap()
@@ -72,13 +73,13 @@ def test_api_create_mysql_db(cfg, client, g_admin_ctx, create_user_result):
         user.remove_from_ldap()
 
 
-def test_api_passwd_reset_mysql(cfg, client, g_admin_ctx, create_user_result):
+def test_api_passwd_reset_mysql(cfg, client, g_admin_ctx, ldap_user):
+    uid = ldap_user.uid
+
     with g_admin_ctx():
         user = User(uid='someone_else', cn='Some Name', terms=['s2021'])
         user.add_to_ldap()
 
-    uid = create_user_result['uid']
-
     status, data = client.post(f"/api/mysql/{uid}", json={})
     assert status == 200
     assert 'password' in data

tests/ceod/api/test_db_psql.py

@@ -4,8 +4,9 @@ from ceod.model import User
 from psycopg2 import connect, OperationalError, ProgrammingError
 
 
-def test_api_create_psql_db(cfg, client, g_admin_ctx, create_user_result):
-    uid = create_user_result['uid']
+def test_api_create_psql_db(cfg, client, g_admin_ctx, ldap_user):
+    uid = ldap_user.uid
+
     with g_admin_ctx():
         user = User(uid='someone_else', cn='Some Name', terms=['s2021'])
         user.add_to_ldap()
@@ -71,13 +72,13 @@ def test_api_create_psql_db(cfg, client, g_admin_ctx, create_user_result):
         user.remove_from_ldap()
 
 
-def test_api_passwd_reset_psql(cfg, client, g_admin_ctx, create_user_result):
+def test_api_passwd_reset_psql(cfg, client, g_admin_ctx, ldap_user):
+    uid = ldap_user.uid
+
     with g_admin_ctx():
         user = User(uid='someone_else', cn='Some Name', terms=['s2021'])
         user.add_to_ldap()
 
-    uid = create_user_result['uid']
-
     status, data = client.post(f"/api/postgresql/{uid}", json={})
     assert status == 200
     assert 'password' in data