add tests for mariadb

seed/dovecot/tests/test_imap.py

@@ -1,6 +1,5 @@
 from yaml import load, SafeLoader
 from os import environ
-from os.path import isdir
 import pytest
 
 from imaplib2 import IMAP4_SSL
@@ -33,12 +32,8 @@ def test_imap_wrong_password(typ, username, password):
 
 @pytest.mark.parametrize('typ, username, password', parameters)
 def test_imap_migration(typ, username, password):
-    if typ == 'family':
-        dirname = f'/var/lib/risotto/srv/{data["dns"]}/home/families/{data["name_family"]}/{username}'
-    else:
-        dirname = f'/var/lib/risotto/srv/{data["dns"]}/home/users/{username}'
     msg = get_msg(username, 'MIGRATION')
-    if not isdir(dirname):
+    if 'FIRST_RUN' in environ:
         smtp = SMTP(data['address'], '587')
         smtp.starttls()
         smtp.login(username, password)

seed/mariadb/dictionaries/20_mariadb.xml

@@ -6,6 +6,7 @@
       <file>/etc/my.cnf.d/risotto.cnf</file>
       <file engine="none" source="tmpfile-mariadb.conf">/tmpfiles.d/0mariadb.conf</file>
       <file mode="600" owner="mysql" group="mysql">/etc/mariadb.sql</file>
+	  <file>/tests/mariadb.yml</file>
     </service>
   </services>
   <variables>

seed/mariadb/templates/mariadb.sql

@@ -1,8 +1,13 @@
+%set %%new_accounts = [('_gateway', 'rougail_test', %%get_password(server_name=%%domain_name_eth0, username='rougail_test', description="remote", type="cleartext", hide=%%hide_secret, temporary=True))]
 %for %%server in %%accounts.remotes
  %set %%name = %%normalize_family(%%server)
  %set %%password = %%accounts['remote_' + %%name]['password_' + %%name]
+ %%new_accounts.append((%%str(%%server), %%name, %%password))
+%end for
+%for %%server, %%name, %%password in %%new_accounts
 CREATE USER IF NOT EXISTS '%%name'@'%%server' IDENTIFIED BY '%%password';
 CREATE DATABASE IF NOT EXISTS %%name CHARACTER SET utf8;
 GRANT ALL PRIVILEGES ON %%name.* TO '%%name'@'%%server' IDENTIFIED BY '%%password';
-FLUSH PRIVILEGES;
 %end for
+FLUSH PRIVILEGES;
+

seed/mariadb/templates/mariadb.yml

@@ -0,0 +1,4 @@
+address: %%ip_eth0
+user: rougail_test
+password: %%get_password(server_name=%%domain_name_eth0, username='rougail_test', description="remote", type="cleartext", hide=%%hide_secret, temporary=True)
+dbname: rougail_test

seed/mariadb/tests/test_mariadb.py

@@ -0,0 +1,80 @@
+from yaml import load, SafeLoader
+from os import environ
+from pytest import raises
+
+from pymysql import connect
+from pymysql.err import OperationalError
+
+
+def test_mariadb_wrong_password():
+    conf_file = f'{environ["MACHINE_TEST_DIR"]}/mariadb.yml'
+    with open(conf_file) as yaml:
+        data = load(yaml, Loader=SafeLoader)
+    with raises(OperationalError):
+        connect(data['address'], data['user'], 'a', data['dbname'])
+
+
+def test_mariadb_connection():
+    conf_file = f'{environ["MACHINE_TEST_DIR"]}/mariadb.yml'
+    with open(conf_file) as yaml:
+        data = load(yaml, Loader=SafeLoader)
+    db = connect(data['address'], data['user'], data['password'], data['dbname'])
+    db.close()
+
+
+def test_mariadb_migration():
+    conf_file = f'{environ["MACHINE_TEST_DIR"]}/mariadb.yml'
+    with open(conf_file) as yaml:
+        data = load(yaml, Loader=SafeLoader)
+    db = connect(data['address'], data['user'], data['password'], data['dbname'])
+    cursor = db.cursor()
+    if 'FIRST_RUN' in environ:
+        sql = """CREATE TABLE test (col  CHAR(20) NOT NULL)"""
+        cursor.execute(sql)
+        sql = """INSERT INTO test (col) VALUES ("test")"""
+        cursor.execute(sql)
+        db.commit()
+    sql = """SELECT * FROM test"""
+    cursor.execute(sql)
+    results = cursor.fetchall()
+    assert len(results) == 1
+    results[0] == ('test',)
+    cursor.close()
+    db.close()
+
+
+def test_mariadb_insert():
+    conf_file = f'{environ["MACHINE_TEST_DIR"]}/mariadb.yml'
+    with open(conf_file) as yaml:
+        data = load(yaml, Loader=SafeLoader)
+    db = connect(data['address'], data['user'], data['password'], data['dbname'])
+    cursor = db.cursor()
+    sql = """INSERT INTO test (col) VALUES ("test2")"""
+    cursor.execute(sql)
+    db.commit()
+    #
+    sql = """SELECT * FROM test WHERE col = 'test2'"""
+    cursor.execute(sql)
+    results = cursor.fetchall()
+    assert len(results) == 1
+    results[0] == ('test2',)
+    cursor.close()
+    db.close()
+
+
+def test_mariadb_delete():
+    conf_file = f'{environ["MACHINE_TEST_DIR"]}/mariadb.yml'
+    with open(conf_file) as yaml:
+        data = load(yaml, Loader=SafeLoader)
+    db = connect(data['address'], data['user'], data['password'], data['dbname'])
+    cursor = db.cursor()
+    sql = """DELETE FROM test WHERE col = 'test2'"""
+    cursor.execute(sql)
+    db.commit()
+    #
+    sql = """SELECT * FROM test WHERE col = 'test2'"""
+    cursor.execute(sql)
+    results = cursor.fetchall()
+    assert len(results) == 0
+    cursor.close()
+    db.close()