Merge "Added more unit-tests to improve code coverage"
This commit is contained in:
Jenkins
@@ -8,6 +8,7 @@ oslosphinx>=2.5.0 # Apache-2.0
|
||||
requests-mock>=0.6.0 # Apache-2.0
|
||||
sphinx>=1.1.2,!=1.2.0,!=1.3b1,<1.3
|
||||
testrepository>=0.0.18
|
||||
testscenarios>=0.4
|
||||
testtools>=0.9.36,!=1.2.0
|
||||
mock>=1.0
|
||||
httplib2>=0.7.5
|
||||
|
||||
517
troveclient/tests/fakes.py
Normal file
517
troveclient/tests/fakes.py
Normal file
@@ -0,0 +1,517 @@
|
||||
# Copyright [2015] Hewlett-Packard Development Company, L.P.
|
||||
#
|
||||
# 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.
|
||||
|
||||
from six.moves.urllib import parse
|
||||
from troveclient import client as base_client
|
||||
from troveclient.tests import utils
|
||||
from troveclient.v1 import client
|
||||
|
||||
|
||||
def get_version_map():
|
||||
return {
|
||||
'1.0': 'troveclient.tests.fakes.FakeClient',
|
||||
}
|
||||
|
||||
|
||||
def assert_has_keys(dict, required=[], optional=[]):
|
||||
keys = dict.keys()
|
||||
for k in required:
|
||||
try:
|
||||
assert k in keys
|
||||
except AssertionError:
|
||||
raise AssertionError("key: %s not found." % k)
|
||||
|
||||
|
||||
class FakeClient(client.Client):
|
||||
|
||||
def __init__(self, *args, **kwargs):
|
||||
client.Client.__init__(self, 'username', 'password',
|
||||
'project_id', 'auth_url',
|
||||
extensions=kwargs.get('extensions'))
|
||||
self.client = FakeHTTPClient(**kwargs)
|
||||
|
||||
def assert_called(self, method, url, body=None, pos=-1):
|
||||
"""Assert than an API method was just called."""
|
||||
expected = (method, url)
|
||||
called = self.client.callstack[pos][0:2]
|
||||
|
||||
assert self.client.callstack, \
|
||||
"Expected %s %s but no calls were made." % expected
|
||||
|
||||
assert expected == called, \
|
||||
'Expected %s %s; got %s %s' % (expected + called)
|
||||
|
||||
if body is not None:
|
||||
if self.client.callstack[pos][2] != body:
|
||||
raise AssertionError('%r != %r' %
|
||||
(self.client.callstack[pos][2], body))
|
||||
|
||||
def assert_called_anytime(self, method, url, body=None):
|
||||
"""Assert than an API method was called anytime in the test."""
|
||||
expected = (method, url)
|
||||
|
||||
assert self.client.callstack, \
|
||||
"Expected %s %s but no calls were made." % expected
|
||||
|
||||
found = False
|
||||
for entry in self.client.callstack:
|
||||
if expected == entry[0:2]:
|
||||
found = True
|
||||
break
|
||||
|
||||
assert found, 'Expected %s; got %s' % (expected, self.client.callstack)
|
||||
if body is not None:
|
||||
try:
|
||||
assert entry[2] == body
|
||||
except AssertionError:
|
||||
print(entry[2])
|
||||
print("!=")
|
||||
print(body)
|
||||
raise
|
||||
|
||||
self.client.callstack = []
|
||||
|
||||
|
||||
class FakeHTTPClient(base_client.HTTPClient):
|
||||
|
||||
def __init__(self, **kwargs):
|
||||
self.username = 'username'
|
||||
self.password = 'password'
|
||||
self.auth_url = 'auth_url'
|
||||
self.management_url = (
|
||||
'http://trove-api:8779/v1.0/14630bc0e9ef4e248c9753eaf57b0f6e')
|
||||
self.tenant_id = 'tenant_id'
|
||||
self.callstack = []
|
||||
self.projectid = 'projectid'
|
||||
self.user = 'user'
|
||||
self.region_name = 'region_name'
|
||||
self.endpoint_type = 'endpoint_type'
|
||||
self.service_type = 'service_type'
|
||||
self.service_name = 'service_name'
|
||||
self.volume_service_name = 'volume_service_name'
|
||||
self.timings = 'timings'
|
||||
self.bypass_url = 'bypass_url'
|
||||
self.os_cache = 'os_cache'
|
||||
self.http_log_debug = 'http_log_debug'
|
||||
|
||||
def _cs_request(self, url, method, **kwargs):
|
||||
# Check that certain things are called correctly
|
||||
if method in ['GET', 'DELETE']:
|
||||
assert 'body' not in kwargs
|
||||
elif method == 'PUT':
|
||||
assert 'body' in kwargs
|
||||
|
||||
if url is not None:
|
||||
# Call the method
|
||||
args = parse.parse_qsl(parse.urlparse(url)[4])
|
||||
kwargs.update(args)
|
||||
munged_url = url.rsplit('?', 1)[0]
|
||||
munged_url = munged_url.strip('/').replace('/', '_')
|
||||
munged_url = munged_url.replace('.', '_')
|
||||
munged_url = munged_url.replace('-', '_')
|
||||
munged_url = munged_url.replace(' ', '_')
|
||||
callback = "%s_%s" % (method.lower(), munged_url)
|
||||
|
||||
if not hasattr(self, callback):
|
||||
raise AssertionError('Called unknown API method: %s %s, '
|
||||
'expected fakes method name: %s' %
|
||||
(method, url, callback))
|
||||
|
||||
# Note the call
|
||||
self.callstack.append((method, url, kwargs.get('body')))
|
||||
|
||||
status, headers, body = getattr(self, callback)(**kwargs)
|
||||
r = utils.TestResponse({
|
||||
"status_code": status,
|
||||
"text": body,
|
||||
"headers": headers,
|
||||
})
|
||||
return r, body
|
||||
|
||||
def get_instances(self, **kw):
|
||||
return (200, {}, {"instances": [
|
||||
{
|
||||
"id": "1234",
|
||||
"name": "test-member-1",
|
||||
"status": "ACTIVE",
|
||||
"ip": ["10.0.0.13"],
|
||||
"volume": {"size": 2},
|
||||
"flavor": {"id": "2"},
|
||||
"datastore": {"version": "5.6", "type": "mysql"}},
|
||||
{
|
||||
"id": "5678",
|
||||
"name": "test-member-2",
|
||||
"status": "ACTIVE",
|
||||
"ip": ["10.0.0.14"],
|
||||
"volume": {"size": 2},
|
||||
"flavor": {"id": "2"},
|
||||
"datastore": {"version": "5.6", "type": "mysql"}}]})
|
||||
|
||||
def get_instances_1234(self, **kw):
|
||||
r = {'instance': self.get_instances()[2]['instances'][0]}
|
||||
return (200, {}, r)
|
||||
|
||||
def post_instances(self, body, **kw):
|
||||
assert_has_keys(
|
||||
body['instance'],
|
||||
required=['name', 'flavorRef'],
|
||||
optional=['volume'])
|
||||
if 'volume' in body['instance']:
|
||||
assert_has_keys(body['instance']['volume'], required=['size'])
|
||||
return (202, {}, self.get_instances_1234()[2])
|
||||
|
||||
def get_flavors(self, **kw):
|
||||
return (200, {}, {"flavors": [
|
||||
{
|
||||
"str_id": "1",
|
||||
"ram": 512,
|
||||
"id": 1,
|
||||
"name": "m1.tiny"},
|
||||
{
|
||||
"str_id": "10",
|
||||
"ram": 768,
|
||||
"id": 10,
|
||||
"name": "eph.rd-smaller"},
|
||||
{
|
||||
"str_id": "2",
|
||||
"ram": 2048,
|
||||
"id": 2,
|
||||
"name": "m1.small"},
|
||||
{
|
||||
"str_id": "3",
|
||||
"ram": 4096,
|
||||
"id": 3,
|
||||
"name": "m1.medium"}]})
|
||||
|
||||
def get_datastores_mysql_versions_some_version_id_flavors(self, **kw):
|
||||
return self.get_flavors()
|
||||
|
||||
def get_flavors_1(self, **kw):
|
||||
r = {'flavor': self.get_flavors()[2]['flavors'][0]}
|
||||
return (200, {}, r)
|
||||
|
||||
def get_clusters(self, **kw):
|
||||
return (200, {}, {"clusters": [
|
||||
{
|
||||
"instances": [
|
||||
{
|
||||
"type": "member",
|
||||
"id": "member-1",
|
||||
"ip": ["10.0.0.3"],
|
||||
"flavor": {"id": "2"},
|
||||
"name": "test-clstr-member-1"
|
||||
},
|
||||
{
|
||||
"type": "member",
|
||||
"id": "member-2",
|
||||
"ip": ["10.0.0.4"],
|
||||
"flavor": {"id": "2"},
|
||||
"name": "test-clstr-member-2"
|
||||
}],
|
||||
"updated": "2015-05-02T11:06:19",
|
||||
"task": {"description": "No tasks for the cluster.", "id": 1,
|
||||
"name": "NONE"},
|
||||
"name": "test-clstr",
|
||||
"created": "2015-05-02T10:37:04",
|
||||
"datastore": {"version": "7.1", "type": "vertica"},
|
||||
"id": "cls-1234"}]})
|
||||
|
||||
def get_clusters_cls_1234(self, **kw):
|
||||
r = {'cluster': self.get_clusters()[2]['clusters'][0]}
|
||||
return (200, {}, r)
|
||||
|
||||
def delete_instances_1234(self, **kw):
|
||||
return (202, {}, None)
|
||||
|
||||
def delete_clusters_cls_1234(self, **kw):
|
||||
return (202, {}, None)
|
||||
|
||||
def patch_instances_1234(self, **kw):
|
||||
return (202, {}, None)
|
||||
|
||||
def post_clusters(self, body, **kw):
|
||||
assert_has_keys(
|
||||
body['cluster'],
|
||||
required=['instances', 'datastore', 'name'])
|
||||
if 'instances' in body['cluster']:
|
||||
for instance in body['cluster']['instances']:
|
||||
assert_has_keys(instance, required=['volume', 'flavorRef'])
|
||||
return (202, {}, self.get_clusters_cls_1234()[2])
|
||||
|
||||
def post_instances_1234_action(self, **kw):
|
||||
return (202, {}, None)
|
||||
|
||||
def get_datastores(self, **kw):
|
||||
return (200, {}, {"datastores": [
|
||||
{
|
||||
"default_version": "v-56",
|
||||
"versions": [{"id": "v-56", "name": "5.6"}],
|
||||
"id": "d-123",
|
||||
"name": "mysql"},
|
||||
{
|
||||
"default_version": "v-71",
|
||||
"versions": [{"id": "v-71", "name": "7.1"}],
|
||||
"id": "d-456",
|
||||
"name": "vertica"
|
||||
}]})
|
||||
|
||||
def get_datastores_d_123(self, **kw):
|
||||
r = {'datastore': self.get_datastores()[2]['datastores'][0]}
|
||||
return (200, {}, r)
|
||||
|
||||
def get_datastores_d_123_versions(self, **kw):
|
||||
return (200, {}, {"versions": [
|
||||
{
|
||||
"datastore": "d-123",
|
||||
"id": "v-56",
|
||||
"name": "5.6"}]})
|
||||
|
||||
def get_datastores_d_123_versions_v_56(self, **kw):
|
||||
r = {'version': self.get_datastores_d_123_versions()[2]['versions'][0]}
|
||||
return (200, {}, r)
|
||||
|
||||
def get_configurations(self, **kw):
|
||||
return (200, {}, {"configurations": [
|
||||
{
|
||||
"datastore_name": "mysql",
|
||||
"updated": "2015-05-16T10:24:29",
|
||||
"name": "test_config",
|
||||
"created": "2015-05-16T10:24:28",
|
||||
"datastore_version_name": "5.6",
|
||||
"id": "c-123",
|
||||
"values": {"max_connections": 5},
|
||||
"datastore_version_id": "d-123", "description": ''}]})
|
||||
|
||||
def get_configurations_c_123(self, **kw):
|
||||
r = {'configuration': self.get_configurations()[2]['configurations'][0]
|
||||
}
|
||||
return (200, {}, r)
|
||||
|
||||
def get_datastores_d_123_versions_v_156_parameters(self, **kw):
|
||||
return (200, {}, {"configuration-parameters": [
|
||||
{
|
||||
"type": "string",
|
||||
"name": "character_set_results",
|
||||
"datastore_version_id": "d-123",
|
||||
"restart_required": "false"},
|
||||
{
|
||||
"name": "connect_timeout",
|
||||
"min": 2,
|
||||
"max": 31536000,
|
||||
"restart_required": "false",
|
||||
"type": "integer",
|
||||
"datastore_version_id": "d-123"},
|
||||
{
|
||||
"type": "string",
|
||||
"name": "character_set_client",
|
||||
"datastore_version_id": "d-123",
|
||||
"restart_required": "false"},
|
||||
{
|
||||
"name": "max_connections",
|
||||
"min": 1,
|
||||
"max": 100000,
|
||||
"restart_required": "false",
|
||||
"type": "integer",
|
||||
"datastore_version_id": "d-123"}]})
|
||||
|
||||
def get_datastores_d_123_versions_v_56_parameters_max_connections(self,
|
||||
**kw):
|
||||
r = self.get_datastores_d_123_versions_v_156_parameters()[
|
||||
2]['configuration-parameters'][3]
|
||||
return (200, {}, r)
|
||||
|
||||
def get_configurations_c_123_instances(self, **kw):
|
||||
return (200, {}, {"instances": []})
|
||||
|
||||
def delete_configurations_c_123(self, **kw):
|
||||
return (202, {}, None)
|
||||
|
||||
def get_instances_1234_configuration(self, **kw):
|
||||
return (200, {}, {"instance": {"configuration": {
|
||||
"tmp_table_size": "15M",
|
||||
"innodb_log_files_in_group": "2",
|
||||
"skip-external-locking": "1",
|
||||
"max_user_connections": "98"}}})
|
||||
|
||||
def put_instances_1234(self, **kw):
|
||||
return (202, {}, None)
|
||||
|
||||
def patch_instances_1234_metadata_key_123(self, **kw):
|
||||
return (202, {}, None)
|
||||
|
||||
def put_instances_1234_metadata_key_123(self, **kw):
|
||||
return (202, {}, None)
|
||||
|
||||
def delete_instances_1234_metadata_key_123(self, **kw):
|
||||
return (202, {}, None)
|
||||
|
||||
def post_instances_1234_metadata_key123(self, body, **kw):
|
||||
return (202, {}, {'metadata': {}})
|
||||
|
||||
def get_instances_1234_metadata(self, **kw):
|
||||
return (200, {}, {"metadata": {}})
|
||||
|
||||
def get_instances_1234_metadata_key123(self, **kw):
|
||||
return (200, {}, {"metadata": {}})
|
||||
|
||||
def get_limits(self, **kw):
|
||||
return (200, {}, {"limits": [
|
||||
{
|
||||
"max_backups": 50,
|
||||
"verb": "ABSOLUTE",
|
||||
"max_volumes": 20,
|
||||
"max_instances": 5}]})
|
||||
|
||||
def get_backups(self, **kw):
|
||||
return (200, {}, {"backups": [
|
||||
{
|
||||
"status": "COMPLETED",
|
||||
"updated": "2015-05-16T14:23:08",
|
||||
"description": None,
|
||||
"datastore": {"version": "5.6", "type": "mysql",
|
||||
"version_id": "v-56"},
|
||||
"id": "bk-1234",
|
||||
"size": 0.11,
|
||||
"name": "bkp_1",
|
||||
"created": "2015-05-16T14:22:28",
|
||||
"instance_id": "1234",
|
||||
"parent_id": None,
|
||||
"locationRef": ("http://backup_srvr/database_backups/"
|
||||
"bk-1234.xbstream.gz.enc")},
|
||||
{
|
||||
"status": "COMPLETED",
|
||||
"updated": "2015-05-16T14:22:12",
|
||||
"description": None,
|
||||
"datastore": {"version": "5.6", "type": "mysql",
|
||||
"version_id": "v-56"},
|
||||
"id": "bk-5678",
|
||||
"size": 0.11,
|
||||
"name": "test_bkp",
|
||||
"created": "2015-05-16T14:21:27",
|
||||
"instance_id": "5678",
|
||||
"parent_id": None,
|
||||
"locationRef": ("http://backup_srvr/database_backups/"
|
||||
"bk-5678.xbstream.gz.enc")}]})
|
||||
|
||||
def get_backups_bk_1234(self, **kw):
|
||||
r = {'backup': self.get_backups()[2]['backups'][0]}
|
||||
return (200, {}, r)
|
||||
|
||||
def get_instances_1234_backups(self, **kw):
|
||||
r = {'backups': [self.get_backups()[2]['backups'][0]]}
|
||||
return (200, {}, r)
|
||||
|
||||
def delete_backups_bk_1234(self, **kw):
|
||||
return (202, {}, None)
|
||||
|
||||
def post_backups(self, body, **kw):
|
||||
assert_has_keys(
|
||||
body['backup'],
|
||||
required=['name'],
|
||||
optional=['description', 'parent'])
|
||||
return (202, {}, self.get_backups_bk_1234()[2])
|
||||
|
||||
def get_instances_1234_databases(self, **kw):
|
||||
return (200, {}, {"databases": [
|
||||
{"name": "db_1"},
|
||||
{"name": "db_2"},
|
||||
{"name": "performance_schema"}]})
|
||||
|
||||
def delete_instances_1234_databases_db_1(self, **kw):
|
||||
return (202, {}, None)
|
||||
|
||||
def post_instances_1234_databases(self, body, **kw):
|
||||
assert_has_keys(
|
||||
body,
|
||||
required=['databases'])
|
||||
for database in body['databases']:
|
||||
assert_has_keys(database, required=['name'],
|
||||
optional=['character_set', 'collate'])
|
||||
return (202, {},
|
||||
self.get_instances_1234_databases()[2]['databases'][0])
|
||||
|
||||
def get_instances_1234_users(self, **kw):
|
||||
return (200, {}, {"users": [
|
||||
{"host": "%", "name": "jacob", "databases": []},
|
||||
{"host": "%", "name": "rocky", "databases": []},
|
||||
{"host": "%", "name": "harry", "databases": [{"name": "db1"}]}]})
|
||||
|
||||
def get_instances_1234_users_jacob(self, **kw):
|
||||
r = {'user': self.get_instances_1234_users()[2]['users'][0]}
|
||||
return (200, {}, r)
|
||||
|
||||
def delete_instances_1234_users_jacob(self, **kw):
|
||||
return (202, {}, None)
|
||||
|
||||
def post_instances_1234_users(self, body, **kw):
|
||||
assert_has_keys(
|
||||
body,
|
||||
required=['users'])
|
||||
for database in body['users']:
|
||||
assert_has_keys(database, required=['name', 'password'],
|
||||
optional=['databases'])
|
||||
return (202, {}, self.get_instances_1234_users()[2]['users'][0])
|
||||
|
||||
def get_instances_1234_users_jacob_databases(self, **kw):
|
||||
r = {'databases': [
|
||||
self.get_instances_1234_databases()[2]['databases'][0],
|
||||
self.get_instances_1234_databases()[2]['databases'][1]]}
|
||||
return (200, {}, r)
|
||||
|
||||
def put_instances_1234_users_jacob(self, **kw):
|
||||
return (202, {}, None)
|
||||
|
||||
def put_instances_1234_users_jacob_databases(self, **kw):
|
||||
return (202, {}, None)
|
||||
|
||||
def delete_instances_1234_users_jacob_databases_db1(self, **kw):
|
||||
return (202, {}, None)
|
||||
|
||||
def post_instances_1234_root(self, **kw):
|
||||
return (202, {}, {"user": {"password": "password", "name": "root"}})
|
||||
|
||||
def get_instances_1234_root(self, **kw):
|
||||
return (200, {}, {"rootEnabled": 'True'})
|
||||
|
||||
def get_security_groups(self, **kw):
|
||||
return (200, {}, {"security_groups": [
|
||||
{
|
||||
"instance_id": "1234",
|
||||
"updated": "2015-05-16T17:29:45",
|
||||
"name": "SecGroup_1234",
|
||||
"created": "2015-05-16T17:29:45",
|
||||
"rules": [{"to_port": 3306, "cidr": "0.0.0.0/0",
|
||||
"from_port": 3306,
|
||||
"protocol": "tcp", "id": "1"}],
|
||||
"id": "2",
|
||||
"description": "Security Group for 1234"}]})
|
||||
|
||||
def get_security_groups_2(self, **kw):
|
||||
r = {'security_group': self.get_security_groups()[
|
||||
2]['security_groups'][0]}
|
||||
return (200, {}, r)
|
||||
|
||||
def delete_security_group_rules_2(self, **kw):
|
||||
return (202, {}, None)
|
||||
|
||||
def post_security_group_rules(self, body, **kw):
|
||||
assert_has_keys(body['security_group_rule'], required=['cidr', 'cidr'])
|
||||
return (202, {}, {"security_group_rule": [
|
||||
{
|
||||
"from_port": 3306,
|
||||
"protocol": "tcp",
|
||||
"created": "2015-05-16T17:55:05",
|
||||
"to_port": 3306,
|
||||
"security_group_id": "2",
|
||||
"cidr": "15.0.0.0/24", "id": 3}]})
|
||||
@@ -15,13 +15,18 @@
|
||||
# License for the specific language governing permissions and limitations
|
||||
# under the License.
|
||||
|
||||
import fixtures
|
||||
from keystoneclient import adapter
|
||||
import logging
|
||||
import mock
|
||||
import requests
|
||||
import testtools
|
||||
|
||||
import troveclient.v1.client
|
||||
|
||||
from troveclient import client as other_client
|
||||
from troveclient import exceptions
|
||||
from troveclient.openstack.common.apiclient import client
|
||||
from troveclient.openstack.common.apiclient import exceptions
|
||||
from troveclient import service_catalog
|
||||
import troveclient.v1.client
|
||||
|
||||
|
||||
class ClientTest(testtools.TestCase):
|
||||
@@ -37,3 +42,446 @@ class ClientTest(testtools.TestCase):
|
||||
self.assertRaises(exceptions.UnsupportedVersion,
|
||||
client.BaseClient.get_class, 'database',
|
||||
'0', version_map)
|
||||
|
||||
def test_client_with_auth_system_without_auth_plugin(self):
|
||||
self.assertRaisesRegexp(
|
||||
exceptions.AuthSystemNotFound, "AuthSystemNotFound: 'something'",
|
||||
other_client.HTTPClient, user='user', password='password',
|
||||
projectid='project', timeout=2, auth_url="http://www.blah.com",
|
||||
auth_system='something')
|
||||
|
||||
def test_client_with_auth_system_without_endpoint(self):
|
||||
auth_plugin = mock.Mock()
|
||||
auth_plugin.get_auth_url = mock.Mock(return_value=None)
|
||||
self.assertRaises(
|
||||
exceptions.EndpointNotFound,
|
||||
other_client.HTTPClient, user='user', password='password',
|
||||
projectid='project', timeout=2, auth_plugin=auth_plugin,
|
||||
auth_url=None, auth_system='something')
|
||||
|
||||
def test_client_with_timeout(self):
|
||||
instance = other_client.HTTPClient(user='user',
|
||||
password='password',
|
||||
projectid='project',
|
||||
timeout=2,
|
||||
auth_url="http://www.blah.com",
|
||||
insecure=True)
|
||||
self.assertEqual(2, instance.timeout)
|
||||
mock_request = mock.Mock()
|
||||
mock_request.return_value = requests.Response()
|
||||
mock_request.return_value.status_code = 200
|
||||
mock_request.return_value.headers = {
|
||||
'x-server-management-url': 'blah.com',
|
||||
'x-auth-token': 'blah',
|
||||
}
|
||||
with mock.patch('requests.request', mock_request):
|
||||
instance.authenticate()
|
||||
requests.request.assert_called_with(
|
||||
mock.ANY, mock.ANY, timeout=2, headers=mock.ANY,
|
||||
verify=mock.ANY)
|
||||
|
||||
def test_client_unauthorized(self):
|
||||
instance = other_client.HTTPClient(user='user',
|
||||
password='password',
|
||||
projectid='project',
|
||||
timeout=2,
|
||||
auth_url="http://www.blah.com",
|
||||
cacert=mock.Mock())
|
||||
instance.auth_token = 'foobar'
|
||||
instance.management_url = 'http://example.com'
|
||||
instance.get_service_url = mock.Mock(return_value='http://example.com')
|
||||
instance.version = 'v2.0'
|
||||
mock_request = mock.Mock()
|
||||
mock_request.side_effect = other_client.exceptions.Unauthorized(401)
|
||||
with mock.patch('requests.request', mock_request):
|
||||
self.assertRaises(
|
||||
exceptions.Unauthorized, instance.get, '/instances')
|
||||
|
||||
def test_client_bad_request(self):
|
||||
instance = other_client.HTTPClient(user='user',
|
||||
password='password',
|
||||
projectid='project',
|
||||
timeout=2,
|
||||
auth_url="http://www.blah.com")
|
||||
instance.auth_token = 'foobar'
|
||||
instance.management_url = 'http://example.com'
|
||||
instance.get_service_url = mock.Mock(return_value='http://example.com')
|
||||
instance.version = 'v2.0'
|
||||
mock_request = mock.Mock()
|
||||
mock_request.side_effect = other_client.exceptions.BadRequest()
|
||||
with mock.patch('requests.request', mock_request):
|
||||
self.assertRaises(
|
||||
exceptions.BadRequest, instance.get, '/instances')
|
||||
|
||||
def test_client_with_client_exception(self):
|
||||
instance = other_client.HTTPClient(user='user',
|
||||
password='password',
|
||||
projectid='project',
|
||||
timeout=2,
|
||||
auth_url="http://www.blah.com",
|
||||
retries=2)
|
||||
instance.auth_token = 'foobar'
|
||||
instance.management_url = 'http://example.com'
|
||||
instance.get_service_url = mock.Mock(return_value='http://example.com')
|
||||
instance.version = 'v2.0'
|
||||
mock_request = mock.Mock()
|
||||
mock_request.side_effect = other_client.exceptions.ClientException()
|
||||
type(mock_request.side_effect).code = mock.PropertyMock(
|
||||
side_effect=[501, 111])
|
||||
with mock.patch('requests.request', mock_request):
|
||||
self.assertRaises(
|
||||
exceptions.ClientException, instance.get, '/instances')
|
||||
|
||||
def test_client_connection_error(self):
|
||||
instance = other_client.HTTPClient(user='user',
|
||||
password='password',
|
||||
projectid='project',
|
||||
timeout=2,
|
||||
auth_url="http://www.blah.com",
|
||||
retries=2)
|
||||
instance.auth_token = 'foobar'
|
||||
instance.management_url = 'http://example.com'
|
||||
instance.get_service_url = mock.Mock(return_value='http://example.com')
|
||||
instance.version = 'v2.0'
|
||||
mock_request = mock.Mock()
|
||||
mock_request.side_effect = requests.exceptions.ConnectionError(
|
||||
'connection refused')
|
||||
with mock.patch('requests.request', mock_request):
|
||||
self.assertRaisesRegexp(
|
||||
exceptions.ClientException,
|
||||
'Unable to establish connection: connection refused',
|
||||
instance.get, '/instances')
|
||||
|
||||
@mock.patch.object(other_client.HTTPClient, 'request',
|
||||
return_value=(200, "{'versions':[]}"))
|
||||
def _check_version_url(self, management_url, version_url, mock_request):
|
||||
projectid = '25e469aa1848471b875e68cde6531bc5'
|
||||
instance = other_client.HTTPClient(user='user',
|
||||
password='password',
|
||||
projectid=projectid,
|
||||
auth_url="http://www.blah.com")
|
||||
instance.auth_token = 'foobar'
|
||||
instance.management_url = management_url % projectid
|
||||
mock_get_service_url = mock.Mock(return_value=instance.management_url)
|
||||
instance.get_service_url = mock_get_service_url
|
||||
instance.version = 'v2.0'
|
||||
|
||||
# If passing None as the part of url, a client accesses the url which
|
||||
# doesn't include "v2/<projectid>" for getting API version info.
|
||||
instance.get('')
|
||||
mock_request.assert_called_once_with(instance.management_url, 'GET',
|
||||
headers=mock.ANY)
|
||||
mock_request.reset_mock()
|
||||
|
||||
# Otherwise, a client accesses the url which includes "v2/<projectid>".
|
||||
instance.get('/instances')
|
||||
url = instance.management_url + '/instances'
|
||||
mock_request.assert_called_once_with(url, 'GET', headers=mock.ANY)
|
||||
|
||||
def test_client_version_url(self):
|
||||
self._check_version_url('http://foo.com/v1/%s', 'http://foo.com/')
|
||||
|
||||
def test_client_version_url_with_tenant_name(self):
|
||||
self._check_version_url('http://foo.com/trove/v1/%s',
|
||||
'http://foo.com/trove/')
|
||||
|
||||
def test_log_req(self):
|
||||
self.logger = self.useFixture(
|
||||
fixtures.FakeLogger(
|
||||
format="%(message)s",
|
||||
level=logging.DEBUG,
|
||||
nuke_handlers=True
|
||||
)
|
||||
)
|
||||
cs = other_client.HTTPClient(user='user',
|
||||
password='password',
|
||||
projectid=None,
|
||||
auth_url="http://www.blah.com",
|
||||
http_log_debug=True)
|
||||
cs.http_log_req(('/foo', 'GET'), {'headers': {}})
|
||||
cs.http_log_req(('/foo', 'GET'),
|
||||
{'headers': {'X-Auth-Token': 'totally_bogus'}})
|
||||
cs.http_log_req(
|
||||
('/foo', 'GET'),
|
||||
{'headers': {},
|
||||
'data': '{"auth": {"passwordCredentials": '
|
||||
'{"password": "password"}}}'})
|
||||
|
||||
output = self.logger.output.split('\n')
|
||||
|
||||
self.assertIn("REQ: curl -i /foo -X GET", output)
|
||||
self.assertIn(
|
||||
"REQ: curl -i /foo -X GET -H "
|
||||
'"X-Auth-Token: totally_bogus"',
|
||||
output)
|
||||
self.assertIn(
|
||||
"REQ: curl -i /foo -X GET -d "
|
||||
'\'{"auth": {"passwordCredentials": {"password":'
|
||||
' "password"}}}\'',
|
||||
output)
|
||||
|
||||
@mock.patch.object(service_catalog, 'ServiceCatalog')
|
||||
def test_client_auth_token(self, mock_service_catalog):
|
||||
auth_url = 'http://www.blah.com'
|
||||
proxy_token = 'foobar'
|
||||
proxy_tenant_id = 'user'
|
||||
mock_service_catalog.return_value.get_token = mock.Mock(
|
||||
return_value=proxy_token)
|
||||
instance = other_client.HTTPClient(proxy_token=proxy_token,
|
||||
proxy_tenant_id=proxy_tenant_id,
|
||||
user=None,
|
||||
password=None,
|
||||
tenant_id=proxy_tenant_id,
|
||||
projectid=None,
|
||||
timeout=2,
|
||||
auth_url=auth_url)
|
||||
instance.management_url = 'http://example.com'
|
||||
instance.get_service_url = mock.Mock(return_value='http://example.com')
|
||||
instance.version = 'v2.0'
|
||||
mock_request = mock.Mock()
|
||||
mock_request.return_value = requests.Response()
|
||||
mock_request.return_value.status_code = 200
|
||||
mock_request.return_value.headers = {
|
||||
'x-server-management-url': 'blah.com',
|
||||
'x-auth-token': 'blah',
|
||||
}
|
||||
|
||||
with mock.patch('requests.request', mock_request):
|
||||
instance.authenticate()
|
||||
mock_request.assert_called_with(
|
||||
'GET', auth_url + '/tokens/foobar?belongsTo=user',
|
||||
headers={'User-Agent': 'python-troveclient',
|
||||
'Accept': 'application/json',
|
||||
'X-Auth-Token': proxy_token},
|
||||
timeout=2, verify=True)
|
||||
|
||||
@mock.patch.object(service_catalog, 'ServiceCatalog', side_effect=KeyError)
|
||||
def test_client_auth_token_authorization_failure(self,
|
||||
mock_service_catalog):
|
||||
auth_url = 'http://www.blah.com'
|
||||
proxy_token = 'foobar'
|
||||
proxy_tenant_id = 'user'
|
||||
mock_service_catalog.return_value.get_token = mock.Mock(
|
||||
return_value=proxy_token)
|
||||
instance = other_client.HTTPClient(proxy_token=proxy_token,
|
||||
proxy_tenant_id=proxy_tenant_id,
|
||||
user=None,
|
||||
password=None,
|
||||
tenant_id=proxy_tenant_id,
|
||||
projectid=None,
|
||||
timeout=2,
|
||||
auth_url=auth_url)
|
||||
instance.management_url = 'http://example.com'
|
||||
instance.get_service_url = mock.Mock(return_value='http://example.com')
|
||||
instance.version = 'v2.0'
|
||||
mock_request = mock.Mock()
|
||||
mock_request.return_value = requests.Response()
|
||||
mock_request.return_value.status_code = 200
|
||||
mock_request.return_value.headers = {
|
||||
'x-server-management-url': 'blah.com',
|
||||
'x-auth-token': 'blah',
|
||||
}
|
||||
|
||||
with mock.patch('requests.request', mock_request):
|
||||
self.assertRaises(exceptions.AuthorizationFailure,
|
||||
instance.authenticate)
|
||||
|
||||
@mock.patch.object(service_catalog, 'ServiceCatalog',
|
||||
side_effect=other_client.exceptions.EndpointNotFound)
|
||||
def test_client_auth_token_endpoint_not_found(self, mock_service_catalog):
|
||||
auth_url = 'http://www.blah.com'
|
||||
proxy_token = 'foobar'
|
||||
proxy_tenant_id = 'user'
|
||||
mock_service_catalog.return_value.get_token = mock.Mock(
|
||||
return_value=proxy_token)
|
||||
instance = other_client.HTTPClient(proxy_token=proxy_token,
|
||||
proxy_tenant_id=proxy_tenant_id,
|
||||
user=None,
|
||||
password=None,
|
||||
tenant_id=proxy_tenant_id,
|
||||
projectid=None,
|
||||
timeout=2,
|
||||
auth_url=auth_url)
|
||||
instance.management_url = 'http://example.com'
|
||||
instance.get_service_url = mock.Mock(return_value='http://example.com')
|
||||
instance.version = 'v2.0'
|
||||
mock_request = mock.Mock()
|
||||
mock_request.return_value = requests.Response()
|
||||
mock_request.return_value.status_code = 200
|
||||
mock_request.return_value.headers = {
|
||||
'x-server-management-url': 'blah.com',
|
||||
'x-auth-token': 'blah',
|
||||
}
|
||||
|
||||
with mock.patch('requests.request', mock_request):
|
||||
self.assertRaises(exceptions.EndpointNotFound,
|
||||
instance.authenticate)
|
||||
|
||||
@mock.patch.object(service_catalog, 'ServiceCatalog')
|
||||
def test_client_auth_token_v1_auth_failure(self, mock_service_catalog):
|
||||
auth_url = 'http://www.blah.com'
|
||||
proxy_token = 'foobar'
|
||||
proxy_tenant_id = 'user'
|
||||
mock_service_catalog.return_value.get_token = mock.Mock(
|
||||
return_value=proxy_token)
|
||||
instance = other_client.HTTPClient(proxy_token=proxy_token,
|
||||
proxy_tenant_id=proxy_tenant_id,
|
||||
user=None,
|
||||
password=None,
|
||||
tenant_id=proxy_tenant_id,
|
||||
projectid=None,
|
||||
timeout=2,
|
||||
auth_url=auth_url)
|
||||
instance.management_url = 'http://example.com'
|
||||
instance.get_service_url = mock.Mock(return_value='http://example.com')
|
||||
instance.version = 'v1.0'
|
||||
mock_request = mock.Mock()
|
||||
mock_request.return_value = requests.Response()
|
||||
mock_request.return_value.status_code = 200
|
||||
mock_request.return_value.headers = {
|
||||
'x-server-management-url': 'blah.com',
|
||||
'x-auth-token': 'blah',
|
||||
}
|
||||
|
||||
with mock.patch('requests.request', mock_request):
|
||||
self.assertRaises(exceptions.NoTokenLookupException,
|
||||
instance.authenticate)
|
||||
|
||||
@mock.patch.object(service_catalog, 'ServiceCatalog')
|
||||
def test_client_auth_token_v1_auth(self, mock_service_catalog):
|
||||
auth_url = 'http://www.blah.com'
|
||||
proxy_token = 'foobar'
|
||||
mock_service_catalog.return_value.get_token = mock.Mock(
|
||||
return_value=proxy_token)
|
||||
instance = other_client.HTTPClient(user='user',
|
||||
password='password',
|
||||
projectid='projectid',
|
||||
timeout=2,
|
||||
auth_url=auth_url)
|
||||
instance.management_url = 'http://example.com'
|
||||
instance.get_service_url = mock.Mock(return_value='http://example.com')
|
||||
instance.version = 'v1.0'
|
||||
mock_request = mock.Mock()
|
||||
mock_request.return_value = requests.Response()
|
||||
mock_request.return_value.status_code = 200
|
||||
mock_request.return_value.headers = {
|
||||
'x-server-management-url': 'blah.com',
|
||||
}
|
||||
headers = {'Content-Type': 'application/json',
|
||||
'Accept': 'application/json',
|
||||
'User-Agent': 'python-troveclient'}
|
||||
with mock.patch('requests.request', mock_request):
|
||||
instance.authenticate()
|
||||
called_args, called_kwargs = mock_request.call_args
|
||||
self.assertEqual(('POST', 'http://www.blah.com/v2.0/tokens'),
|
||||
called_args)
|
||||
self.assertDictEqual(headers, called_kwargs['headers'])
|
||||
|
||||
def test_client_get(self):
|
||||
auth_url = 'http://www.blah.com'
|
||||
instance = other_client.HTTPClient(user='user',
|
||||
password='password',
|
||||
projectid='project_id',
|
||||
timeout=2,
|
||||
auth_url=auth_url)
|
||||
instance._cs_request = mock.Mock()
|
||||
|
||||
instance.get('clusters')
|
||||
instance._cs_request.assert_called_with('clusters', 'GET')
|
||||
|
||||
def test_client_patch(self):
|
||||
auth_url = 'http://www.blah.com'
|
||||
body = mock.Mock()
|
||||
instance = other_client.HTTPClient(user='user',
|
||||
password='password',
|
||||
projectid='project_id',
|
||||
timeout=2,
|
||||
auth_url=auth_url)
|
||||
instance._cs_request = mock.Mock()
|
||||
|
||||
instance.patch('instances/dummy-instance-id', body=body)
|
||||
instance._cs_request.assert_called_with(
|
||||
'instances/dummy-instance-id', 'PATCH', body=body)
|
||||
|
||||
def test_client_post(self):
|
||||
auth_url = 'http://www.blah.com'
|
||||
body = {"add_shard": {}}
|
||||
instance = other_client.HTTPClient(user='user',
|
||||
password='password',
|
||||
projectid='project_id',
|
||||
timeout=2,
|
||||
auth_url=auth_url)
|
||||
instance._cs_request = mock.Mock()
|
||||
|
||||
instance.post('clusters/dummy-cluster-id', body=body)
|
||||
instance._cs_request.assert_called_with(
|
||||
'clusters/dummy-cluster-id', 'POST', body=body)
|
||||
|
||||
def test_client_put(self):
|
||||
auth_url = 'http://www.blah.com'
|
||||
body = {"user": {"password": "new_password"}}
|
||||
instance = other_client.HTTPClient(user='user',
|
||||
password='password',
|
||||
projectid='project_id',
|
||||
timeout=2,
|
||||
auth_url=auth_url)
|
||||
instance._cs_request = mock.Mock()
|
||||
|
||||
instance.put('instances/dummy-instance-id/user/dummy-user', body=body)
|
||||
instance._cs_request.assert_called_with(
|
||||
'instances/dummy-instance-id/user/dummy-user', 'PUT', body=body)
|
||||
|
||||
def test_client_delete(self):
|
||||
auth_url = 'http://www.blah.com'
|
||||
instance = other_client.HTTPClient(user='user',
|
||||
password='password',
|
||||
projectid='project_id',
|
||||
timeout=2,
|
||||
auth_url=auth_url)
|
||||
instance._cs_request = mock.Mock()
|
||||
|
||||
instance.delete('/backups/dummy-backup-id')
|
||||
instance._cs_request.assert_called_with('/backups/dummy-backup-id',
|
||||
'DELETE')
|
||||
|
||||
@mock.patch.object(adapter.LegacyJsonAdapter, 'request')
|
||||
def test_database_service_name(self, m_request):
|
||||
m_request.return_value = (mock.MagicMock(status_code=200), None)
|
||||
|
||||
client = other_client.SessionClient(session=mock.MagicMock(),
|
||||
auth=mock.MagicMock())
|
||||
client.request("http://no.where", 'GET')
|
||||
self.assertIsNone(client.database_service_name)
|
||||
|
||||
client = other_client.SessionClient(session=mock.MagicMock(),
|
||||
auth=mock.MagicMock(),
|
||||
database_service_name='myservice')
|
||||
client.request("http://no.where", 'GET')
|
||||
self.assertEqual('myservice', client.database_service_name)
|
||||
|
||||
@mock.patch.object(adapter.LegacyJsonAdapter, 'request')
|
||||
@mock.patch.object(adapter.LegacyJsonAdapter, 'get_endpoint',
|
||||
return_value=None)
|
||||
def test_error_sessionclient(self, m_end_point, m_request):
|
||||
m_request.return_value = (mock.MagicMock(status_code=200), None)
|
||||
|
||||
self.assertRaises(exceptions.EndpointNotFound,
|
||||
other_client.SessionClient,
|
||||
session=mock.MagicMock(),
|
||||
auth=mock.MagicMock())
|
||||
|
||||
def test_construct_http_client(self):
|
||||
mock_request = mock.Mock()
|
||||
mock_request.return_value = requests.Response()
|
||||
mock_request.return_value.status_code = 200
|
||||
mock_request.return_value.headers = {
|
||||
'x-server-management-url': 'blah.com',
|
||||
'x-auth-token': 'blah',
|
||||
}
|
||||
with mock.patch('requests.request', mock_request):
|
||||
self.assertIsInstance(other_client._construct_http_client(),
|
||||
other_client.HTTPClient)
|
||||
self.assertIsInstance(
|
||||
other_client._construct_http_client(session=mock.Mock(),
|
||||
auth=mock.Mock()),
|
||||
other_client.SessionClient)
|
||||
|
||||
@@ -60,21 +60,17 @@ class SecGroupTest(testtools.TestCase):
|
||||
|
||||
class SecGroupRuleTest(testtools.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
@mock.patch.object(security_groups.SecurityGroupRules, '__init__',
|
||||
mock.Mock(return_value=None))
|
||||
@mock.patch.object(security_groups.SecurityGroupRule, '__init__',
|
||||
mock.Mock(return_value=None))
|
||||
def setUp(self, *args):
|
||||
super(SecGroupRuleTest, self).setUp()
|
||||
self.orig__init = security_groups.SecurityGroupRule.__init__
|
||||
security_groups.SecurityGroupRule.__init__ = mock.Mock(
|
||||
return_value=None
|
||||
)
|
||||
security_groups.SecurityGroupRules.__init__ = mock.Mock(
|
||||
return_value=None
|
||||
)
|
||||
self.security_group_rule = security_groups.SecurityGroupRule()
|
||||
self.security_group_rules = security_groups.SecurityGroupRules()
|
||||
|
||||
def tearDown(self):
|
||||
super(SecGroupRuleTest, self).tearDown()
|
||||
security_groups.SecurityGroupRule.__init__ = self.orig__init
|
||||
|
||||
def test___repr__(self):
|
||||
self.security_group_rule.group_id = 1
|
||||
|
||||
443
troveclient/tests/test_v1_shell.py
Normal file
443
troveclient/tests/test_v1_shell.py
Normal file
@@ -0,0 +1,443 @@
|
||||
# Copyright [2015] Hewlett-Packard Development Company, L.P.
|
||||
# All Rights Reserved.
|
||||
#
|
||||
# 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 fixtures
|
||||
import mock
|
||||
import six
|
||||
|
||||
import troveclient.client
|
||||
from troveclient import exceptions
|
||||
import troveclient.shell
|
||||
from troveclient.tests import fakes
|
||||
from troveclient.tests import utils
|
||||
import troveclient.v1.shell
|
||||
|
||||
|
||||
class ShellFixture(fixtures.Fixture):
|
||||
def setUp(self):
|
||||
super(ShellFixture, self).setUp()
|
||||
self.shell = troveclient.shell.OpenStackTroveShell()
|
||||
|
||||
def tearDown(self):
|
||||
if hasattr(self.shell, 'cs'):
|
||||
self.shell.cs.clear_callstack()
|
||||
super(ShellFixture, self).tearDown()
|
||||
|
||||
|
||||
class ShellTest(utils.TestCase):
|
||||
FAKE_ENV = {
|
||||
'OS_USERNAME': 'username',
|
||||
'OS_PASSWORD': 'password',
|
||||
'OS_PROJECT_ID': 'project_id',
|
||||
'OS_AUTH_URL': 'http://no.where/v2.0',
|
||||
}
|
||||
|
||||
def setUp(self, *args):
|
||||
"""Run before each test."""
|
||||
super(ShellTest, self).setUp()
|
||||
|
||||
for var in self.FAKE_ENV:
|
||||
self.useFixture(fixtures.EnvironmentVariable(var,
|
||||
self.FAKE_ENV[var]))
|
||||
self.shell = self.useFixture(ShellFixture()).shell
|
||||
|
||||
@mock.patch('sys.stdout', new_callable=six.StringIO)
|
||||
@mock.patch('troveclient.client.get_version_map',
|
||||
return_value=fakes.get_version_map())
|
||||
def run_command(self, cmd, mock_stdout, mock_get_version_map):
|
||||
if isinstance(cmd, list):
|
||||
self.shell.main(cmd)
|
||||
else:
|
||||
self.shell.main(cmd.split())
|
||||
return mock_stdout.getvalue()
|
||||
|
||||
def assert_called(self, method, url, body=None, **kwargs):
|
||||
return self.shell.cs.assert_called(method, url, body, **kwargs)
|
||||
|
||||
def assert_called_anytime(self, method, url, body=None):
|
||||
return self.shell.cs.assert_called_anytime(method, url, body)
|
||||
|
||||
def test_instance_list(self):
|
||||
self.run_command('list')
|
||||
self.assert_called('GET', '/instances')
|
||||
|
||||
def test_instance_show(self):
|
||||
self.run_command('show 1234')
|
||||
self.assert_called('GET', '/instances/1234')
|
||||
|
||||
def test_instance_delete(self):
|
||||
self.run_command('delete 1234')
|
||||
self.assert_called('DELETE', '/instances/1234')
|
||||
|
||||
def test_instance_update(self):
|
||||
self.run_command('update 1234')
|
||||
self.assert_called('PATCH', '/instances/1234')
|
||||
|
||||
def test_resize_flavor(self):
|
||||
self.run_command('resize-flavor 1234 1')
|
||||
self.assert_called('POST', '/instances/1234/action')
|
||||
|
||||
def test_resize_volume(self):
|
||||
self.run_command('resize-volume 1234 3')
|
||||
self.assert_called('POST', '/instances/1234/action')
|
||||
|
||||
def test_restart(self):
|
||||
self.run_command('restart 1234')
|
||||
self.assert_called('POST', '/instances/1234/action')
|
||||
|
||||
def test_detach_replica(self):
|
||||
self.run_command('detach-replica 1234')
|
||||
self.assert_called('PATCH', '/instances/1234')
|
||||
|
||||
def test_promote_to_replica_source(self):
|
||||
self.run_command('promote-to-replica-source 1234')
|
||||
self.assert_called('POST', '/instances/1234/action')
|
||||
|
||||
def test_eject_replica_source(self):
|
||||
self.run_command('eject-replica-source 1234')
|
||||
self.assert_called('POST', '/instances/1234/action')
|
||||
|
||||
def test_flavor_list(self):
|
||||
self.run_command('flavor-list')
|
||||
self.assert_called('GET', '/flavors')
|
||||
|
||||
def test_flavor_list_with_datastore(self):
|
||||
cmd = ('flavor-list --datastore_type mysql '
|
||||
'--datastore_version_id some-version-id')
|
||||
self.run_command(cmd)
|
||||
self.assert_called(
|
||||
'GET', '/datastores/mysql/versions/some-version-id/flavors')
|
||||
|
||||
def test_flavor_list_error(self):
|
||||
cmd = 'flavor-list --datastore_type mysql'
|
||||
exepcted_error_msg = ('Specify both <datastore_type> and '
|
||||
'<datastore_version_id> to list datastore '
|
||||
'version associated flavors')
|
||||
self.assertRaisesRegexp(
|
||||
exceptions.CommandError, exepcted_error_msg, self.run_command, cmd)
|
||||
|
||||
def test_flavor_show(self):
|
||||
self.run_command('flavor-show 1')
|
||||
self.assert_called('GET', '/flavors/1')
|
||||
|
||||
def test_cluster_list(self):
|
||||
self.run_command('cluster-list')
|
||||
self.assert_called('GET', '/clusters')
|
||||
|
||||
def test_cluster_show(self):
|
||||
self.run_command('cluster-show cls-1234')
|
||||
self.assert_called('GET', '/clusters/cls-1234')
|
||||
|
||||
def test_cluster_instances(self):
|
||||
self.run_command('cluster-instances cls-1234')
|
||||
self.assert_called('GET', '/clusters/cls-1234')
|
||||
|
||||
def test_cluster_delete(self):
|
||||
self.run_command('cluster-delete cls-1234')
|
||||
self.assert_called('DELETE', '/clusters/cls-1234')
|
||||
|
||||
def test_boot(self):
|
||||
self.run_command('create test-member-1 1 --size 1')
|
||||
self.assert_called_anytime(
|
||||
'POST', '/instances',
|
||||
{'instance': {
|
||||
'volume': {'size': 1},
|
||||
'flavorRef': '1',
|
||||
'name': 'test-member-1',
|
||||
'replica_count': 1
|
||||
}})
|
||||
|
||||
def test_boot_nic_error(self):
|
||||
cmd = ('create test-member-1 1 --size 1 '
|
||||
'--nic net-id=some-id,port-id=some-id')
|
||||
self.assertRaisesRegexp(
|
||||
exceptions.CommandError, 'Invalid nic argument',
|
||||
self.run_command, cmd)
|
||||
|
||||
def test_cluster_create(self):
|
||||
cmd = ('cluster-create test-clstr vertica 7.1 '
|
||||
'--instance flavor_id=2,volume=2 '
|
||||
'--instance flavor_id=2,volume=1')
|
||||
self.run_command(cmd)
|
||||
self.assert_called_anytime(
|
||||
'POST', '/clusters',
|
||||
{'cluster': {
|
||||
'instances': [
|
||||
{
|
||||
'volume': {'size': '2'},
|
||||
'flavorRef': '2'
|
||||
},
|
||||
{
|
||||
'volume': {'size': '1'},
|
||||
'flavorRef': '2'
|
||||
}],
|
||||
'datastore': {'version': '7.1', 'type': 'vertica'},
|
||||
'name': 'test-clstr'}})
|
||||
|
||||
def test_cluster_create_error(self):
|
||||
cmd = ('cluster-create test-clstr vertica 7.1 --instance volume=2 '
|
||||
'--instance flavor_id=2,volume=1')
|
||||
self.assertRaisesRegexp(
|
||||
exceptions.CommandError, 'flavor_id is required',
|
||||
self.run_command, cmd)
|
||||
|
||||
def test_datastore_list(self):
|
||||
self.run_command('datastore-list')
|
||||
self.assert_called('GET', '/datastores')
|
||||
|
||||
def test_datastore_show(self):
|
||||
self.run_command('datastore-show d-123')
|
||||
self.assert_called('GET', '/datastores/d-123')
|
||||
|
||||
def test_datastore_version_list(self):
|
||||
self.run_command('datastore-version-list d-123')
|
||||
self.assert_called('GET', '/datastores/d-123/versions')
|
||||
|
||||
def test_datastore_version_show(self):
|
||||
self.run_command('datastore-version-show v-56 --datastore d-123')
|
||||
self.assert_called('GET', '/datastores/d-123/versions/v-56')
|
||||
|
||||
def test_datastore_version_show_error(self):
|
||||
expected_error_msg = ('The datastore name or id is required to '
|
||||
'retrieve a datastore version by name.')
|
||||
self.assertRaisesRegexp(exceptions.NoUniqueMatch, expected_error_msg,
|
||||
self.run_command,
|
||||
'datastore-version-show v-56')
|
||||
|
||||
def test_configuration_list(self):
|
||||
self.run_command('configuration-list')
|
||||
self.assert_called('GET', '/configurations')
|
||||
|
||||
def test_configuration_show(self):
|
||||
self.run_command('configuration-show c-123')
|
||||
self.assert_called('GET', '/configurations/c-123')
|
||||
|
||||
def test_configuration_create(self):
|
||||
cmd = "configuration-create c-123 some-thing"
|
||||
self.assertRaises(ValueError, self.run_command, cmd)
|
||||
|
||||
def test_configuration_update(self):
|
||||
cmd = "configuration-update c-123 some-thing"
|
||||
self.assertRaises(ValueError, self.run_command, cmd)
|
||||
|
||||
def test_configuration_patch(self):
|
||||
cmd = "configuration-patch c-123 some-thing"
|
||||
self.assertRaises(ValueError, self.run_command, cmd)
|
||||
|
||||
def test_configuration_parameter_list(self):
|
||||
cmd = 'configuration-parameter-list v-156 --datastore d-123'
|
||||
self.run_command(cmd)
|
||||
self.assert_called('GET',
|
||||
'/datastores/d-123/versions/v-156/parameters')
|
||||
|
||||
def test_configuration_parameter_list_error(self):
|
||||
expected_error_msg = ('The datastore name or id is required to '
|
||||
'retrieve the parameters for the configuration '
|
||||
'group by name')
|
||||
self.assertRaisesRegexp(
|
||||
exceptions.NoUniqueMatch, expected_error_msg,
|
||||
self.run_command, 'configuration-parameter-list v-156')
|
||||
|
||||
def test_configuration_parameter_show(self):
|
||||
cmd = ('configuration-parameter-show v_56 '
|
||||
'max_connections --datastore d_123')
|
||||
self.run_command(cmd)
|
||||
self.assert_called(
|
||||
'GET',
|
||||
'/datastores/d_123/versions/v_56/parameters/max_connections')
|
||||
|
||||
def test_configuration_instances(self):
|
||||
cmd = 'configuration-instances c-123'
|
||||
self.run_command(cmd)
|
||||
self.assert_called('GET', '/configurations/c-123/instances')
|
||||
|
||||
def test_configuration_delete(self):
|
||||
self.run_command('configuration-delete c-123')
|
||||
self.assert_called('DELETE', '/configurations/c-123')
|
||||
|
||||
def test_configuration_default(self):
|
||||
self.run_command('configuration-default 1234')
|
||||
self.assert_called('GET', '/instances/1234/configuration')
|
||||
|
||||
def test_configuration_attach(self):
|
||||
self.run_command('configuration-attach 1234 c-123')
|
||||
self.assert_called('PUT', '/instances/1234')
|
||||
|
||||
def test_configuration_detach(self):
|
||||
self.run_command('configuration-detach 1234')
|
||||
self.assert_called('PUT', '/instances/1234')
|
||||
|
||||
def test_metadata_edit(self):
|
||||
self.run_command('metadata-edit 1234 key-123 value-123')
|
||||
self.assert_called('PATCH', '/instances/1234/metadata/key-123')
|
||||
|
||||
def test_metadata_update(self):
|
||||
self.run_command('metadata-update 1234 key-123 key-456 value-123')
|
||||
self.assert_called('PUT', '/instances/1234/metadata/key-123')
|
||||
|
||||
def test_metadata_delete(self):
|
||||
self.run_command('metadata-delete 1234 key-123')
|
||||
self.assert_called('DELETE', '/instances/1234/metadata/key-123')
|
||||
|
||||
def test_metadata_create(self):
|
||||
self.run_command('metadata-create 1234 key123 value123')
|
||||
self.assert_called_anytime(
|
||||
'POST', '/instances/1234/metadata/key123',
|
||||
{'metadata': {'value': 'value123'}})
|
||||
|
||||
def test_metadata_list(self):
|
||||
self.run_command('metadata-list 1234')
|
||||
self.assert_called('GET', '/instances/1234/metadata')
|
||||
|
||||
def test_metadata_show(self):
|
||||
self.run_command('metadata-show 1234 key123')
|
||||
self.assert_called('GET', '/instances/1234/metadata/key123')
|
||||
|
||||
def test_limit_list(self):
|
||||
self.run_command('limit-list')
|
||||
self.assert_called('GET', '/limits')
|
||||
|
||||
def test_backup_list(self):
|
||||
self.run_command('backup-list')
|
||||
self.assert_called('GET', '/backups')
|
||||
|
||||
def test_backup_show(self):
|
||||
self.run_command('backup-show bk-1234')
|
||||
self.assert_called('GET', '/backups/bk-1234')
|
||||
|
||||
def test_backup_list_instance(self):
|
||||
self.run_command('backup-list-instance 1234')
|
||||
self.assert_called('GET', '/instances/1234/backups')
|
||||
|
||||
def test_backup_delete(self):
|
||||
self.run_command('backup-delete bk-1234')
|
||||
self.assert_called('DELETE', '/backups/bk-1234')
|
||||
|
||||
def test_backup_create(self):
|
||||
self.run_command('backup-create 1234 bkp_1')
|
||||
self.assert_called_anytime(
|
||||
'POST', '/backups',
|
||||
{'backup': {
|
||||
'instance': '1234',
|
||||
'name': 'bkp_1'
|
||||
}})
|
||||
|
||||
def test_backup_copy(self):
|
||||
self.run_command('backup-copy new_bkp bk-1234')
|
||||
self.assert_called_anytime(
|
||||
'POST', '/backups',
|
||||
{'backup': {
|
||||
'name': 'new_bkp',
|
||||
'backup': {'region': None, 'id': 'bk-1234'}
|
||||
}})
|
||||
|
||||
def test_database_list(self):
|
||||
self.run_command('database-list 1234')
|
||||
self.assert_called('GET', '/instances/1234/databases')
|
||||
|
||||
def test_database_delete(self):
|
||||
self.run_command('database-delete 1234 db_1')
|
||||
self.assert_called('DELETE', '/instances/1234/databases/db_1')
|
||||
|
||||
def test_database_create(self):
|
||||
cmd = ('database-create 1234 db_1 --character_set utf8 '
|
||||
'--collate utf8_general_ci')
|
||||
self.run_command(cmd)
|
||||
self.assert_called_anytime(
|
||||
'POST', '/instances/1234/databases',
|
||||
{'databases': [{'character_set': 'utf8',
|
||||
'name': 'db_1',
|
||||
'collate': 'utf8_general_ci'}]})
|
||||
|
||||
def test_user_list(self):
|
||||
self.run_command('user-list 1234')
|
||||
self.assert_called('GET', '/instances/1234/users')
|
||||
|
||||
def test_user_show(self):
|
||||
self.run_command('user-show 1234 jacob')
|
||||
self.assert_called('GET', '/instances/1234/users/jacob')
|
||||
|
||||
def test_user_delete(self):
|
||||
self.run_command('user-delete 1234 jacob')
|
||||
self.assert_called('DELETE', '/instances/1234/users/jacob')
|
||||
|
||||
def test_user_create(self):
|
||||
self.run_command('user-create 1234 jacob password')
|
||||
self.assert_called_anytime(
|
||||
'POST', '/instances/1234/users',
|
||||
{'users': [{
|
||||
'password': 'password',
|
||||
'name': 'jacob',
|
||||
'databases': []}]})
|
||||
|
||||
def test_user_show_access(self):
|
||||
self.run_command('user-show-access 1234 jacob')
|
||||
self.assert_called('GET', '/instances/1234/users/jacob/databases')
|
||||
|
||||
def test_user_update_host(self):
|
||||
cmd = 'user-update-attributes 1234 jacob --new_host 10.0.0.1'
|
||||
self.run_command(cmd)
|
||||
self.assert_called('PUT', '/instances/1234/users/jacob')
|
||||
|
||||
def test_user_update_name(self):
|
||||
self.run_command('user-update-attributes 1234 jacob --new_name sam')
|
||||
self.assert_called('PUT', '/instances/1234/users/jacob')
|
||||
|
||||
def test_user_update_password(self):
|
||||
cmd = 'user-update-attributes 1234 jacob --new_password new_pwd'
|
||||
self.run_command(cmd)
|
||||
self.assert_called('PUT', '/instances/1234/users/jacob')
|
||||
|
||||
def test_user_grant_access(self):
|
||||
self.run_command('user-grant-access 1234 jacob db1 db2')
|
||||
self.assert_called('PUT', '/instances/1234/users/jacob/databases')
|
||||
|
||||
def test_user_revoke_access(self):
|
||||
self.run_command('user-revoke-access 1234 jacob db1')
|
||||
self.assert_called('DELETE',
|
||||
'/instances/1234/users/jacob/databases/db1')
|
||||
|
||||
def test_root_enable(self):
|
||||
self.run_command('root-enable 1234')
|
||||
self.assert_called_anytime('POST', '/instances/1234/root')
|
||||
|
||||
def test_root_show(self):
|
||||
self.run_command('root-show 1234')
|
||||
self.assert_called('GET', '/instances/1234/root')
|
||||
|
||||
def test_secgroup_list(self):
|
||||
self.run_command('secgroup-list')
|
||||
self.assert_called('GET', '/security-groups')
|
||||
|
||||
def test_secgroup_show(self):
|
||||
self.run_command('secgroup-show 2')
|
||||
self.assert_called('GET', '/security-groups/2')
|
||||
|
||||
def test_secgroup_list_rules(self):
|
||||
self.run_command('secgroup-list-rules 2')
|
||||
self.assert_called('GET', '/security-groups/2')
|
||||
|
||||
def test_secgroup_delete_rule(self):
|
||||
self.run_command('secgroup-delete-rule 2')
|
||||
self.assert_called('DELETE', '/security-group-rules/2')
|
||||
|
||||
def test_secgroup_add_rule(self):
|
||||
self.run_command('secgroup-add-rule 2 15.0.0.0/24')
|
||||
self.assert_called_anytime(
|
||||
'POST', '/security-group-rules',
|
||||
{'security_group_rule': {
|
||||
'cidr': '15.0.0.0/24',
|
||||
'group_id': '2',
|
||||
}})
|
||||
81
troveclient/tests/utils.py
Normal file
81
troveclient/tests/utils.py
Normal file
@@ -0,0 +1,81 @@
|
||||
# Copyright [2015] Hewlett-Packard Development Company, L.P.
|
||||
#
|
||||
# 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 os
|
||||
|
||||
import fixtures
|
||||
import mock
|
||||
import requests
|
||||
import testtools
|
||||
|
||||
AUTH_URL = "http://localhost:5002/auth_url"
|
||||
AUTH_URL_V1 = "http://localhost:5002/auth_url/v1.0"
|
||||
AUTH_URL_V2 = "http://localhost:5002/auth_url/v2.0"
|
||||
|
||||
|
||||
def _patch_mock_to_raise_for_invalid_assert_calls():
|
||||
def raise_for_invalid_assert_calls(wrapped):
|
||||
def wrapper(_self, name):
|
||||
valid_asserts = [
|
||||
'assert_called_with',
|
||||
'assert_called_once_with',
|
||||
'assert_has_calls',
|
||||
'assert_any_calls']
|
||||
|
||||
if name.startswith('assert') and name not in valid_asserts:
|
||||
raise AttributeError('%s is not a valid mock assert method'
|
||||
% name)
|
||||
|
||||
return wrapped(_self, name)
|
||||
return wrapper
|
||||
mock.Mock.__getattr__ = raise_for_invalid_assert_calls(
|
||||
mock.Mock.__getattr__)
|
||||
|
||||
# NOTE(gibi): needs to be called only once at import time
|
||||
# to patch the mock lib
|
||||
_patch_mock_to_raise_for_invalid_assert_calls()
|
||||
|
||||
|
||||
class TestCase(testtools.TestCase):
|
||||
TEST_REQUEST_BASE = {
|
||||
'verify': True,
|
||||
}
|
||||
|
||||
def setUp(self):
|
||||
super(TestCase, self).setUp()
|
||||
if (os.environ.get('OS_STDOUT_CAPTURE') == 'True' or
|
||||
os.environ.get('OS_STDOUT_CAPTURE') == '1'):
|
||||
stdout = self.useFixture(fixtures.StringStream('stdout')).stream
|
||||
self.useFixture(fixtures.MonkeyPatch('sys.stdout', stdout))
|
||||
if (os.environ.get('OS_STDERR_CAPTURE') == 'True' or
|
||||
os.environ.get('OS_STDERR_CAPTURE') == '1'):
|
||||
stderr = self.useFixture(fixtures.StringStream('stderr')).stream
|
||||
self.useFixture(fixtures.MonkeyPatch('sys.stderr', stderr))
|
||||
|
||||
|
||||
class TestResponse(requests.Response):
|
||||
"""Class used to wrap requests.Response and provide some
|
||||
convenience to initialize with a dict
|
||||
"""
|
||||
|
||||
def __init__(self, data):
|
||||
super(TestResponse, self).__init__()
|
||||
self._text = None
|
||||
if isinstance(data, dict):
|
||||
self.status_code = data.get('status_code')
|
||||
self.headers = data.get('headers')
|
||||
# Fake the text attribute to streamline Response creation
|
||||
self._text = data.get('text')
|
||||
else:
|
||||
self.status_code = data
|
||||
Reference in New Issue
Block a user