190 lines
7.4 KiB
Python
190 lines
7.4 KiB
Python
# Copyright 2011 OpenStack Foundation
|
|
#
|
|
# Licensed under the Apache License, Version 2.0 (the "License"); you may
|
|
# not use this file except in compliance with the License. You may obtain
|
|
# a copy of the License at
|
|
#
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
#
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
# distributed under the License is distributed on an "AS IS" BASIS, 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 proboscis.asserts import assert_equal
|
|
from proboscis.asserts import assert_false
|
|
from proboscis.asserts import assert_raises
|
|
from proboscis.asserts import assert_true
|
|
from proboscis import before_class
|
|
from proboscis import test
|
|
from troveclient.compat import exceptions
|
|
|
|
from trove import tests
|
|
from trove.tests.api.instances import instance_info
|
|
from trove.tests import util
|
|
from trove.tests.util import test_config
|
|
|
|
FAKE = test_config.values['fake_mode']
|
|
|
|
|
|
@test(depends_on_groups=[tests.DBAAS_API_USERS_ACCESS],
|
|
groups=[tests.DBAAS_API_DATABASES])
|
|
class TestDatabases(object):
|
|
"""Test the creation and deletion of additional MySQL databases"""
|
|
|
|
dbname = "third #?@some_-"
|
|
dbname_urlencoded = "third%20%23%3F%40some_-"
|
|
|
|
dbname2 = "seconddb"
|
|
created_dbs = [dbname, dbname2]
|
|
system_dbs = ['information_schema', 'mysql', 'lost+found']
|
|
|
|
@before_class
|
|
def setUp(self):
|
|
self.dbaas = util.create_dbaas_client(instance_info.user)
|
|
self.dbaas_admin = util.create_dbaas_client(instance_info.admin_user)
|
|
|
|
@test
|
|
def test_cannot_create_taboo_database_names(self):
|
|
for name in self.system_dbs:
|
|
databases = [{"name": name, "character_set": "latin2",
|
|
"collate": "latin2_general_ci"}]
|
|
assert_raises(exceptions.BadRequest, self.dbaas.databases.create,
|
|
instance_info.id, databases)
|
|
assert_equal(400, self.dbaas.last_http_code)
|
|
|
|
@test
|
|
def test_create_database(self):
|
|
databases = []
|
|
databases.append({"name": self.dbname, "character_set": "latin2",
|
|
"collate": "latin2_general_ci"})
|
|
databases.append({"name": self.dbname2})
|
|
|
|
self.dbaas.databases.create(instance_info.id, databases)
|
|
assert_equal(202, self.dbaas.last_http_code)
|
|
if not FAKE:
|
|
time.sleep(5)
|
|
|
|
@test(depends_on=[test_create_database])
|
|
def test_create_database_list(self):
|
|
databases = self.dbaas.databases.list(instance_info.id)
|
|
assert_equal(200, self.dbaas.last_http_code)
|
|
found = False
|
|
for db in self.created_dbs:
|
|
for result in databases:
|
|
if result.name == db:
|
|
found = True
|
|
assert_true(found, "Database '%s' not found in result" % db)
|
|
found = False
|
|
|
|
@test(depends_on=[test_create_database])
|
|
def test_fails_when_creating_a_db_twice(self):
|
|
databases = []
|
|
databases.append({"name": self.dbname, "character_set": "latin2",
|
|
"collate": "latin2_general_ci"})
|
|
databases.append({"name": self.dbname2})
|
|
|
|
assert_raises(exceptions.BadRequest, self.dbaas.databases.create,
|
|
instance_info.id, databases)
|
|
assert_equal(400, self.dbaas.last_http_code)
|
|
|
|
@test
|
|
def test_create_database_list_system(self):
|
|
# Databases that should not be returned in the list
|
|
databases = self.dbaas.databases.list(instance_info.id)
|
|
assert_equal(200, self.dbaas.last_http_code)
|
|
found = False
|
|
for db in self.system_dbs:
|
|
found = any(result.name == db for result in databases)
|
|
msg = "Database '%s' SHOULD NOT be found in result" % db
|
|
assert_false(found, msg)
|
|
found = False
|
|
|
|
@test
|
|
def test_create_database_on_missing_instance(self):
|
|
databases = [{"name": "invalid_db", "character_set": "latin2",
|
|
"collate": "latin2_general_ci"}]
|
|
assert_raises(exceptions.NotFound, self.dbaas.databases.create,
|
|
-1, databases)
|
|
assert_equal(404, self.dbaas.last_http_code)
|
|
|
|
@test(runs_after=[test_create_database])
|
|
def test_delete_database(self):
|
|
self.dbaas.databases.delete(instance_info.id, self.dbname_urlencoded)
|
|
assert_equal(202, self.dbaas.last_http_code)
|
|
if not FAKE:
|
|
time.sleep(5)
|
|
dbs = self.dbaas.databases.list(instance_info.id)
|
|
assert_equal(200, self.dbaas.last_http_code)
|
|
found = any(result.name == self.dbname_urlencoded for result in dbs)
|
|
assert_false(found, "Database '%s' SHOULD NOT be found in result" %
|
|
self.dbname_urlencoded)
|
|
|
|
@test(runs_after=[test_delete_database])
|
|
def test_cannot_delete_taboo_database_names(self):
|
|
for name in self.system_dbs:
|
|
assert_raises(exceptions.BadRequest, self.dbaas.databases.delete,
|
|
instance_info.id, name)
|
|
assert_equal(400, self.dbaas.last_http_code)
|
|
|
|
@test(runs_after=[test_delete_database])
|
|
def test_delete_database_on_missing_instance(self):
|
|
assert_raises(exceptions.NotFound, self.dbaas.databases.delete,
|
|
-1, self.dbname_urlencoded)
|
|
assert_equal(404, self.dbaas.last_http_code)
|
|
|
|
@test
|
|
def test_database_name_too_long(self):
|
|
databases = []
|
|
name = ("aasdlkhaglkjhakjdkjgfakjgadgfkajsg"
|
|
"34523dfkljgasldkjfglkjadsgflkjagsdd")
|
|
databases.append({"name": name})
|
|
assert_raises(exceptions.BadRequest, self.dbaas.databases.create,
|
|
instance_info.id, databases)
|
|
assert_equal(400, self.dbaas.last_http_code)
|
|
|
|
@test
|
|
def test_invalid_database_name(self):
|
|
databases = []
|
|
databases.append({"name": "sdfsd,"})
|
|
assert_raises(exceptions.BadRequest, self.dbaas.databases.create,
|
|
instance_info.id, databases)
|
|
assert_equal(400, self.dbaas.last_http_code)
|
|
|
|
@test
|
|
def test_pagination(self):
|
|
databases = []
|
|
databases.append({"name": "Sprockets", "character_set": "latin2",
|
|
"collate": "latin2_general_ci"})
|
|
databases.append({"name": "Cogs"})
|
|
databases.append({"name": "Widgets"})
|
|
|
|
self.dbaas.databases.create(instance_info.id, databases)
|
|
assert_equal(202, self.dbaas.last_http_code)
|
|
if not FAKE:
|
|
time.sleep(5)
|
|
limit = 2
|
|
databases = self.dbaas.databases.list(instance_info.id, limit=limit)
|
|
assert_equal(200, self.dbaas.last_http_code)
|
|
marker = databases.next
|
|
|
|
# Better get only as many as we asked for
|
|
assert_true(len(databases) <= limit)
|
|
assert_true(databases.next is not None)
|
|
assert_equal(marker, databases[-1].name)
|
|
marker = databases.next
|
|
|
|
# I better get new databases if I use the marker I was handed.
|
|
databases = self.dbaas.databases.list(instance_info.id, limit=limit,
|
|
marker=marker)
|
|
assert_equal(200, self.dbaas.last_http_code)
|
|
assert_true(marker not in [database.name for database in databases])
|
|
|
|
# Now fetch again with a larger limit.
|
|
databases = self.dbaas.databases.list(instance_info.id)
|
|
assert_equal(200, self.dbaas.last_http_code)
|
|
assert_true(databases.next is None)
|