openstack/trove-tempest-plugin
Code Issues Proposed changes

Add basic tests for postgresql database

Browse Source 
Change-Id: Ia901a8ec9c9254283332402c4e27e8961d2b57c2
This commit is contained in:
wu.chunyang 2024-09-24 16:34:28 +08:00
parent 88b343c42c
commit d3b20510dd
2 changed files with 164 additions and 1 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
trove_tempest_plugin/tests/scenario/base_basic.py
View File

@ -256,6 +256,13 @@ class TestInstanceBasicMySQLBase(TestInstanceBasicBase):
cur_db_names = [db['name'] for db in databases]
self.assertIn(db1, cur_db_names)
self.assertNotIn(db2, cur_db_names)
# test update_attributes interface
new_user_body = {"user": {"name": "new_user"}}
self.client.put_resource(
f'instances/{self.instance_id}/users/{user1}', new_user_body)
users = self.get_users(self.instance_id)
cur_user_names = [user['name'] for user in users]
self.assertIn("new_user", cur_user_names)
@decorators.idempotent_id("ce8277b0-af7c-11ea-b87c-00224d6b7bc1")
def test_configuration(self):

158
trove_tempest_plugin/tests/scenario/test_instance_basic.py
View File

@ -11,9 +11,13 @@
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
import time
from oslo_log import log as logging
from tempest.lib import decorators
from trove_tempest_plugin.tests import constants
from trove_tempest_plugin.tests.scenario import base_basic
from trove_tempest_plugin.tests import utils
@ -30,9 +34,31 @@ class TestInstanceBasicMariaDB(base_basic.TestInstanceBasicMySQLBase):
class TestInstanceBasicPostgreSQL(base_basic.TestInstanceBasicBase):
datastore = 'postgresql'
create_user = False
create_user = True
enable_root = True
def _check_db_privilege(self, ip, username, password, database):
db_url = f'postgresql://{username}:{password}@{ip}:5432/{database}'
with utils.SQLClient(db_url) as db_client:
# Check if the user can create a table
try:
cmd = "CREATE TABLE test_table (id INT);"
db_client.pgsql_execute(cmd)
# If successful, drop the table
cmd = "DROP TABLE test_table;"
db_client.pgsql_execute(cmd)
return True
except Exception:
return False
def _access_db(self, ip, username=constants.DB_USER,
password=constants.DB_PASS, database=constants.DB_NAME):
db_url = f'postgresql+psycopg2://{username}:{password}@{ip}:5432/'\
f'{database}'
with utils.SQLClient(db_url) as db_client:
cmd = "SELECT 1;"
db_client.pgsql_execute(cmd)
def get_config_value(self, ip, option):
db_url = (f'postgresql+psycopg2://root:{self.password}@'
f'{ip}:5432/postgres')
@ -44,6 +70,136 @@ class TestInstanceBasicPostgreSQL(base_basic.TestInstanceBasicBase):
self.assertEqual(1, len(rows))
return int(rows[0][0])
@decorators.idempotent_id('e9f78628-8bf9-46fb-84b2-367de3f0e0fe')
def test_database_access(self):
databases = self.get_databases(self.instance_id)
db_names = [db['name'] for db in databases]
self.assertIn(constants.DB_NAME, db_names)
users = self.get_users(self.instance_id)
user_names = [user['name'] for user in users]
self.assertIn(constants.DB_USER, user_names)
LOG.info(f"Accessing database on {self.instance_ip}")
self._access_db(self.instance_ip)
@decorators.idempotent_id('6c2c710c-0138-4215-8e08-6dfe605ba6a6')
def test_user_database(self):
db1 = 'foo'
db2 = 'bar'
user1 = 'foo_user'
user2 = 'bar_user'
users = self.get_users(self.instance_id)
cur_user_names = [user['name'] for user in users]
self.assertNotIn(user1, cur_user_names)
self.assertNotIn(user2, cur_user_names)
databases = self.get_databases(self.instance_id)
cur_db_names = [db['name'] for db in databases]
self.assertNotIn(db1, cur_db_names)
self.assertNotIn(db2, cur_db_names)
LOG.info(f"Creating databases in instance {self.instance_id}")
create_db = {
"databases": [
{
"name": db1
},
{
"name": db2
}
]
}
self.client.create_resource(f"instances/{self.instance_id}/databases",
create_db, expected_status_code=202,
need_response=False)
# sleep 3s to wait for db creation
time.sleep(3)
databases = self.get_databases(self.instance_id)
cur_db_names = [db['name'] for db in databases]
self.assertIn(db1, cur_db_names)
self.assertIn(db2, cur_db_names)
LOG.info(f"Creating users in instance {self.instance_id}")
create_user = {
"users": [
{
"databases": [
{
"name": db1
}
],
"name": user1,
"password": constants.DB_PASS
},
{
"name": user2,
"password": constants.DB_PASS
}
]
}
self.client.create_resource(f"instances/{self.instance_id}/users",
create_user, expected_status_code=202,
need_response=False)
# wait the users to be created.
time.sleep(3)
users = self.get_users(self.instance_id)
cur_user_names = [user['name'] for user in users]
self.assertIn(user1, cur_user_names)
self.assertIn(user2, cur_user_names)
# user1 should have access to db1
LOG.info(f"Accessing database on {self.instance_ip}, user: {user1}, "
f"db: {db1}")
self._access_db(self.instance_ip, user1, constants.DB_PASS, db1)
# user1 should also have previlege to create table
self.assertTrue(self._check_db_privilege(self.instance_ip,
user1,
constants.DB_PASS, db1))
# by default, users in postgresql are allowed to
# connect to other databases
self._access_db(self.instance_ip, user2, constants.DB_PASS, db2)
self.assertTrue(self._check_db_privilege(self.instance_ip,
user2,
constants.DB_PASS, db2))
LOG.info(f"Revoking user {user1} access to database {db1}")
self.client.delete_resource(
f'instances/{self.instance_id}/users/{user1}/databases', db1)
# FIXME: user1 still have access to db1 because of the
# havior of public schema
# self.assertRaises(exceptions.TempestException, self._access_db,
# self.instance_ip, user1, constants.DB_PASS, db1)
# self.assertFalse(self._check_db_privilege(self.instance_ip,
# user1,
# constants.DB_PASS, db1))
LOG.info(f"Deleting user {user2}")
self.client.delete_resource(
f'instances/{self.instance_id}/users', user2)
users = self.get_users(self.instance_id)
cur_user_names = [user['name'] for user in users]
self.assertIn(user1, cur_user_names)
self.assertNotIn(user2, cur_user_names)
LOG.info(f"Deleting database {db2}")
self.client.delete_resource(
f'instances/{self.instance_id}/databases', db2)
time.sleep(2)
databases = self.get_databases(self.instance_id)
cur_db_names = [db['name'] for db in databases]
self.assertIn(db1, cur_db_names)
self.assertNotIn(db2, cur_db_names)
# test update_attributes interface
LOG.info(f"Updating user {user1} to new_user")
new_user_body = {"user": {"name": "new_user"}}
self.client.put_resource(
f'instances/{self.instance_id}/users/{user1}', new_user_body)
time.sleep(2)
users = self.get_users(self.instance_id)
cur_user_names = [user['name'] for user in users]
self.assertIn("new_user", cur_user_names)
@decorators.idempotent_id("b6c03cb6-f40f-11ea-a950-00224d6b7bc1")
def test_configuration(self):
# Default is 100