1900fca535
fixed up the instance create and user/database level tests fixed up the resize up test fixed up and able to run through the entire suite of tests fixed issues with the guest not writing the my.cnf file correctly Fixes Bug #1095929 Change-Id: I2f971a072728380f83d82fdcb7595604a78a0511
223 lines
8.4 KiB
Python
223 lines
8.4 KiB
Python
# Copyright 2011 OpenStack LLC
|
|
#
|
|
# 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 reddwarfclient import exceptions
|
|
|
|
from proboscis import before_class
|
|
from proboscis import test
|
|
from proboscis.asserts import assert_equal
|
|
from proboscis.asserts import assert_false
|
|
from proboscis.asserts import assert_not_equal
|
|
from proboscis.asserts import assert_raises
|
|
from proboscis.asserts import assert_true
|
|
from proboscis.asserts import fail
|
|
from proboscis.decorators import expect_exception
|
|
from proboscis.decorators import time_out
|
|
|
|
from reddwarf import tests
|
|
|
|
from reddwarf.tests import util
|
|
from reddwarf.tests.api.instances import GROUP_START
|
|
from reddwarf.tests.api.instances import instance_info
|
|
from reddwarf.tests.util import test_config
|
|
|
|
GROUP = "dbaas.api.databases"
|
|
FAKE = test_config.values['fake_mode']
|
|
|
|
|
|
@test(depends_on_groups=[GROUP_START],
|
|
groups=[tests.INSTANCES, "dbaas.guest.mysql"],
|
|
enabled=not test_config.values['fake_mode'])
|
|
class TestMysqlAccess(object):
|
|
"""
|
|
Make sure that MySQL server was secured.
|
|
"""
|
|
|
|
@time_out(60 * 2)
|
|
@test
|
|
def test_mysql_admin(self):
|
|
"""Ensure we aren't allowed access with os_admin and wrong password."""
|
|
util.mysql_connection().assert_fails(
|
|
"os_admin", "asdfd-asdf234", instance_info.get_address())
|
|
|
|
@test
|
|
def test_mysql_root(self):
|
|
"""Ensure we aren't allowed access with root and wrong password."""
|
|
util.mysql_connection().assert_fails(
|
|
"root", "dsfgnear", instance_info.get_address())
|
|
|
|
|
|
@test(depends_on_groups=[GROUP_START],
|
|
depends_on_classes=[TestMysqlAccess],
|
|
groups=[tests.DBAAS_API, GROUP, tests.INSTANCES])
|
|
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, "charset": "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, "charset": "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, "charset": "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", "charset": "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", "charset": "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)
|