installers/charm/mariadb-k8s/tests/basic_deployment.py

   1 #!/usr/bin/python3
   2 # Copyright 2021 Canonical Ltd.
   3 #
   4 # Licensed under the Apache License, Version 2.0 (the "License"); you may
   5 # not use this file except in compliance with the License. You may obtain
   6 # a copy of the License at
   7 #
   8 #         http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 # Unless required by applicable law or agreed to in writing, software
  11 # distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  12 # WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  13 # License for the specific language governing permissions and limitations
  14 # under the License.
  15 #
  16 # For those usages not covered by the Apache License, Version 2.0 please
  17 # contact: legal@canonical.com
  18 #
  19 # To get in touch with the maintainers, please contact:
  20 # osm-charmers@lists.launchpad.net
  21 ##
  22
  23 import unittest
  24 import zaza.model as model
  25
  26 import mysql.connector as mysql
  27
  28 # from mysql.connector import errorcode
  29
  30 APPLICATION_NAME = "mariadb-k8s"
  31 UNIT_NAME = "mariadb-k8s/0"
  32 ROOT_USER = "root"
  33 ROOT_PASSWORD = "osm4u"
  34 USER = "mano"
  35 PASSWORD = "manopw"
  36 ACTION_SUCCESS_STATUS = "completed"
  37
  38
  39 def create_database(cnx, database_name):
  40     try:
  41         if not database_exists(cnx, database_name):
  42             cursor = cnx.cursor()
  43             cursor.execute(
  44                 "CREATE DATABASE {} DEFAULT CHARACTER SET 'utf8'".format(database_name)
  45             )
  46             return database_exists(cnx, database_name)
  47         else:
  48             return True
  49     except mysql.Error as err:
  50         print("Failed creating database {}: {}".format(database_name, err))
  51
  52
  53 def delete_database(cnx, database_name):
  54     try:
  55         if database_exists(cnx, database_name):
  56             cursor = cnx.cursor()
  57             cursor.execute("DROP DATABASE {}".format(database_name))
  58             return not database_exists(cnx, database_name)
  59         else:
  60             return True
  61     except mysql.Error as err:
  62         print("Failed deleting database {}: {}".format(database_name, err))
  63
  64
  65 def database_exists(cnx, database_name):
  66     try:
  67         cursor = cnx.cursor()
  68         cursor.execute("SHOW DATABASES")
  69         databases = cursor.fetchall()
  70         exists = False
  71         for database in databases:
  72             if database[0] == database_name:
  73                 exists = True
  74         cursor.close()
  75         return exists
  76     except mysql.Error as err:
  77         print("Failed deleting database {}: {}".format(database_name, err))
  78         return False
  79
  80
  81 class BasicDeployment(unittest.TestCase):
  82     def setUp(self):
  83         super().setUp()
  84         self.ip = model.get_status().applications[APPLICATION_NAME]["public-address"]
  85         try:
  86             self.cnx = mysql.connect(
  87                 user=ROOT_USER, password=ROOT_PASSWORD, host=self.ip
  88             )
  89         except mysql.Error as err:
  90             print("Couldn't connect to mariadb-k8s : {}".format(err))
  91
  92     def tearDown(self):
  93         super().tearDown()
  94         self.cnx.close()
  95
  96     def test_mariadb_connection_root(self):
  97         pass
  98
  99     def test_mariadb_connection_user(self):
 100         try:
 101             cnx = mysql.connect(user=USER, password=PASSWORD, host=self.ip)
 102             cnx.close()
 103         except mysql.Error as err:
 104             print("Couldn't connect to mariadb-k8s with user creds: {}".format(err))
 105
 106     def test_mariadb_create_database(self):
 107         created = create_database(self.cnx, "test_database")
 108         self.failIf(not created)
 109
 110     def test_mariadb_backup_action(self, db_name="test_backup"):
 111         created = create_database(self.cnx, db_name)
 112         self.failIf(not created)
 113         try:
 114             action = model.run_action(UNIT_NAME, "backup", raise_on_failure=True)
 115             self.assertEqual(action.status, ACTION_SUCCESS_STATUS)
 116         except model.ActionFailed as err:
 117             print("Action failed: {}".format(err))
 118
 119     def test_mariadb_remove_backup_action(self):
 120         self.test_mariadb_backup_action(db_name="test_remove_backup")
 121         try:
 122             action = model.run_action(UNIT_NAME, "remove-backup", raise_on_failure=True)
 123             self.assertEqual(action.status, ACTION_SUCCESS_STATUS)
 124         except model.ActionFailed as err:
 125             print("Action failed: {}".format(err))
 126
 127     def test_mariadb_restore_action(self):
 128         self.test_mariadb_backup_action(db_name="test_restore")
 129         deleted = delete_database(self.cnx, "test_restore")
 130         self.failIf(not deleted)
 131         try:
 132             action = model.run_action(UNIT_NAME, "restore", raise_on_failure=True)
 133             self.assertEqual(action.status, "completed")
 134             self.assertTrue(database_exists(self.cnx, "test_restore"))
 135         except model.ActionFailed as err:
 136             print("Action failed: {}".format(err))