2 # Copyright 2021 Canonical Ltd.
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
8 # http://www.apache.org/licenses/LICENSE-2.0
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
16 # For those usages not covered by the Apache License, Version 2.0 please
17 # contact: legal@canonical.com
19 # To get in touch with the maintainers, please contact:
20 # osm-charmers@lists.launchpad.net
24 import zaza
.model
as model
26 import mysql
.connector
as mysql
28 # from mysql.connector import errorcode
30 APPLICATION_NAME
= "mariadb-k8s"
31 UNIT_NAME
= "mariadb-k8s/0"
33 ROOT_PASSWORD
= "osm4u"
36 ACTION_SUCCESS_STATUS
= "completed"
39 def create_database(cnx
, database_name
):
41 if not database_exists(cnx
, database_name
):
44 "CREATE DATABASE {} DEFAULT CHARACTER SET 'utf8'".format(database_name
)
46 return database_exists(cnx
, database_name
)
49 except mysql
.Error
as err
:
50 print("Failed creating database {}: {}".format(database_name
, err
))
53 def delete_database(cnx
, database_name
):
55 if database_exists(cnx
, database_name
):
57 cursor
.execute("DROP DATABASE {}".format(database_name
))
58 return not database_exists(cnx
, database_name
)
61 except mysql
.Error
as err
:
62 print("Failed deleting database {}: {}".format(database_name
, err
))
65 def database_exists(cnx
, database_name
):
68 cursor
.execute("SHOW DATABASES")
69 databases
= cursor
.fetchall()
71 for database
in databases
:
72 if database
[0] == database_name
:
76 except mysql
.Error
as err
:
77 print("Failed deleting database {}: {}".format(database_name
, err
))
81 class BasicDeployment(unittest
.TestCase
):
84 self
.ip
= model
.get_status().applications
[APPLICATION_NAME
]["public-address"]
86 self
.cnx
= mysql
.connect(
87 user
=ROOT_USER
, password
=ROOT_PASSWORD
, host
=self
.ip
89 except mysql
.Error
as err
:
90 print("Couldn't connect to mariadb-k8s : {}".format(err
))
96 def test_mariadb_connection_root(self
):
99 def test_mariadb_connection_user(self
):
101 cnx
= mysql
.connect(user
=USER
, password
=PASSWORD
, host
=self
.ip
)
103 except mysql
.Error
as err
:
104 print("Couldn't connect to mariadb-k8s with user creds: {}".format(err
))
106 def test_mariadb_create_database(self
):
107 created
= create_database(self
.cnx
, "test_database")
108 self
.failIf(not created
)
110 def test_mariadb_backup_action(self
, db_name
="test_backup"):
111 created
= create_database(self
.cnx
, db_name
)
112 self
.failIf(not created
)
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
))
119 def test_mariadb_remove_backup_action(self
):
120 self
.test_mariadb_backup_action(db_name
="test_remove_backup")
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
))
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
)
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
))