openstack/trove-tempest-plugin
Code Issues Proposed changes

Add tests: user management

Browse Source 
Change-Id: Iea8d73af802de3b35cbd6e14eed7e5c12280974c
This commit is contained in:
Lingxian Kong 2020-06-16 13:47:52 +12:00
parent 6186071009
commit 8d0e7345f9
3 changed files with 126 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
trove_tempest_plugin/services/client.py
View File

@ -71,7 +71,7 @@ class TroveClient(rest_client.RestClient):
self.delete_resource('instances', id, ignore_notfound=True)
def create_resource(self, obj, req_body, extra_headers={},
expected_status_code=200):
expected_status_code=200, need_response=True):
headers = {"Content-Type": "application/json"}
headers = dict(headers, **extra_headers)
url = '/%s' % obj
@ -79,7 +79,8 @@ class TroveClient(rest_client.RestClient):
resp, body = self.post(url, json.dumps(req_body), headers=headers)
self.expected_success(expected_status_code, resp.status)
return rest_client.ResponseBody(resp, json.loads(body))
if need_response:
return rest_client.ResponseBody(resp, json.loads(body))
def patch_resource(self, obj, id, req_body, expected_status_code=202):
url = '/{obj}/{id}'.format(obj=obj, id=id)
@ -87,3 +88,10 @@ class TroveClient(rest_client.RestClient):
resp, _ = self.patch(url, json.dumps(req_body), headers=headers)
self.expected_success(expected_status_code, resp.status)
def put_resource(self, url, req_body, expected_status_code=202):
url = '/%s' % url
headers = {"Content-Type": "application/json"}
resp, _ = self.put(url, json.dumps(req_body), headers=headers)
self.expected_success(expected_status_code, resp.status)

2
trove_tempest_plugin/tests/constants.py
View File

@ -13,5 +13,5 @@
# limitations under the License.
DB_USER = 'test_user'
DB_PASS = 'password'
DB_PASS = 'PassW0rd'
DB_NAME = 'test_db'

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

@ -13,6 +13,7 @@
# limitations under the License.
from oslo_log import log as logging
from tempest.lib import decorators
from tempest.lib import exceptions
from trove_tempest_plugin.tests import base as trove_base
from trove_tempest_plugin.tests import constants
@ -23,10 +24,9 @@ LOG = logging.getLogger(__name__)
class TestInstanceBasicMySQLBase(trove_base.BaseTroveTest):
def _access_db(self, ip, username=constants.DB_USER,
password=constants.DB_PASS):
LOG.info('Trying to access the database %s', ip)
db_url = f'mysql+pymysql://{username}:{password}@{ip}:3306'
password=constants.DB_PASS, database=constants.DB_NAME):
db_url = f'mysql+pymysql://{username}:{password}@{ip}:3306/{database}'
LOG.info(f'Trying to access the database {db_url}')
db_client = utils.SQLClient(db_url)
cmd = "SELECT 1;"
@ -43,3 +43,114 @@ class TestInstanceBasicMySQLBase(trove_base.BaseTroveTest):
self.assertIn(constants.DB_USER, user_names)
self._access_db(self.instance_ip)
@decorators.idempotent_id("c5a9dcda-af5b-11ea-b87c-00224d6b7bc1")
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)
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)
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
self._access_db(self.instance_ip, user1, constants.DB_PASS, db1)
# user2 should not have access to db2
self.assertRaises(exceptions.TempestException, self._access_db,
self.instance_ip, user2, constants.DB_PASS, db2)
LOG.info(f"Granting user {user2} access to database {db2}")
grant_access = {
"databases": [
{
"name": db2
}
]
}
self.client.put_resource(
f'/instances/{self.instance_id}/users/{user2}/databases',
grant_access)
user2_dbs = self.client.list_resources(
f'instances/{self.instance_id}/users/{user2}/databases')
user2_dbs = [db['name'] for db in user2_dbs['databases']]
self.assertIn(db2, user2_dbs)
# Now user2 should have access to db2
self._access_db(self.instance_ip, user2, constants.DB_PASS, db2)
LOG.info(f"Revoking user {user2} access to database {db2}")
self.client.delete_resource(
f'instances/{self.instance_id}/users/{user2}/databases', db2)
# user2 should not have access to db2
self.assertRaises(exceptions.TempestException, self._access_db,
self.instance_ip, user2, constants.DB_PASS, db2)
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)
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)