Remove code related to V3

Since V3 API is not supported, code related to it should be removed.

Discussion in mailing:
http://lists.openstack.org/pipermail/openstack-dev/2014-December/052095.html

Change-Id: Iac5c5e6d81479cbeb8bf10cfcda1cc5617680de8
This commit is contained in:
Andrey Kurilin 2014-12-03 14:11:39 +02:00
parent 1aa020989e
commit cc62efef41
50 changed files with 3 additions and 8095 deletions

View File

@ -763,7 +763,7 @@ def get_client_class(version):
version_map = {
'1.1': 'novaclient.v1_1.client.Client',
'2': 'novaclient.v1_1.client.Client',
'3': 'novaclient.v3.client.Client',
'3': 'novaclient.v1_1.client.Client',
}
try:
client_path = version_map[str(version)]

View File

@ -56,7 +56,6 @@ from novaclient.i18n import _
from novaclient.openstack.common import cliutils
from novaclient import utils
from novaclient.v1_1 import shell as shell_v1_1
from novaclient.v3 import shell as shell_v3
DEFAULT_OS_COMPUTE_API_VERSION = "1.1"
DEFAULT_NOVA_ENDPOINT_TYPE = 'publicURL'
@ -431,7 +430,7 @@ class OpenStackComputeShell(object):
actions_module = {
'1.1': shell_v1_1,
'2': shell_v1_1,
'3': shell_v3,
'3': shell_v1_1,
}[version]
except KeyError:
actions_module = shell_v1_1

View File

@ -89,9 +89,3 @@ class V1(base.Fixture):
self.requests.register_uri('GET', self.url('detail'),
json=get_os_zone_detail,
headers=self.json_headers)
class V3(V1):
zone_info_key = 'availability_zone_info'
zone_name_key = 'zone_name'
zone_state_key = 'zone_state'

View File

@ -16,7 +16,6 @@ from keystoneclient import fixture
from keystoneclient import session
from novaclient.v1_1 import client as v1_1client
from novaclient.v3 import client as v3client
IDENTITY_URL = 'http://identityserver:5000/v2.0'
COMPUTE_URL = 'http://compute.host'
@ -58,26 +57,9 @@ class V1(fixtures.Fixture):
auth_url=self.identity_url)
class V3(V1):
def new_client(self):
return v3client.Client(username='xx',
password='xx',
project_id='xx',
auth_url=self.identity_url)
class SessionV1(V1):
def new_client(self):
self.session = session.Session()
self.session.auth = v2.Password(self.identity_url, 'xx', 'xx')
return v1_1client.Client(session=self.session)
class SessionV3(V1):
def new_client(self):
self.session = session.Session()
self.session.auth = v2.Password(self.identity_url, 'xx', 'xx')
return v3client.Client(session=self.session)

View File

@ -147,23 +147,3 @@ class V1(BaseFixture):
def get_host_shutdown(self):
return {'host': 'sample_host',
'power_action': 'shutdown'}
class V3(V1):
def put_host_1(self):
return {'host': super(V3, self).put_host_1()}
def put_host_2(self):
return {'host': super(V3, self).put_host_2()}
def put_host_3(self):
return {'host': super(V3, self).put_host_3()}
def get_host_reboot(self):
return {'host': super(V3, self).get_host_reboot()}
def get_host_startup(self):
return {'host': super(V3, self).get_host_startup()}
def get_host_shutdown(self):
return {'host': super(V3, self).get_host_shutdown()}

View File

@ -180,36 +180,3 @@ class V1(base.Fixture):
self.requests.register_uri('GET', self.url(1234, 'uptime'),
json=get_os_hypervisors_uptime,
headers=self.headers)
class V3(V1):
def setUp(self):
super(V3, self).setUp()
get_os_hypervisors_search = {
'hypervisors': [
{'id': 1234, 'hypervisor_hostname': 'hyper1'},
{'id': 5678, 'hypervisor_hostname': 'hyper2'}
]
}
self.requests.register_uri('GET',
self.url('search', query='hyper'),
json=get_os_hypervisors_search,
headers=self.headers)
get_1234_servers = {
'hypervisor': {
'id': 1234,
'hypervisor_hostname': 'hyper1',
'servers': [
{'name': 'inst1', 'id': 'uuid1'},
{'name': 'inst2', 'id': 'uuid2'}
]
},
}
self.requests.register_uri('GET', self.url(1234, 'servers'),
json=get_1234_servers,
headers=self.headers)

View File

@ -45,8 +45,3 @@ class V1(base.Fixture):
self.requests.register_uri('POST', self.url(),
json=post_os_keypairs,
headers=headers)
class V3(V1):
base_url = 'keypairs'

View File

@ -64,21 +64,3 @@ class V1(base.Fixture):
'security_groups': 1,
'security_group_rules': 1
}
class V3(V1):
def setUp(self):
super(V3, self).setUp()
get_detail = {
'quota_set': {
'cores': {'reserved': 0, 'in_use': 0, 'limit': 10},
'instances': {'reserved': 0, 'in_use': 4, 'limit': 50},
'ram': {'reserved': 0, 'in_use': 1024, 'limit': 51200}
}
}
self.requests.register_uri('GET', self.url('test', 'detail'),
json=get_detail,
headers=self.headers)

View File

@ -493,131 +493,3 @@ class V1(Base):
else:
raise AssertionError("Unexpected server action: %s" % action)
return {'server': _body}
class V3(Base):
def setUp(self):
super(V3, self).setUp()
get_interfaces = {
"interface_attachments": [
{
"port_state": "ACTIVE",
"net_id": "net-id-1",
"port_id": "port-id-1",
"mac_address": "aa:bb:cc:dd:ee:ff",
"fixed_ips": [{"ip_address": "1.2.3.4"}],
},
{
"port_state": "ACTIVE",
"net_id": "net-id-1",
"port_id": "port-id-1",
"mac_address": "aa:bb:cc:dd:ee:ff",
"fixed_ips": [{"ip_address": "1.2.3.4"}],
}
]
}
self.requests.register_uri('GET',
self.url('1234', 'os-attach-interfaces'),
json=get_interfaces,
headers=self.json_headers)
attach_body = {'interface_attachment': {}}
self.requests.register_uri('POST',
self.url('1234', 'os-attach-interfaces'),
json=attach_body,
headers=self.json_headers)
self.requests.register_uri('GET',
self.url('1234', 'os-server-diagnostics'),
json=self.diagnostic)
url = self.url('1234', 'os-attach-interfaces', 'port-id')
self.requests.register_uri('DELETE', url)
self.requests.register_uri('GET',
self.url(1234, 'os-server-password'),
json={'password': ''})
def post_servers(self, request, context):
body = jsonutils.loads(request.body)
assert set(body.keys()) <= set(['server'])
fakes.assert_has_keys(
body['server'],
required=['name', 'image_ref', 'flavor_ref'],
optional=['metadata', 'personality',
'os-scheduler-hints:scheduler_hints'])
if body['server']['name'] == 'some-bad-server':
body = self.server_1235
else:
body = self.server_1234
context.status_code = 202
return {'server': body}
def post_servers_1234_action(self, request, context):
context.status_code = 202
body_is_none_list = [
'revert_resize', 'migrate', 'stop', 'start', 'force_delete',
'restore', 'pause', 'unpause', 'lock', 'unlock', 'unrescue',
'resume', 'suspend', 'lock', 'unlock', 'shelve', 'shelve_offload',
'unshelve', 'reset_network', 'rescue', 'confirm_resize']
body_return_map = {
'rescue': {'admin_password': 'RescuePassword'},
'get_console_output': {'output': 'foo'},
'rebuild': {'server': self.server_1234},
}
body_param_check_exists = {
'rebuild': 'image_ref',
'resize': 'flavor_ref'}
body_params_check_exact = {
'reboot': ['type'],
'add_fixed_ip': ['network_id'],
'evacuate': ['host', 'on_shared_storage'],
'remove_fixed_ip': ['address'],
'change_password': ['admin_password'],
'get_console_output': ['length'],
'get_vnc_console': ['type'],
'get_spice_console': ['type'],
'get_serial_console': ['type'],
'reset_state': ['state'],
'create_image': ['name', 'metadata'],
'migrate_live': ['host', 'block_migration', 'disk_over_commit'],
'create_backup': ['name', 'backup_type', 'rotation'],
'attach': ['volume_id', 'device'],
'detach': ['volume_id'],
'swap_volume_attachment': ['old_volume_id', 'new_volume_id']}
body = jsonutils.loads(request.body)
assert len(body.keys()) == 1
action = list(body)[0]
_body = body_return_map.get(action, '')
if action in body_is_none_list:
assert body[action] is None
if action in body_param_check_exists:
assert body_param_check_exists[action] in body[action]
if action == 'evacuate':
body[action].pop('admin_password', None)
if action in body_params_check_exact:
assert set(body[action]) == set(body_params_check_exact[action])
if action == 'reboot':
assert body[action]['type'] in ['HARD', 'SOFT']
elif action == 'confirm_resize':
# This one method returns a different response code
context.status_code = 204
elif action == 'create_image':
context.headers['location'] = "http://blah/images/456"
if action not in set.union(set(body_is_none_list),
set(body_params_check_exact.keys()),
set(body_param_check_exists.keys())):
raise AssertionError("Unexpected server action: %s" % action)
return _body

View File

@ -23,10 +23,8 @@ import requests
import novaclient.client
import novaclient.extension
import novaclient.tests.fakes as fakes
from novaclient.tests import utils
import novaclient.v1_1.client
import novaclient.v3.client
class ClientConnectionPoolTest(utils.TestCase):
@ -140,7 +138,7 @@ class ClientTest(utils.TestCase):
def test_get_client_class_v3(self):
output = novaclient.client.get_client_class('3')
self.assertEqual(output, novaclient.v3.client.Client)
self.assertEqual(output, novaclient.v1_1.client.Client)
def test_get_client_class_v2(self):
output = novaclient.client.get_client_class('2')
@ -199,44 +197,6 @@ class ClientTest(utils.TestCase):
cs.reset_timings()
self.assertEqual(0, len(cs.get_timings()))
def test_client_set_management_url_v3(self):
cs = novaclient.v3.client.Client("user", "password", "project_id",
auth_url="foo/v2")
cs.set_management_url("blabla")
self.assertEqual("blabla", cs.client.management_url)
def test_client_get_reset_timings_v3(self):
cs = novaclient.v3.client.Client("user", "password", "project_id",
auth_url="foo/v2")
self.assertEqual(0, len(cs.get_timings()))
cs.client.times.append("somevalue")
self.assertEqual(["somevalue"], cs.get_timings())
cs.reset_timings()
self.assertEqual(0, len(cs.get_timings()))
def test_clent_extensions_v3(self):
fake_attribute_name1 = "FakeAttribute1"
fake_attribute_name2 = "FakeAttribute2"
extensions = [
novaclient.extension.Extension(fake_attribute_name1, fakes),
novaclient.extension.Extension(fake_attribute_name2, utils),
]
cs = novaclient.v3.client.Client("user", "password", "project_id",
auth_url="foo/v2",
extensions=extensions)
self.assertIsInstance(getattr(cs, fake_attribute_name1, None),
fakes.FakeManager)
self.assertFalse(hasattr(cs, fake_attribute_name2))
@mock.patch.object(novaclient.client.HTTPClient, 'authenticate')
def test_authenticate_call_v3(self, mock_authenticate):
cs = novaclient.v3.client.Client("user", "password", "project_id",
auth_url="foo/v2")
cs.authenticate()
self.assertTrue(mock_authenticate.called)
@mock.patch('novaclient.client.HTTPClient')
def test_contextmanager_v1_1(self, mock_http_client):
fake_client = mock.Mock()
@ -247,16 +207,6 @@ class ClientTest(utils.TestCase):
self.assertTrue(fake_client.open_session.called)
self.assertTrue(fake_client.close_session.called)
@mock.patch('novaclient.client.HTTPClient')
def test_contextmanager_v3(self, mock_http_client):
fake_client = mock.Mock()
mock_http_client.return_value = fake_client
with novaclient.v3.client.Client("user", "password", "project_id",
auth_url="foo/v2"):
pass
self.assertTrue(fake_client.open_session.called)
self.assertTrue(fake_client.close_session.called)
def test_get_password_simple(self):
cs = novaclient.client.HTTPClient("user", "password", "", "")
cs.password_func = mock.Mock()

View File

@ -1,388 +0,0 @@
# Copyright (c) 2011 X.commerce, a business unit of eBay Inc.
# Copyright 2011 OpenStack Foundation
# Copyright 2013 IBM Corp.
#
# 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 datetime
from oslo.utils import strutils
from novaclient.tests import fakes
from novaclient.tests.v1_1 import fakes as fakes_v1_1
from novaclient.v3 import client
class FakeClient(fakes.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)
class FakeHTTPClient(fakes_v1_1.FakeHTTPClient):
#
# Hosts
#
def put_os_hosts_sample_host_1(self, body, **kw):
return (200, {}, {'host': {'host': 'sample-host_1',
'status': 'enabled'}})
def put_os_hosts_sample_host_2(self, body, **kw):
return (200, {}, {'host': {'host': 'sample-host_2',
'maintenance_mode': 'on_maintenance'}})
def put_os_hosts_sample_host_3(self, body, **kw):
return (200, {}, {'host': {'host': 'sample-host_3',
'status': 'enabled',
'maintenance_mode': 'on_maintenance'}})
def get_os_hosts_sample_host_reboot(self, **kw):
return (200, {}, {'host': {'host': 'sample_host',
'power_action': 'reboot'}})
def get_os_hosts_sample_host_startup(self, **kw):
return (200, {}, {'host': {'host': 'sample_host',
'power_action': 'startup'}})
def get_os_hosts_sample_host_shutdown(self, **kw):
return (200, {}, {'host': {'host': 'sample_host',
'power_action': 'shutdown'}})
#
# Flavors
#
post_flavors_1_flavor_extra_specs = (
fakes_v1_1.FakeHTTPClient.post_flavors_1_os_extra_specs)
post_flavors_4_flavor_extra_specs = (
fakes_v1_1.FakeHTTPClient.post_flavors_4_os_extra_specs)
delete_flavors_1_flavor_extra_specs_k1 = (
fakes_v1_1.FakeHTTPClient.delete_flavors_1_os_extra_specs_k1)
def get_flavors_detail(self, **kw):
flavors = {'flavors': [
{'id': 1, 'name': '256 MB Server', 'ram': 256, 'disk': 10,
'ephemeral': 10,
'flavor-access:is_public': True,
'links': {}},
{'id': 2, 'name': '512 MB Server', 'ram': 512, 'disk': 20,
'ephemeral': 20,
'flavor-access:is_public': False,
'links': {}},
{'id': 4, 'name': '1024 MB Server', 'ram': 1024, 'disk': 10,
'ephemeral': 10,
'flavor-access:is_public': True,
'links': {}},
{'id': 'aa1', 'name': '128 MB Server', 'ram': 128, 'disk': 0,
'ephemeral': 0,
'flavor-access:is_public': True,
'links': {}}
]}
if 'is_public' not in kw:
filter_is_public = True
else:
if kw['is_public'].lower() == 'none':
filter_is_public = None
else:
filter_is_public = strutils.bool_from_string(kw['is_public'],
True)
if filter_is_public is not None:
if filter_is_public:
flavors['flavors'] = [
v for v in flavors['flavors']
if v['flavor-access:is_public']
]
else:
flavors['flavors'] = [
v for v in flavors['flavors']
if not v['flavor-access:is_public']
]
return (200, {}, flavors)
#
# Flavor access
#
get_flavors_2_flavor_access = (
fakes_v1_1.FakeHTTPClient.get_flavors_2_os_flavor_access)
get_flavors_2_flavor_extra_specs = (
fakes_v1_1.FakeHTTPClient.get_flavors_2_os_extra_specs)
get_flavors_aa1_flavor_extra_specs = (
fakes_v1_1.FakeHTTPClient.get_flavors_aa1_os_extra_specs)
#
# Images
#
get_v1_images_detail = fakes_v1_1.FakeHTTPClient.get_images_detail
get_v1_images = fakes_v1_1.FakeHTTPClient.get_images
def head_v1_images_1(self, **kw):
headers = {
'x-image-meta-id': '1',
'x-image-meta-name': 'CentOS 5.2',
'x-image-meta-updated': '2010-10-10T12:00:00Z',
'x-image-meta-created': '2010-10-10T12:00:00Z',
'x-image-meta-status': 'ACTIVE',
'x-image-meta-property-test_key': 'test_value'}
return 200, headers, ''
#
# Servers
#
get_servers_1234_os_server_diagnostics = (
fakes_v1_1.FakeHTTPClient.get_servers_1234_diagnostics)
delete_servers_1234_os_attach_interfaces_port_id = (
fakes_v1_1.FakeHTTPClient.delete_servers_1234_os_interface_port_id)
def get_servers_1234_os_attach_interfaces(self, **kw):
return (200, {}, {
"interface_attachments": [
{"port_state": "ACTIVE",
"net_id": "net-id-1",
"port_id": "port-id-1",
"mac_address": "aa:bb:cc:dd:ee:ff",
"fixed_ips": [{"ip_address": "1.2.3.4"}],
},
{"port_state": "ACTIVE",
"net_id": "net-id-1",
"port_id": "port-id-1",
"mac_address": "aa:bb:cc:dd:ee:ff",
"fixed_ips": [{"ip_address": "1.2.3.4"}]}]
})
def post_servers_1234_os_attach_interfaces(self, **kw):
return (200, {}, {'interface_attachment': {}})
def post_servers(self, body, **kw):
assert set(body.keys()) <= set(['server'])
fakes.assert_has_keys(body['server'],
required=['name', 'image_ref', 'flavor_ref'],
optional=['metadata', 'personality',
'os-scheduler-hints:scheduler_hints'])
if body['server']['name'] == 'some-bad-server':
return (202, {}, self.get_servers_1235()[2])
else:
return (202, {}, self.get_servers_1234()[2])
#
# Server Actions
#
def post_servers_1234_action(self, body, **kw):
_headers = None
resp = 202
body_is_none_list = [
'revert_resize', 'migrate', 'stop', 'start', 'force_delete',
'restore', 'pause', 'unpause', 'lock', 'unlock', 'unrescue',
'resume', 'suspend', 'lock', 'unlock', 'shelve', 'shelve_offload',
'unshelve', 'reset_network', 'rescue', 'confirm_resize']
body_return_map = {
'rescue': {'admin_password': 'RescuePassword'},
'get_console_output': {'output': 'foo'},
'rebuild': self.get_servers_1234()[2],
}
body_param_check_exists = {
'rebuild': 'image_ref',
'resize': 'flavor_ref',
'evacuate': 'on_shared_storage'}
body_params_check_exact = {
'reboot': ['type'],
'add_fixed_ip': ['network_id'],
'remove_fixed_ip': ['address'],
'change_password': ['admin_password'],
'get_console_output': ['length'],
'get_vnc_console': ['type'],
'get_spice_console': ['type'],
'reset_state': ['state'],
'create_image': ['name', 'metadata'],
'migrate_live': ['host', 'block_migration', 'disk_over_commit'],
'create_backup': ['name', 'backup_type', 'rotation'],
'detach': ['volume_id'],
'swap_volume_attachment': ['old_volume_id', 'new_volume_id']}
body_params_check_superset = {
'attach': ['volume_id', 'device']}
assert len(body.keys()) == 1
action = list(body)[0]
_body = body_return_map.get(action)
if action in body_is_none_list:
assert body[action] is None
if action in body_param_check_exists:
assert body_param_check_exists[action] in body[action]
if action in body_params_check_exact:
assert set(body[action]) == set(body_params_check_exact[action])
if action in body_params_check_superset:
assert set(body[action]) >= set(body_params_check_superset[action])
if action == 'reboot':
assert body[action]['type'] in ['HARD', 'SOFT']
elif action == 'confirm_resize':
# This one method returns a different response code
resp = 204
elif action == 'create_image':
_headers = dict(location="http://blah/images/456")
if action not in set.union(set(body_is_none_list),
set(body_params_check_exact.keys()),
set(body_param_check_exists.keys()),
set(body_params_check_superset.keys())):
raise AssertionError("Unexpected server action: %s" % action)
return (resp, _headers, _body)
#
# Server password
#
def get_servers_1234_os_server_password(self, **kw):
return (200, {}, {'password': ''})
def delete_servers_1234_os_server_password(self, **kw):
return (202, {}, None)
#
# Availability Zones
#
def get_os_availability_zone(self, **kw):
return (200, {}, {
"availability_zone_info": [
{"zone_name": "zone-1", "zone_state": {"available": True},
"hosts": None},
{"zone_name": "zone-2", "zone_state": {"available": False},
"hosts": None}]})
def get_os_availability_zone_detail(self, **kw):
return (200, {}, {
"availability_zone_info": [
{"zone_name": "zone-1",
"zone_state": {"available": True},
"hosts": {
"fake_host-1": {
"nova-compute": {
"active": True,
"available": True,
"updated_at": datetime.datetime(
2012, 12, 26, 14, 45, 25, 0)}}}},
{"zone_name": "internal",
"zone_state": {
"available": True},
"hosts": {
"fake_host-1": {
"nova-sched": {
"active": True,
"available": True,
"updated_at": datetime.datetime(
2012, 12, 26, 14, 45, 25, 0)}},
"fake_host-2": {
"nova-network": {
"active": True,
"available": False,
"updated_at": datetime.datetime(
2012, 12, 26, 14, 45, 24, 0)}}}},
{"zone_name": "zone-2",
"zone_state": {"available": False},
"hosts": None}]})
#
# Quotas
#
def put_os_quota_sets_97f4c221bff44578b0300df4ef119353(self, body, **kw):
assert list(body) == ['quota_set']
return (200, {}, {
'quota_set': {
'tenant_id': '97f4c221bff44578b0300df4ef119353',
'metadata_items': [],
'injected_file_content_bytes': 1,
'injected_file_path_bytes': 1,
'ram': 1,
'floating_ips': 1,
'instances': 1,
'injected_files': 1,
'cores': 1,
'keypairs': 1,
'security_groups': 1,
'security_group_rules': 1,
'server_groups': 1,
'server_group_members': 1}})
def get_os_quota_sets_test_detail(self, **kw):
return (200, {}, {'quota_set': {
'cores': {'reserved': 0, 'in_use': 0, 'limit': 10},
'instances': {'reserved': 0, 'in_use': 4, 'limit': 50},
'ram': {'reserved': 0, 'in_use': 1024, 'limit': 51200}}})
#
# Hypervisors
#
def get_os_hypervisors_search(self, **kw):
if kw['query'] == 'hyper1':
return (200, {}, {'hypervisors': [
{'id': 1234, 'hypervisor_hostname': 'hyper1'}]})
return (200, {}, {
'hypervisors': [
{'id': 1234, 'hypervisor_hostname': 'hyper1'},
{'id': 5678, 'hypervisor_hostname': 'hyper2'}]})
def get_os_hypervisors_1234_servers(self, **kw):
return (200, {}, {
'hypervisor':
{'id': 1234, 'hypervisor_hostname': 'hyper1',
'servers': [
{'name': 'inst1', 'id': 'uuid1'},
{'name': 'inst2', 'id': 'uuid2'}]}})
#
# Keypairs
#
get_keypairs_test = fakes_v1_1.FakeHTTPClient.get_os_keypairs_test
get_keypairs = fakes_v1_1.FakeHTTPClient.get_os_keypairs
delete_keypairs_test = fakes_v1_1.FakeHTTPClient.delete_os_keypairs_test
post_keypairs = fakes_v1_1.FakeHTTPClient.post_os_keypairs
#
# List all extensions
#
def get_extensions(self, **kw):
exts = [
{
"alias": "os-multinic",
"description": "Multiple network support",
"name": "Multinic",
"version": 1,
},
{
"alias": "os-extended-server-attributes",
"description": "Extended Server Attributes support.",
"name": "ExtendedServerAttributes",
"version": 1,
},
{
"alias": "os-extended-status",
"description": "Extended Status support",
"name": "ExtendedStatus",
"version": 1,
},
]
return (200, {}, {
"extensions": exts,
})

View File

@ -1,29 +0,0 @@
# Copyright 2012 IBM Corp.
# 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.
from novaclient.tests.fixture_data import client
from novaclient.tests.v1_1 import test_agents
class AgentsTest(test_agents.AgentsTest):
scenarios = [('original', {'client_fixture_class': client.V3}),
('session', {'client_fixture_class': client.SessionV3})]
def _build_example_update_body(self):
return {"agent": {
"url": "/yyy/yyyy/yyyy",
"version": "8.0",
"md5hash": "add6bb58e139be103324d04d82d8f546"}}

View File

@ -1,22 +0,0 @@
# Copyright 2013 IBM Corp.
#
# 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 novaclient.tests.fixture_data import client
from novaclient.tests.v1_1 import test_aggregates
class AggregatesTest(test_aggregates.AggregatesTest):
scenarios = [('original', {'client_fixture_class': client.V3}),
('session', {'client_fixture_class': client.SessionV3})]

View File

@ -1,36 +0,0 @@
# Copyright 2011 OpenStack Foundation
# Copyright 2013 IBM Corp.
# 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.
from novaclient.tests.fixture_data import availability_zones as data
from novaclient.tests.fixture_data import client
from novaclient.tests.v1_1 import test_availability_zone
from novaclient.v3 import availability_zones
class AvailabilityZoneTest(test_availability_zone.AvailabilityZoneTest):
from novaclient.v3 import shell # noqa
data_fixture_class = data.V3
scenarios = [('original', {'client_fixture_class': client.V3}),
('session', {'client_fixture_class': client.SessionV3})]
def _assertZone(self, zone, name, status):
self.assertEqual(zone.zone_name, name)
self.assertEqual(zone.zone_state, status)
def _get_availability_zone_type(self):
return availability_zones.AvailabilityZone

View File

@ -1,21 +0,0 @@
#
# 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 novaclient.tests.fixture_data import client
from novaclient.tests.v1_1 import test_certs
class CertsTest(test_certs.CertsTest):
scenarios = [('original', {'client_fixture_class': client.V3}),
('session', {'client_fixture_class': client.SessionV3})]

View File

@ -1,63 +0,0 @@
# Copyright 2012 OpenStack Foundation
# Copyright 2013 IBM Corp.
# 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.
from novaclient.tests import utils
from novaclient.tests.v3 import fakes
from novaclient.v3 import flavor_access
cs = fakes.FakeClient()
class FlavorAccessTest(utils.TestCase):
def test_list_access_by_flavor_private(self):
kwargs = {'flavor': cs.flavors.get(2)}
r = cs.flavor_access.list(**kwargs)
cs.assert_called('GET', '/flavors/2/flavor-access')
for access_list in r:
self.assertIsInstance(access_list,
flavor_access.FlavorAccess)
def test_add_tenant_access(self):
flavor = cs.flavors.get(2)
tenant = 'proj2'
r = cs.flavor_access.add_tenant_access(flavor, tenant)
body = {
"add_tenant_access": {
"tenant_id": "proj2"
}
}
cs.assert_called('POST', '/flavors/2/action', body)
for a in r:
self.assertIsInstance(a, flavor_access.FlavorAccess)
def test_remove_tenant_access(self):
flavor = cs.flavors.get(2)
tenant = 'proj2'
r = cs.flavor_access.remove_tenant_access(flavor, tenant)
body = {
"remove_tenant_access": {
"tenant_id": "proj2"
}
}
cs.assert_called('POST', '/flavors/2/action', body)
for a in r:
self.assertIsInstance(a, flavor_access.FlavorAccess)

View File

@ -1,74 +0,0 @@
# Copyright (c) 2013, OpenStack
# Copyright 2013 IBM Corp.
#
# 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 mock
from novaclient.tests.v1_1 import test_flavors
from novaclient.tests.v3 import fakes
from novaclient.v3 import flavors
class FlavorsTest(test_flavors.FlavorsTest):
def _get_fake_client(self):
return fakes.FakeClient()
def _get_flavor_type(self):
return flavors.Flavor
def _create_body(self, name, ram, vcpus, disk, ephemeral, id, swap,
rxtx_factor, is_public):
return {
"flavor": {
"name": name,
"ram": ram,
"vcpus": vcpus,
"disk": disk,
"ephemeral": ephemeral,
"id": id,
"swap": swap,
"os-flavor-rxtx:rxtx_factor": rxtx_factor,
"flavor-access:is_public": is_public,
}
}
def test_set_keys(self):
f = self.cs.flavors.get(1)
f.set_keys({'k1': 'v1'})
self.cs.assert_called('POST', '/flavors/1/flavor-extra-specs',
{"extra_specs": {'k1': 'v1'}})
def test_set_with_valid_keys(self):
valid_keys = ['key4', 'month.price', 'I-Am:AK-ey.44-',
'key with spaces and _']
f = self.cs.flavors.get(4)
for key in valid_keys:
f.set_keys({key: 'v4'})
self.cs.assert_called('POST', '/flavors/4/flavor-extra-specs',
{"extra_specs": {key: 'v4'}})
@mock.patch.object(flavors.FlavorManager, '_delete')
def test_unset_keys(self, mock_delete):
f = self.cs.flavors.get(1)
keys = ['k1', 'k2']
f.unset_keys(keys)
mock_delete.assert_has_calls([
mock.call("/flavors/1/flavor-extra-specs/k1"),
mock.call("/flavors/1/flavor-extra-specs/k2")
])
def test_get_flavor_details_diablo(self):
# Don't need for V3 API to work against diablo
pass

View File

@ -1,99 +0,0 @@
# Copyright 2013 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.
from novaclient.tests.fixture_data import client
from novaclient.tests.fixture_data import hosts as data
from novaclient.tests import utils
from novaclient.v3 import hosts
class HostsTest(utils.FixturedTestCase):
client_fixture_class = client.V3
data_fixture_class = data.V3
def test_describe_resource(self):
hs = self.cs.hosts.get('host')
self.assert_called('GET', '/os-hosts/host')
for h in hs:
self.assertIsInstance(h, hosts.Host)
def test_list_host(self):
hs = self.cs.hosts.list()
self.assert_called('GET', '/os-hosts')
for h in hs:
self.assertIsInstance(h, hosts.Host)
self.assertEqual('nova1', h.zone)
def test_list_host_with_zone(self):
hs = self.cs.hosts.list('nova')
self.assert_called('GET', '/os-hosts?zone=nova')
for h in hs:
self.assertIsInstance(h, hosts.Host)
self.assertEqual('nova', h.zone)
def test_list_host_with_service(self):
hs = self.cs.hosts.list(service='nova-compute')
self.assert_called('GET', '/os-hosts?service=nova-compute')
for h in hs:
self.assertIsInstance(h, hosts.Host)
self.assertEqual(h.service, 'nova-compute')
def test_list_host_with_zone_and_service(self):
hs = self.cs.hosts.list(service='nova-compute', zone='nova')
self.assert_called('GET', '/os-hosts?zone=nova&service=nova-compute')
for h in hs:
self.assertIsInstance(h, hosts.Host)
self.assertEqual(h.zone, 'nova')
self.assertEqual(h.service, 'nova-compute')
def test_update_enable(self):
host = self.cs.hosts.get('sample_host')[0]
values = {"status": "enabled"}
result = host.update(values)
self.assert_called('PUT', '/os-hosts/sample_host', {"host": values})
self.assertIsInstance(result, hosts.Host)
def test_update_maintenance(self):
host = self.cs.hosts.get('sample_host')[0]
values = {"maintenance_mode": "enable"}
result = host.update(values)
self.assert_called('PUT', '/os-hosts/sample_host', {"host": values})
self.assertIsInstance(result, hosts.Host)
def test_update_both(self):
host = self.cs.hosts.get('sample_host')[0]
values = {"status": "enabled",
"maintenance_mode": "enable"}
result = host.update(values)
self.assert_called('PUT', '/os-hosts/sample_host', {"host": values})
self.assertIsInstance(result, hosts.Host)
def test_host_startup(self):
host = self.cs.hosts.get('sample_host')[0]
host.startup()
self.assert_called(
'GET', '/os-hosts/sample_host/startup')
def test_host_reboot(self):
host = self.cs.hosts.get('sample_host')[0]
host.reboot()
self.assert_called(
'GET', '/os-hosts/sample_host/reboot')
def test_host_shutdown(self):
host = self.cs.hosts.get('sample_host')[0]
host.shutdown()
self.assert_called(
'GET', '/os-hosts/sample_host/shutdown')

View File

@ -1,47 +0,0 @@
# Copyright 2012 OpenStack Foundation
# 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.
from novaclient.tests.fixture_data import client
from novaclient.tests.fixture_data import hypervisors as data
from novaclient.tests.v1_1 import test_hypervisors
class HypervisorsTest(test_hypervisors.HypervisorsTest):
client_fixture_class = client.V3
data_fixture_class = data.V3
def test_hypervisor_search(self):
expected = [
dict(id=1234, hypervisor_hostname='hyper1'),
dict(id=5678, hypervisor_hostname='hyper2')]
result = self.cs.hypervisors.search('hyper')
self.assert_called('GET', '/os-hypervisors/search?query=hyper')
for idx, hyper in enumerate(result):
self.compare_to_expected(expected[idx], hyper)
def test_hypervisor_servers(self):
expected = dict(id=1234,
hypervisor_hostname='hyper1',
servers=[
dict(name='inst1', id='uuid1'),
dict(name='inst2', id='uuid2')])
result = self.cs.hypervisors.servers('1234')
self.assert_called('GET', '/os-hypervisors/1234/servers')
self.compare_to_expected(expected, result)

View File

@ -1,57 +0,0 @@
# Copyright 2013 IBM Corp.
#
# 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 novaclient.tests.fixture_data import client
from novaclient.tests.fixture_data import images as data
from novaclient.tests import utils
from novaclient.v3 import images
class ImagesTest(utils.FixturedTestCase):
client_fixture_class = client.V3
data_fixture_class = data.V3
def test_list_images(self):
il = self.cs.images.list()
self.assert_called('GET', '/v1/images/detail')
for i in il:
self.assertIsInstance(i, images.Image)
def test_list_images_undetailed(self):
il = self.cs.images.list(detailed=False)
self.assert_called('GET', '/v1/images')
for i in il:
self.assertIsInstance(i, images.Image)
def test_list_images_with_limit(self):
self.cs.images.list(limit=4)
self.assert_called('GET', '/v1/images/detail?limit=4')
def test_get_image_details(self):
i = self.cs.images.get(1)
self.assert_called('HEAD', '/v1/images/1')
self.assertIsInstance(i, images.Image)
self.assertEqual('1', i.id)
self.assertEqual('CentOS 5.2', i.name)
def test_find(self):
i = self.cs.images.find(name="CentOS 5.2")
self.assertEqual('1', i.id)
self.assert_called('HEAD', '/v1/images/1')
iml = self.cs.images.findall(status='SAVING')
self.assertEqual(1, len(iml))
self.assertEqual('My Server Backup', iml[0].name)

View File

@ -1,30 +0,0 @@
# Copyright 2013 IBM Corp.
#
# 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 novaclient.tests.fixture_data import client
from novaclient.tests.fixture_data import keypairs as data
from novaclient.tests.v1_1 import test_keypairs
from novaclient.v3 import keypairs
class KeypairsTest(test_keypairs.KeypairsTest):
client_fixture_class = client.V3
data_fixture_class = data.V3
def _get_keypair_type(self):
return keypairs.Keypair
def _get_keypair_prefix(self):
return keypairs.KeypairManager.keypair_prefix

View File

@ -1,88 +0,0 @@
#
# 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 novaclient.tests.fixture_data import client
from novaclient.tests.fixture_data import limits as data
from novaclient.tests import utils
from novaclient.v3 import limits
class LimitsTest(utils.FixturedTestCase):
client_fixture_class = client.V3
data_fixture_class = data.Fixture
def test_get_limits(self):
obj = self.cs.limits.get()
self.assert_called('GET', '/limits')
self.assertIsInstance(obj, limits.Limits)
def test_get_limits_for_a_tenant(self):
obj = self.cs.limits.get(tenant_id=1234)
self.assert_called('GET', '/limits?tenant_id=1234')
self.assertIsInstance(obj, limits.Limits)
def test_absolute_limits(self):
obj = self.cs.limits.get()
expected = (
limits.AbsoluteLimit("maxTotalRAMSize", 51200),
limits.AbsoluteLimit("maxServerMeta", 5),
limits.AbsoluteLimit("maxImageMeta", 5),
limits.AbsoluteLimit("maxPersonality", 5),
limits.AbsoluteLimit("maxPersonalitySize", 10240),
)
abs_limits = list(obj.absolute)
self.assertEqual(len(abs_limits), len(expected))
for limit in abs_limits:
self.assertTrue(limit in expected)
def test_absolute_limits_reserved(self):
obj = self.cs.limits.get(reserved=True)
expected = (
limits.AbsoluteLimit("maxTotalRAMSize", 51200),
limits.AbsoluteLimit("maxServerMeta", 5),
limits.AbsoluteLimit("maxImageMeta", 5),
limits.AbsoluteLimit("maxPersonality", 5),
limits.AbsoluteLimit("maxPersonalitySize", 10240),
)
self.assert_called('GET', '/limits?reserved=1')
abs_limits = list(obj.absolute)
self.assertEqual(len(abs_limits), len(expected))
for limit in abs_limits:
self.assertTrue(limit in expected)
def test_rate_limits(self):
obj = self.cs.limits.get()
expected = (
limits.RateLimit('POST', '*', '.*', 10, 2, 'MINUTE',
'2011-12-15T22:42:45Z'),
limits.RateLimit('PUT', '*', '.*', 10, 2, 'MINUTE',
'2011-12-15T22:42:45Z'),
limits.RateLimit('DELETE', '*', '.*', 100, 100, 'MINUTE',
'2011-12-15T22:42:45Z'),
limits.RateLimit('POST', '*/servers', '^/servers', 25, 24, 'DAY',
'2011-12-15T22:42:45Z'),
)
rate_limits = list(obj.rate)
self.assertEqual(len(rate_limits), len(expected))
for limit in rate_limits:
self.assertTrue(limit in expected)

View File

@ -1,33 +0,0 @@
# Copyright 2014 NEC Corporation. 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.
from novaclient import extension
from novaclient.tests import utils
from novaclient.tests.v3 import fakes
from novaclient.v3 import list_extensions
extensions = [
extension.Extension("list_extensions", list_extensions),
]
cs = fakes.FakeClient(extensions=extensions)
class ListExtensionsTests(utils.TestCase):
def test_list_extensions(self):
all_exts = cs.list_extensions.show_all()
cs.assert_called('GET', '/extensions')
self.assertTrue(len(all_exts) > 0)
for r in all_exts:
self.assertTrue(len(r.summary) > 0)

View File

@ -1,43 +0,0 @@
# Copyright IBM Corp. 2013
#
# 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 novaclient.tests.fixture_data import client
from novaclient.tests.fixture_data import quotas as data
from novaclient.tests.v1_1 import test_quotas
class QuotaSetsTest(test_quotas.QuotaSetsTest):
client_fixture_class = client.V3
data_fixture_class = data.V3
def test_force_update_quota(self):
q = self.cs.quotas.get('97f4c221bff44578b0300df4ef119353')
q.update(cores=2, force=True)
self.assert_called(
'PUT', '/os-quota-sets/97f4c221bff44578b0300df4ef119353',
{'quota_set': {'force': True,
'cores': 2}})
def test_tenant_quotas_get_detail(self):
tenant_id = 'test'
self.cs.quotas.get(tenant_id, detail=True)
self.assert_called('GET', '/os-quota-sets/%s/detail' % tenant_id)
def test_user_quotas_get_detail(self):
tenant_id = 'test'
user_id = 'fake_user'
self.cs.quotas.get(tenant_id, user_id=user_id, detail=True)
url = '/os-quota-sets/%s/detail?user_id=%s' % (tenant_id, user_id)
self.assert_called('GET', url)

View File

@ -1,478 +0,0 @@
# -*- coding: utf-8 -*-
# Copyright 2013 IBM Corp.
#
# 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 mock
import six
from novaclient import exceptions
from novaclient.tests.fixture_data import client
from novaclient.tests.fixture_data import servers as data
from novaclient.tests import utils
from novaclient.v3 import servers
class ServersTest(utils.FixturedTestCase):
client_fixture_class = client.V3
data_fixture_class = data.V3
def test_list_servers(self):
sl = self.cs.servers.list()
self.assert_called('GET', '/servers/detail')
for s in sl:
self.assertIsInstance(s, servers.Server)
def test_list_servers_undetailed(self):
sl = self.cs.servers.list(detailed=False)
self.assert_called('GET', '/servers')
for s in sl:
self.assertIsInstance(s, servers.Server)
def test_list_servers_with_marker_limit(self):
sl = self.cs.servers.list(marker=1234, limit=2)
self.assert_called('GET', '/servers/detail?limit=2&marker=1234')
for s in sl:
self.assertIsInstance(s, servers.Server)
def test_get_server_details(self):
s = self.cs.servers.get(1234)
self.assert_called('GET', '/servers/1234')
self.assertIsInstance(s, servers.Server)
self.assertEqual(1234, s.id)
self.assertEqual('BUILD', s.status)
def test_get_server_promote_details(self):
s1 = self.cs.servers.list(detailed=False)[0]
s2 = self.cs.servers.list(detailed=True)[0]
self.assertNotEqual(s1._info, s2._info)
s1.get()
self.assertEqual(s1._info, s2._info)
def test_create_server(self):
s = self.cs.servers.create(
name="My server",
image=1,
flavor=1,
meta={'foo': 'bar'},
userdata="hello moto",
key_name="fakekey",
files={
'/etc/passwd': 'some data', # a file
'/tmp/foo.txt': six.StringIO('data'), # a stream
}
)
self.assert_called('POST', '/servers')
self.assertIsInstance(s, servers.Server)
def test_create_server_boot_with_nics_ipv4(self):
old_boot = self.cs.servers._boot
nics = [{'net-id': '11111111-1111-1111-1111-111111111111',
'v4-fixed-ip': '10.10.0.7'}]
def wrapped_boot(url, key, *boot_args, **boot_kwargs):
self.assertEqual(boot_kwargs['nics'], nics)
return old_boot(url, key, *boot_args, **boot_kwargs)
with mock.patch.object(self.cs.servers, '_boot', wrapped_boot):
s = self.cs.servers.create(
name="My server",
image=1,
flavor=1,
meta={'foo': 'bar'},
userdata="hello moto",
key_name="fakekey",
nics=nics
)
self.assert_called('POST', '/servers')
self.assertIsInstance(s, servers.Server)
def test_create_server_boot_with_nics_ipv6(self):
old_boot = self.cs.servers._boot
nics = [{'net-id': '11111111-1111-1111-1111-111111111111',
'v6-fixed-ip': '2001:db9:0:1::10'}]
def wrapped_boot(url, key, *boot_args, **boot_kwargs):
self.assertEqual(nics, boot_kwargs['nics'])
return old_boot(url, key, *boot_args, **boot_kwargs)
with mock.patch.object(self.cs.servers, '_boot', wrapped_boot):
s = self.cs.servers.create(
name="My server",
image=1,
flavor=1,
meta={'foo': 'bar'},
userdata="hello moto",
key_name="fakekey",
nics=nics
)
self.assert_called('POST', '/servers')
self.assertIsInstance(s, servers.Server)
def test_create_server_userdata_file_object(self):
s = self.cs.servers.create(
name="My server",
image=1,
flavor=1,
meta={'foo': 'bar'},
userdata=six.StringIO('hello moto'),
files={
'/etc/passwd': 'some data', # a file
'/tmp/foo.txt': six.StringIO('data'), # a stream
},
)
self.assert_called('POST', '/servers')
self.assertIsInstance(s, servers.Server)
def test_create_server_userdata_unicode(self):
s = self.cs.servers.create(
name="My server",
image=1,
flavor=1,
meta={'foo': 'bar'},
userdata=six.u('こんにちは'),
key_name="fakekey",
files={
'/etc/passwd': 'some data', # a file
'/tmp/foo.txt': six.StringIO('data'), # a stream
},
)
self.assert_called('POST', '/servers')
self.assertIsInstance(s, servers.Server)
def test_create_server_userdata_utf8(self):
s = self.cs.servers.create(
name="My server",
image=1,
flavor=1,
meta={'foo': 'bar'},
userdata='こんにちは',
key_name="fakekey",
files={
'/etc/passwd': 'some data', # a file
'/tmp/foo.txt': six.StringIO('data'), # a stream
},
)
self.assert_called('POST', '/servers')
self.assertIsInstance(s, servers.Server)
def test_create_server_return_reservation_id(self):
s = self.cs.servers.create(
name="My server",
image=1,
flavor=1,
reservation_id=True
)
expected_body = {
'server': {
'name': 'My server',
'image_ref': '1',
'flavor_ref': '1',
'os-multiple-create:min_count': 1,
'os-multiple-create:max_count': 1,
'os-multiple-create:return_reservation_id': True,
}
}
self.assert_called('POST', '/servers', expected_body)
self.assertIsInstance(s, servers.Server)
def test_update_server(self):
s = self.cs.servers.get(1234)
# Update via instance
s.update(name='hi')
self.assert_called('PUT', '/servers/1234')
s.update(name='hi')
self.assert_called('PUT', '/servers/1234')
# Silly, but not an error
s.update()
# Update via manager
self.cs.servers.update(s, name='hi')
self.assert_called('PUT', '/servers/1234')
def test_delete_server(self):
s = self.cs.servers.get(1234)
s.delete()
self.assert_called('DELETE', '/servers/1234')
self.cs.servers.delete(1234)
self.assert_called('DELETE', '/servers/1234')
self.cs.servers.delete(s)
self.assert_called('DELETE', '/servers/1234')
def test_delete_server_meta(self):
self.cs.servers.delete_meta(1234, ['test_key'])
self.assert_called('DELETE', '/servers/1234/metadata/test_key')
def test_set_server_meta(self):
self.cs.servers.set_meta(1234, {'test_key': 'test_value'})
self.assert_called('POST', '/servers/1234/metadata',
{'metadata': {'test_key': 'test_value'}})
def test_find(self):
server = self.cs.servers.find(name='sample-server')
self.assert_called('GET', '/servers/1234')
self.assertEqual('sample-server', server.name)
self.assertRaises(exceptions.NoUniqueMatch, self.cs.servers.find,
flavor={"id": 1, "name": "256 MB Server"})
sl = self.cs.servers.findall(flavor={"id": 1, "name": "256 MB Server"})
self.assertEqual([1234, 5678, 9012], [s.id for s in sl])
def test_reboot_server(self):
s = self.cs.servers.get(1234)
s.reboot()
self.assert_called('POST', '/servers/1234/action')
self.cs.servers.reboot(s, reboot_type='HARD')
self.assert_called('POST', '/servers/1234/action')
def test_rebuild_server(self):
s = self.cs.servers.get(1234)
s.rebuild(image=1)
self.assert_called('POST', '/servers/1234/action')
self.cs.servers.rebuild(s, image=1)
self.assert_called('POST', '/servers/1234/action')
s.rebuild(image=1, password='5678')
self.assert_called('POST', '/servers/1234/action')
self.cs.servers.rebuild(s, image=1, password='5678')
self.assert_called('POST', '/servers/1234/action')
def test_resize_server(self):
s = self.cs.servers.get(1234)
s.resize(flavor=1)
self.assert_called('POST', '/servers/1234/action')
self.cs.servers.resize(s, flavor=1)
self.assert_called('POST', '/servers/1234/action')
def test_confirm_resized_server(self):
s = self.cs.servers.get(1234)
s.confirm_resize()
self.assert_called('POST', '/servers/1234/action')
self.cs.servers.confirm_resize(s)
self.assert_called('POST', '/servers/1234/action')
def test_revert_resized_server(self):
s = self.cs.servers.get(1234)
s.revert_resize()
self.assert_called('POST', '/servers/1234/action')
self.cs.servers.revert_resize(s)
self.assert_called('POST', '/servers/1234/action')
def test_migrate_server(self):
s = self.cs.servers.get(1234)
s.migrate()
self.assert_called('POST', '/servers/1234/action')
self.cs.servers.migrate(s)
self.assert_called('POST', '/servers/1234/action')
def test_add_fixed_ip(self):
s = self.cs.servers.get(1234)
s.add_fixed_ip(1)
self.assert_called('POST', '/servers/1234/action')
self.cs.servers.add_fixed_ip(s, 1)
self.assert_called('POST', '/servers/1234/action')
def test_remove_fixed_ip(self):
s = self.cs.servers.get(1234)
s.remove_fixed_ip('10.0.0.1')
self.assert_called('POST', '/servers/1234/action')
self.cs.servers.remove_fixed_ip(s, '10.0.0.1')
self.assert_called('POST', '/servers/1234/action')
def test_stop(self):
s = self.cs.servers.get(1234)
s.stop()
self.assert_called('POST', '/servers/1234/action')
self.cs.servers.stop(s)
self.assert_called('POST', '/servers/1234/action')
def test_force_delete(self):
s = self.cs.servers.get(1234)
s.force_delete()
self.assert_called('POST', '/servers/1234/action')
self.cs.servers.force_delete(s)
self.assert_called('POST', '/servers/1234/action')
def test_restore(self):
s = self.cs.servers.get(1234)
s.restore()
self.assert_called('POST', '/servers/1234/action')
self.cs.servers.restore(s)
self.assert_called('POST', '/servers/1234/action')
def test_start(self):
s = self.cs.servers.get(1234)
s.start()
self.assert_called('POST', '/servers/1234/action')
self.cs.servers.start(s)
self.assert_called('POST', '/servers/1234/action')
def test_rescue(self):
s = self.cs.servers.get(1234)
s.rescue()
self.assert_called('POST', '/servers/1234/action')
self.cs.servers.rescue(s)
self.assert_called('POST', '/servers/1234/action')
def test_unrescue(self):
s = self.cs.servers.get(1234)
s.unrescue()
self.assert_called('POST', '/servers/1234/action')
self.cs.servers.unrescue(s)
self.assert_called('POST', '/servers/1234/action')
def test_lock(self):
s = self.cs.servers.get(1234)
s.lock()
self.assert_called('POST', '/servers/1234/action')
self.cs.servers.lock(s)
self.assert_called('POST', '/servers/1234/action')
def test_unlock(self):
s = self.cs.servers.get(1234)
s.unlock()
self.assert_called('POST', '/servers/1234/action')
self.cs.servers.unlock(s)
self.assert_called('POST', '/servers/1234/action')
def test_backup(self):
s = self.cs.servers.get(1234)
s.backup('back1', 'daily', 1)
self.assert_called('POST', '/servers/1234/action')
self.cs.servers.backup(s, 'back1', 'daily', 2)
self.assert_called('POST', '/servers/1234/action')
def test_get_console_output_without_length(self):
success = 'foo'
s = self.cs.servers.get(1234)
s.get_console_output()
self.assertEqual(success, s.get_console_output())
self.assert_called('POST', '/servers/1234/action')
self.cs.servers.get_console_output(s)
self.assertEqual(success, self.cs.servers.get_console_output(s))
self.assert_called('POST', '/servers/1234/action',
{'get_console_output': {'length': -1}})
def test_get_console_output_with_length(self):
success = 'foo'
s = self.cs.servers.get(1234)
s.get_console_output(length=50)
self.assertEqual(success, s.get_console_output(length=50))
self.assert_called('POST', '/servers/1234/action',
{'get_console_output': {'length': 50}})
self.cs.servers.get_console_output(s, length=50)
self.assertEqual(success,
self.cs.servers.get_console_output(s, length=50))
self.assert_called('POST', '/servers/1234/action',
{'get_console_output': {'length': 50}})
def test_get_password(self):
s = self.cs.servers.get(1234)
self.assertEqual('', s.get_password('/foo/id_rsa'))
self.assert_called('GET', '/servers/1234/os-server-password')
def test_clear_password(self):
s = self.cs.servers.get(1234)
s.clear_password()
self.assert_called('DELETE', '/servers/1234/os-server-password')
def test_get_server_diagnostics(self):
s = self.cs.servers.get(1234)
diagnostics = s.diagnostics()
self.assertTrue(diagnostics is not None)
self.assert_called('GET', '/servers/1234/os-server-diagnostics')
diagnostics_from_manager = self.cs.servers.diagnostics(1234)
self.assertTrue(diagnostics_from_manager is not None)
self.assert_called('GET', '/servers/1234/os-server-diagnostics')
self.assertEqual(diagnostics_from_manager[1], diagnostics[1])
def test_get_vnc_console(self):
s = self.cs.servers.get(1234)
s.get_vnc_console('fake')
self.assert_called('POST', '/servers/1234/action')
self.cs.servers.get_vnc_console(s, 'fake')
self.assert_called('POST', '/servers/1234/action')
def test_get_spice_console(self):
s = self.cs.servers.get(1234)
s.get_spice_console('fake')
self.assert_called('POST', '/servers/1234/action')
self.cs.servers.get_spice_console(s, 'fake')
self.assert_called('POST', '/servers/1234/action')
def test_create_image(self):
s = self.cs.servers.get(1234)
s.create_image('123')
self.assert_called('POST', '/servers/1234/action')
s.create_image('123', {})
self.assert_called('POST', '/servers/1234/action')
self.cs.servers.create_image(s, '123')
self.assert_called('POST', '/servers/1234/action')
self.cs.servers.create_image(s, '123', {})
def test_live_migrate_server(self):
s = self.cs.servers.get(1234)
s.live_migrate(host='hostname', block_migration=False,
disk_over_commit=False)
self.assert_called('POST', '/servers/1234/action')
self.cs.servers.live_migrate(s, host='hostname', block_migration=False,
disk_over_commit=False)
self.assert_called('POST', '/servers/1234/action')
def test_reset_state(self):
s = self.cs.servers.get(1234)
s.reset_state('newstate')
self.assert_called('POST', '/servers/1234/action')
self.cs.servers.reset_state(s, 'newstate')
self.assert_called('POST', '/servers/1234/action')
def test_reset_network(self):
s = self.cs.servers.get(1234)
s.reset_network()
self.assert_called('POST', '/servers/1234/action')
self.cs.servers.reset_network(s)
self.assert_called('POST', '/servers/1234/action')
def test_evacuate(self):
s = self.cs.servers.get(1234)
s.evacuate('fake_target_host', 'True')
self.assert_called('POST', '/servers/1234/action')
self.cs.servers.evacuate(s, 'fake_target_host',
'False', 'NewAdminPassword')
self.assert_called('POST', '/servers/1234/action')
def test_interface_list(self):
s = self.cs.servers.get(1234)
s.interface_list()
self.assert_called('GET', '/servers/1234/os-attach-interfaces')
def test_interface_attach(self):
s = self.cs.servers.get(1234)
s.interface_attach(None, None, None)
self.assert_called('POST', '/servers/1234/os-attach-interfaces')
def test_interface_detach(self):
s = self.cs.servers.get(1234)
s.interface_detach('port-id')
self.assert_called('DELETE',
'/servers/1234/os-attach-interfaces/port-id')

View File

@ -1,38 +0,0 @@
# Copyright 2012 IBM Corp.
# 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.
from novaclient.tests import utils
from novaclient.tests.v3 import fakes
from novaclient.v3 import services
class ServicesTest(utils.TestCase):
def setUp(self):
super(ServicesTest, self).setUp()
self.cs = self._get_fake_client()
self.service_type = self._get_service_type()
def _get_fake_client(self):
return fakes.FakeClient()
def _get_service_type(self):
return services.Service
def _update_body(self, host, binary, disabled_reason=None):
body = {"host": host,
"binary": binary}
if disabled_reason is not None:
body["disabled_reason"] = disabled_reason
return body

View File

@ -1,766 +0,0 @@
# Copyright 2013 Cloudwatt
# Copyright 2010 Jacob Kaplan-Moss
# Copyright 2011 OpenStack Foundation
# Copyright 2012 IBM Corp.
# 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 base64
import os
import fixtures
import mock
import six
from novaclient import exceptions
import novaclient.shell
from novaclient.tests import utils
from novaclient.tests.v3 import fakes
class ShellFixture(fixtures.Fixture):
def setUp(self):
super(ShellFixture, self).setUp()
self.shell = novaclient.shell.OpenStackComputeShell()
def tearDown(self):
# For some method like test_image_meta_bad_action we are
# testing a SystemExit to be thrown and object self.shell has
# no time to get instantatiated which is OK in this case, so
# we make sure the method is there before launching it.
if hasattr(self.shell, 'cs'):
self.shell.cs.clear_callstack()
super(ShellFixture, self).tearDown()
class ShellTest(utils.TestCase):
FAKE_ENV = {
'NOVA_USERNAME': 'username',
'NOVA_PASSWORD': 'password',
'NOVA_PROJECT_ID': 'project_id',
'OS_COMPUTE_API_VERSION': '3',
'NOVA_URL': 'http://no.where',
'OS_AUTH_URL': 'http://no.where/v2.0',
}
def setUp(self):
"""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
self.useFixture(fixtures.MonkeyPatch(
'novaclient.client.get_client_class',
lambda *_: fakes.FakeClient))
@mock.patch('sys.stdout', new_callable=six.StringIO)
def run_command(self, cmd, mock_stdout):
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_list_deleted(self):
self.run_command('list --deleted')
self.assert_called('GET', '/servers/detail?deleted=True')
def test_aggregate_list(self):
self.run_command('aggregate-list')
self.assert_called('GET', '/os-aggregates')
def test_aggregate_create(self):
self.run_command('aggregate-create test_name nova1')
body = {"aggregate": {"name": "test_name",
"availability_zone": "nova1"}}
self.assert_called('POST', '/os-aggregates', body, pos=-2)
self.assert_called('GET', '/os-aggregates/1', pos=-1)
def test_aggregate_delete_by_id(self):
self.run_command('aggregate-delete 1')
self.assert_called('DELETE', '/os-aggregates/1')
def test_aggregate_delete_by_name(self):
self.run_command('aggregate-delete test')
self.assert_called('DELETE', '/os-aggregates/1')
def test_aggregate_update_by_id(self):
self.run_command('aggregate-update 1 new_name')
body = {"aggregate": {"name": "new_name"}}
self.assert_called('PUT', '/os-aggregates/1', body, pos=-2)
self.assert_called('GET', '/os-aggregates/1', pos=-1)
def test_aggregate_update_by_name(self):
self.run_command('aggregate-update test new_name')
body = {"aggregate": {"name": "new_name"}}
self.assert_called('PUT', '/os-aggregates/1', body, pos=-2)
self.assert_called('GET', '/os-aggregates/1', pos=-1)
def test_aggregate_update_with_availability_zone_by_id(self):
self.run_command('aggregate-update 1 foo new_zone')
body = {"aggregate": {"name": "foo", "availability_zone": "new_zone"}}
self.assert_called('PUT', '/os-aggregates/1', body, pos=-2)
self.assert_called('GET', '/os-aggregates/1', pos=-1)
def test_aggregate_update_with_availability_zone_by_name(self):
self.run_command('aggregate-update test foo new_zone')
body = {"aggregate": {"name": "foo", "availability_zone": "new_zone"}}
self.assert_called('PUT', '/os-aggregates/1', body, pos=-2)
self.assert_called('GET', '/os-aggregates/1', pos=-1)
def test_aggregate_set_metadata_add_by_id(self):
self.run_command('aggregate-set-metadata 3 foo=bar')
body = {"set_metadata": {"metadata": {"foo": "bar"}}}
self.assert_called('POST', '/os-aggregates/3/action', body, pos=-2)
self.assert_called('GET', '/os-aggregates/3', pos=-1)
def test_aggregate_set_metadata_add_duplicate_by_id(self):
cmd = 'aggregate-set-metadata 3 test=dup'
self.assertRaises(exceptions.CommandError, self.run_command, cmd)
def test_aggregate_set_metadata_delete_by_id(self):
self.run_command('aggregate-set-metadata 3 none_key')
body = {"set_metadata": {"metadata": {"none_key": None}}}
self.assert_called('POST', '/os-aggregates/3/action', body, pos=-2)
self.assert_called('GET', '/os-aggregates/3', pos=-1)
def test_aggregate_set_metadata_delete_missing_by_id(self):
cmd = 'aggregate-set-metadata 3 delete_key2'
self.assertRaises(exceptions.CommandError, self.run_command, cmd)
def test_aggregate_set_metadata_by_name(self):
self.run_command('aggregate-set-metadata test foo=bar')
body = {"set_metadata": {"metadata": {"foo": "bar"}}}
self.assert_called('POST', '/os-aggregates/1/action', body, pos=-2)
self.assert_called('GET', '/os-aggregates/1', pos=-1)
def test_aggregate_add_host_by_id(self):
self.run_command('aggregate-add-host 1 host1')
body = {"add_host": {"host": "host1"}}
self.assert_called('POST', '/os-aggregates/1/action', body, pos=-2)
self.assert_called('GET', '/os-aggregates/1', pos=-1)
def test_aggregate_add_host_by_name(self):
self.run_command('aggregate-add-host test host1')
body = {"add_host": {"host": "host1"}}
self.assert_called('POST', '/os-aggregates/1/action', body, pos=-2)
self.assert_called('GET', '/os-aggregates/1', pos=-1)
def test_aggregate_remove_host_by_id(self):
self.run_command('aggregate-remove-host 1 host1')
body = {"remove_host": {"host": "host1"}}
self.assert_called('POST', '/os-aggregates/1/action', body, pos=-2)
self.assert_called('GET', '/os-aggregates/1', pos=-1)
def test_aggregate_remove_host_by_name(self):
self.run_command('aggregate-remove-host test host1')
body = {"remove_host": {"host": "host1"}}
self.assert_called('POST', '/os-aggregates/1/action', body, pos=-2)
self.assert_called('GET', '/os-aggregates/1', pos=-1)
def test_aggregate_details_by_id(self):
self.run_command('aggregate-details 1')
self.assert_called('GET', '/os-aggregates/1')
def test_aggregate_details_by_name(self):
self.run_command('aggregate-details test')
self.assert_called('GET', '/os-aggregates')
def test_boot(self):
self.run_command('boot --flavor 1 --image 1 some-server')
self.assert_called_anytime(
'POST', '/servers',
{'server': {
'flavor_ref': '1',
'name': 'some-server',
'image_ref': '1',
'os-multiple-create:min_count': 1,
'os-multiple-create:max_count': 1,
}},
)
def test_boot_image_with(self):
self.run_command("boot --flavor 1"
" --image-with test_key=test_value some-server")
self.assert_called_anytime(
'POST', '/servers',
{'server': {
'flavor_ref': '1',
'name': 'some-server',
'image_ref': '1',
'os-multiple-create:min_count': 1,
'os-multiple-create:max_count': 1,
}},
)
def test_boot_key(self):
self.run_command('boot --flavor 1 --image 1 --key_name 1 some-server')
self.assert_called_anytime(
'POST', '/servers',
{'server': {
'flavor_ref': '1',
'name': 'some-server',
'image_ref': '1',
'key_name': '1',
'os-multiple-create:min_count': 1,
'os-multiple-create:max_count': 1,
}},
)
def test_boot_user_data(self):
file_text = 'text'
with mock.patch('novaclient.v3.shell.open', create=True) as mock_open:
mock_open.return_value = file_text
testfile = 'some_dir/some_file.txt'
self.run_command('boot --flavor 1 --image 1 --user_data %s '
'some-server' % testfile)
mock_open.assert_called_once_with(testfile)
user_data = base64.b64encode(file_text.encode('utf-8')).decode('utf-8')
self.assert_called_anytime(
'POST', '/servers',
{'server': {
'flavor_ref': '1',
'name': 'some-server',
'image_ref': '1',
'os-multiple-create:min_count': 1,
'os-multiple-create:max_count': 1,
'user_data': user_data}},
)
def test_boot_avzone(self):
self.run_command(
'boot --flavor 1 --image 1 --availability-zone avzone '
'some-server')
self.assert_called_anytime(
'POST', '/servers',
{'server': {
'flavor_ref': '1',
'name': 'some-server',
'image_ref': '1',
'os-availability-zone:availability_zone': 'avzone',
'os-multiple-create:min_count': 1,
'os-multiple-create:max_count': 1
}},
)
def test_boot_secgroup(self):
self.run_command(
'boot --flavor 1 --image 1 --security-groups secgroup1,'
'secgroup2 some-server')
self.assert_called_anytime(
'POST', '/servers',
{'server': {
'security_groups': [{'name': 'secgroup1'},
{'name': 'secgroup2'}],
'flavor_ref': '1',
'name': 'some-server',
'image_ref': '1',
'os-multiple-create:min_count': 1,
'os-multiple-create:max_count': 1,
}},
)
def test_boot_config_drive(self):
self.run_command(
'boot --flavor 1 --image 1 --config-drive 1 some-server')
self.assert_called_anytime(
'POST', '/servers',
{'server': {
'flavor_ref': '1',
'name': 'some-server',
'image_ref': '1',
'os-multiple-create:min_count': 1,
'os-multiple-create:max_count': 1,
'os-config-drive:config_drive': True
}},
)
def test_boot_config_drive_custom(self):
self.run_command(
'boot --flavor 1 --image 1 --config-drive /dev/hda some-server')
self.assert_called_anytime(
'POST', '/servers',
{'server': {
'flavor_ref': '1',
'name': 'some-server',
'image_ref': '1',
'os-multiple-create:min_count': 1,
'os-multiple-create:max_count': 1,
'os-config-drive:config_drive': '/dev/hda'
}},
)
def test_boot_invalid_user_data(self):
invalid_file = os.path.join(os.path.dirname(__file__),
'no_such_file')
cmd = ('boot some-server --flavor 1 --image 1'
' --user_data %s' % invalid_file)
self.assertRaises(exceptions.CommandError, self.run_command, cmd)
def test_boot_no_image_no_bdms(self):
cmd = 'boot --flavor 1 some-server'
self.assertRaises(exceptions.CommandError, self.run_command, cmd)
def test_boot_no_flavor(self):
cmd = 'boot --image 1 some-server'
self.assertRaises(exceptions.CommandError, self.run_command, cmd)
def test_boot_no_image_bdms(self):
self.run_command(
'boot --flavor 1 --block_device_mapping vda=blah:::0 some-server'
)
self.assert_called_anytime(
'POST', '/servers',
{'server': {
'flavor_ref': '1',
'name': 'some-server',
'block_device_mapping': [
{
'volume_id': 'blah',
'delete_on_termination': '0',
'device_name': 'vda',
'boot_index': 0,
'uuid': 'blah',
'source_type': ''
}
],
'image_ref': '',
'os-multiple-create:min_count': 1,
'os-multiple-create:max_count': 1,
}},
)
def test_boot_image_bdms(self):
self.run_command(
'boot --flavor 1 --image 1 --block-device id=fake-id,'
'source=volume,dest=volume,device=vda,size=1,format=ext4,'
'type=disk,shutdown=preserve some-server'
)
id = ('fake-id,source=volume,dest=volume,device=vda,size=1,'
'format=ext4,type=disk,shutdown=preserve')
self.assert_called_anytime(
'POST', '/servers',
{'server': {
'flavor_ref': '1',
'name': 'some-server',
'block_device_mapping': [
{'device_name': 'id', 'volume_id': id,
'source_type': 'volume', 'boot_index': 0, 'uuid': id}],
'image_ref': '1',
'os-multiple-create:min_count': 1,
'os-multiple-create:max_count': 1,
}},
)
def test_boot_metadata(self):
self.run_command('boot --image 1 --flavor 1 --meta foo=bar=pants'
' --meta spam=eggs some-server ')
self.assert_called_anytime(
'POST', '/servers',
{'server': {
'flavor_ref': '1',
'name': 'some-server',
'image_ref': '1',
'metadata': {'foo': 'bar=pants', 'spam': 'eggs'},
'os-multiple-create:min_count': 1,
'os-multiple-create:max_count': 1,
}},
)
def test_boot_hints(self):
self.run_command('boot --image 1 --flavor 1 '
'--hint a=b1=c1 --hint a2=b2=c2 --hint a=b0=c0 '
'some-server')
self.assert_called_anytime(
'POST', '/servers',
{
'server': {
'flavor_ref': '1',
'name': 'some-server',
'image_ref': '1',
'os-multiple-create:min_count': 1,
'os-multiple-create:max_count': 1,
'os-scheduler-hints:scheduler_hints': {
'a': ['b1=c1', 'b0=c0'], 'a2': 'b2=c2'},
},
},
)
def test_boot_nics(self):
cmd = ('boot --image 1 --flavor 1 '
'--nic net-id=a=c,v4-fixed-ip=10.0.0.1 some-server')
self.run_command(cmd)
self.assert_called_anytime(
'POST', '/servers',
{
'server': {
'flavor_ref': '1',
'name': 'some-server',
'image_ref': '1',
'os-multiple-create:min_count': 1,
'os-multiple-create:max_count': 1,
'networks': [
{'uuid': 'a=c', 'fixed_ip': '10.0.0.1'},
],
},
},
)
def test_boot_nics_ipv6(self):
cmd = ('boot --image 1 --flavor 1 '
'--nic net-id=a=c,v6-fixed-ip=2001:db9:0:1::10 some-server')
self.run_command(cmd)
self.assert_called_anytime(
'POST', '/servers',
{
'server': {
'flavor_ref': '1',
'name': 'some-server',
'image_ref': '1',
'os-multiple-create:min_count': 1,
'os-multiple-create:max_count': 1,
'networks': [
{'uuid': 'a=c', 'fixed_ip': '2001:db9:0:1::10'},
],
},
},
)
def test_boot_nics_both_ipv4_and_ipv6(self):
cmd = ('boot --image 1 --flavor 1 '
'--nic net-id=a=c,v4-fixed-ip=10.0.0.1,'
'v6-fixed-ip=2001:db9:0:1::10 some-server')
self.assertRaises(exceptions.CommandError, self.run_command, cmd)
def test_boot_nics_no_value(self):
cmd = ('boot --image 1 --flavor 1 '
'--nic net-id some-server')
self.assertRaises(exceptions.CommandError, self.run_command, cmd)
def test_boot_nics_random_key(self):
cmd = ('boot --image 1 --flavor 1 '
'--nic net-id=a=c,v4-fixed-ip=10.0.0.1,foo=bar some-server')
self.assertRaises(exceptions.CommandError, self.run_command, cmd)
def test_boot_nics_no_netid_or_portid(self):
cmd = ('boot --image 1 --flavor 1 '
'--nic v4-fixed-ip=10.0.0.1 some-server')
self.assertRaises(exceptions.CommandError, self.run_command, cmd)
def test_boot_nics_netid_and_portid(self):
cmd = ('boot --image 1 --flavor 1 '
'--nic port-id=some=port,net-id=some=net some-server')
self.assertRaises(exceptions.CommandError, self.run_command, cmd)
def test_boot_num_instances(self):
self.run_command('boot --image 1 --flavor 1 --num-instances 3 server')
self.assert_called_anytime(
'POST', '/servers',
{
'server': {
'flavor_ref': '1',
'name': 'server',
'image_ref': '1',
'os-multiple-create:min_count': 1,
'os-multiple-create:max_count': 3,
}
})
def test_boot_invalid_num_instances(self):
cmd = 'boot --image 1 --flavor 1 --num-instances 0 server'
self.assertRaises(exceptions.CommandError, self.run_command, cmd)
def test_boot_num_instances_and_count(self):
cmd = 'boot --image 1 --flavor 1 --num-instances 3 --min-count 3 serv'
self.assertRaises(exceptions.CommandError, self.run_command, cmd)
cmd = 'boot --image 1 --flavor 1 --num-instances 3 --max-count 3 serv'
self.assertRaises(exceptions.CommandError, self.run_command, cmd)
def test_boot_min_max_count(self):
self.run_command('boot --image 1 --flavor 1 --max-count 3 server')
self.assert_called_anytime(
'POST', '/servers',
{
'server': {
'flavor_ref': '1',
'name': 'server',
'image_ref': '1',
'os-multiple-create:min_count': 1,
'os-multiple-create:max_count': 3,
}
})
self.run_command('boot --image 1 --flavor 1 --min-count 3 server')
self.assert_called_anytime(
'POST', '/servers',
{
'server': {
'flavor_ref': '1',
'name': 'server',
'image_ref': '1',
'os-multiple-create:min_count': 3,
'os-multiple-create:max_count': 3,
}
})
self.run_command('boot --image 1 --flavor 1 '
'--min-count 3 --max-count 3 server')
self.assert_called_anytime(
'POST', '/servers',
{
'server': {
'flavor_ref': '1',
'name': 'server',
'image_ref': '1',
'os-multiple-create:min_count': 3,
'os-multiple-create:max_count': 3,
}
})
self.run_command('boot --image 1 --flavor 1 '
'--min-count 3 --max-count 5 server')
self.assert_called_anytime(
'POST', '/servers',
{
'server': {
'flavor_ref': '1',
'name': 'server',
'image_ref': '1',
'os-multiple-create:min_count': 3,
'os-multiple-create:max_count': 5,
}
})
cmd = 'boot --image 1 --flavor 1 --min-count 3 --max-count 1 serv'
self.assertRaises(exceptions.CommandError, self.run_command, cmd)
@mock.patch('novaclient.v3.shell._poll_for_status')
def test_boot_with_poll(self, poll_method):
self.run_command('boot --flavor 1 --image 1 some-server --poll')
self.assert_called_anytime(
'POST', '/servers',
{'server': {
'flavor_ref': '1',
'name': 'some-server',
'image_ref': '1',
'os-multiple-create:min_count': 1,
'os-multiple-create:max_count': 1,
}},
)
self.assertEqual(1, poll_method.call_count)
poll_method.assert_has_calls(
[mock.call(self.shell.cs.servers.get, 1234, 'building',
['active'])])
def test_boot_with_poll_to_check_VM_state_error(self):
self.assertRaises(exceptions.InstanceInErrorState, self.run_command,
'boot --flavor 1 --image 1 some-bad-server --poll')
def test_evacuate(self):
self.run_command('evacuate sample-server new_host')
self.assert_called('POST', '/servers/1234/action',
{'evacuate': {'host': 'new_host',
'on_shared_storage': False}})
self.run_command('evacuate sample-server new_host '
'--password NewAdminPass')
self.assert_called('POST', '/servers/1234/action',
{'evacuate': {'host': 'new_host',
'on_shared_storage': False,
'admin_password': 'NewAdminPass'}})
self.run_command('evacuate sample-server new_host')
self.assert_called('POST', '/servers/1234/action',
{'evacuate': {'host': 'new_host',
'on_shared_storage': False}})
self.run_command('evacuate sample-server new_host '
'--on-shared-storage')
self.assert_called('POST', '/servers/1234/action',
{'evacuate': {'host': 'new_host',
'on_shared_storage': True}})
def test_evacuate_with_no_target_host(self):
self.run_command('evacuate sample-server')
self.assert_called('POST', '/servers/1234/action',
{'evacuate': {'on_shared_storage': False}})
self.run_command('evacuate sample-server --password NewAdminPass')
self.assert_called('POST', '/servers/1234/action',
{'evacuate': {'on_shared_storage': False,
'admin_password': 'NewAdminPass'}})
self.run_command('evacuate sample-server --on-shared-storage')
self.assert_called('POST', '/servers/1234/action',
{'evacuate': {'on_shared_storage': True}})
def test_boot_named_flavor(self):
self.run_command(["boot", "--image", "1",
"--flavor", "512 MB Server",
"--max-count", "3", "server"])
self.assert_called('GET', '/flavors/512 MB Server', pos=0)
self.assert_called('GET', '/flavors?is_public=None', pos=1)
self.assert_called('GET', '/flavors?is_public=None', pos=2)
self.assert_called('GET', '/flavors/2', pos=3)
self.assert_called(
'POST', '/servers',
{
'server': {
'flavor_ref': '2',
'name': 'server',
'image_ref': '1',
'os-multiple-create:min_count': 1,
'os-multiple-create:max_count': 3,
}
}, pos=4)
def test_flavor_show_by_name(self):
self.run_command(['flavor-show', '128 MB Server'])
self.assert_called('GET', '/flavors/128 MB Server', pos=0)
self.assert_called('GET', '/flavors?is_public=None', pos=1)
self.assert_called('GET', '/flavors?is_public=None', pos=2)
self.assert_called('GET', '/flavors/aa1', pos=3)
self.assert_called('GET', '/flavors/aa1/flavor-extra-specs', pos=4)
def test_flavor_show_by_name_priv(self):
self.run_command(['flavor-show', '512 MB Server'])
self.assert_called('GET', '/flavors/512 MB Server', pos=0)
self.assert_called('GET', '/flavors?is_public=None', pos=1)
self.assert_called('GET', '/flavors?is_public=None', pos=2)
self.assert_called('GET', '/flavors/2', pos=3)
self.assert_called('GET', '/flavors/2/flavor-extra-specs', pos=4)
def test_host_evacuate_live_with_no_target_host(self):
self.run_command('host-evacuate-live hyper1')
self.assert_called('GET', '/os-hypervisors/search?query=hyper1', pos=0)
self.assert_called('GET', '/os-hypervisors/1234/servers', pos=1)
body = {'migrate_live': {'host': None,
'block_migration': False,
'disk_over_commit': False}}
self.assert_called('POST', '/servers/uuid1/action', body, pos=2)
self.assert_called('POST', '/servers/uuid2/action', body, pos=3)
def test_host_evacuate_live_with_target_host(self):
self.run_command('host-evacuate-live hyper1 '
'--target-host hostname')
self.assert_called('GET', '/os-hypervisors/search?query=hyper1', pos=0)
self.assert_called('GET', '/os-hypervisors/1234/servers', pos=1)
body = {'migrate_live': {'host': 'hostname',
'block_migration': False,
'disk_over_commit': False}}
self.assert_called('POST', '/servers/uuid1/action', body, pos=2)
self.assert_called('POST', '/servers/uuid2/action', body, pos=3)
def test_host_evacuate_live_with_block_migration(self):
self.run_command('host-evacuate-live --block-migrate hyper1')
self.assert_called('GET', '/os-hypervisors/search?query=hyper1', pos=0)
self.assert_called('GET', '/os-hypervisors/1234/servers', pos=1)
body = {'migrate_live': {'host': None,
'block_migration': True,
'disk_over_commit': False}}
self.assert_called('POST', '/servers/uuid1/action', body, pos=2)
self.assert_called('POST', '/servers/uuid2/action', body, pos=3)
def test_host_evacuate_live_with_disk_over_commit(self):
self.run_command('host-evacuate-live --disk-over-commit hyper1')
self.assert_called('GET', '/os-hypervisors/search?query=hyper1', pos=0)
self.assert_called('GET', '/os-hypervisors/1234/servers', pos=1)
body = {'migrate_live': {'host': None,
'block_migration': False,
'disk_over_commit': True}}
self.assert_called('POST', '/servers/uuid1/action', body, pos=2)
self.assert_called('POST', '/servers/uuid2/action', body, pos=3)
def test_delete(self):
self.run_command('delete 1234')
self.assert_called('DELETE', '/servers/1234')
self.run_command('delete sample-server')
self.assert_called('DELETE', '/servers/1234')
def test_delete_two_with_two_existent(self):
self.run_command('delete 1234 5678')
self.assert_called('DELETE', '/servers/1234', pos=-3)
self.assert_called('DELETE', '/servers/5678', pos=-1)
self.run_command('delete sample-server sample-server2')
self.assert_called('GET', '/servers?name=sample-server', pos=-6)
self.assert_called('GET', '/servers/1234', pos=-5)
self.assert_called('DELETE', '/servers/1234', pos=-4)
self.assert_called('GET', '/servers?name=sample-server2', pos=-3)
self.assert_called('GET', '/servers/5678', pos=-2)
self.assert_called('DELETE', '/servers/5678', pos=-1)
def test_delete_two_with_one_nonexistent(self):
cmd = 'delete 1234 123456789'
self.assertRaises(exceptions.CommandError, self.run_command, cmd)
self.assert_called_anytime('DELETE', '/servers/1234')
cmd = 'delete sample-server nonexistentserver'
self.assertRaises(exceptions.CommandError, self.run_command, cmd)
self.assert_called_anytime('DELETE', '/servers/1234')
def test_delete_one_with_one_nonexistent(self):
cmd = 'delete 123456789'
self.assertRaises(exceptions.CommandError, self.run_command, cmd)
cmd = 'delete nonexistent-server1'
self.assertRaises(exceptions.CommandError, self.run_command, cmd)
def test_delete_two_with_two_nonexistent(self):
cmd = 'delete 123456789 987654321'
self.assertRaises(exceptions.CommandError, self.run_command, cmd)
cmd = 'delete nonexistent-server1 nonexistent-server2'
self.assertRaises(exceptions.CommandError, self.run_command, cmd)
class GetFirstEndpointTest(utils.TestCase):
def test_only_one_endpoint(self):
"""If there is only one endpoint, it is returned."""
endpoint = {"url": "test"}
result = novaclient.v3.shell._get_first_endpoint([endpoint], "XYZ")
self.assertEqual(endpoint, result)
def test_multiple_endpoints(self):
"""If there are multiple endpoints, the first one of the appropriate
region is returned.
"""
endpoints = [
{"region": "XYZ"},
{"region": "ORD", "number": 1},
{"region": "ORD", "number": 2}
]
result = novaclient.v3.shell._get_first_endpoint(endpoints, "ORD")
self.assertEqual(endpoints[1], result)
def test_multiple_endpoints_but_none_suitable(self):
"""If there are multiple endpoints but none of them are suitable, an
exception is raised.
"""
endpoints = [
{"region": "XYZ"},
{"region": "PQR"},
{"region": "STU"}
]
self.assertRaises(LookupError,
novaclient.v3.shell._get_first_endpoint,
endpoints, "ORD")
def test_no_endpoints(self):
"""If there are no endpoints available, an exception is raised."""
self.assertRaises(LookupError,
novaclient.v3.shell._get_first_endpoint,
[], "ORD")

View File

@ -1,30 +0,0 @@
# Copyright 2013 IBM Corp.
#
# 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 novaclient.tests.v1_1 import fakes
from novaclient.tests.v1_1 import test_usage
from novaclient.v3 import usage
class UsageTest(test_usage.UsageTest):
def setUp(self):
super(UsageTest, self).setUp()
self.cs = self._get_fake_client()
self.usage_type = self._get_usage_type()
def _get_fake_client(self):
return fakes.FakeClient()
def _get_usage_type(self):
return usage.Usage

View File

@ -1,64 +0,0 @@
# Copyright 2013 IBM Corp.
# 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.
from novaclient.tests import utils
from novaclient.tests.v3 import fakes
class VolumesTest(utils.TestCase):
def setUp(self):
super(VolumesTest, self).setUp()
self.cs = self._get_fake_client()
def _get_fake_client(self):
return fakes.FakeClient()
def test_attach_server_volume(self):
self.cs.volumes.attach_server_volume(
server=1234,
volume_id='15e59938-07d5-11e1-90e3-e3dffe0c5983',
device='/dev/vdb'
)
self.cs.assert_called('POST', '/servers/1234/action')
def test_attach_server_volume_disk_bus_device_type(self):
volume_id = '15e59938-07d5-11e1-90e3-e3dffe0c5983'
device = '/dev/vdb'
disk_bus = 'ide'
device_type = 'cdrom'
self.cs.volumes.attach_server_volume(server=1234,
volume_id=volume_id,
device=device,
disk_bus=disk_bus,
device_type=device_type)
body_params = {'volume_id': volume_id,
'device': device,
'disk_bus': disk_bus,
'device_type': device_type}
body = {'attach': body_params}
self.cs.assert_called('POST', '/servers/1234/action', body)
def test_update_server_volume(self):
vol_id = '15e59938-07d5-11e1-90e3-e3dffe0c5983'
self.cs.volumes.update_server_volume(
server=1234,
old_volume_id='Work',
new_volume_id=vol_id
)
self.cs.assert_called('POST', '/servers/1234/action')
def test_delete_server_volume(self):
self.cs.volumes.delete_server_volume(1234, 'Work')
self.cs.assert_called('POST', '/servers/1234/action')

View File

@ -1,17 +0,0 @@
# Copyright (c) 2012 OpenStack Foundation
#
# 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.
from novaclient.v3.client import Client # noqa

View File

@ -1,34 +0,0 @@
# Copyright 2012 IBM Corp.
# 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.
"""
agent interface
"""
from novaclient.v1_1 import agents
class Agent(agents.Agent):
pass
class AgentsManager(agents.AgentsManager):
resource_class = Agent
def _build_update_body(self, version, url, md5hash):
return {'agent': {
'version': version,
'url': url,
'md5hash': md5hash}}

View File

@ -1,26 +0,0 @@
# Copyright 2012 OpenStack Foundation
# 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.
"""Aggregate interface."""
from novaclient.v1_1 import aggregates
class Aggregate(aggregates.Aggregate):
pass
class AggregateManager(aggregates.AggregateManager):
resource_class = Aggregate

View File

@ -1,33 +0,0 @@
# Copyright 2011 OpenStack Foundation
# Copyright 2013 IBM Corp.
# 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.
"""
Availability Zone interface.
"""
from novaclient.v1_1 import availability_zones
class AvailabilityZone(availability_zones.AvailabilityZone):
pass
class AvailabilityZoneManager(availability_zones.AvailabilityZoneManager):
"""
Manage :class:`AvailabilityZone` resources.
"""
resource_class = AvailabilityZone
return_parameter_name = 'availability_zone_info'

View File

@ -1,30 +0,0 @@
# Copyright 2010 Jacob Kaplan-Moss
# Copyright 2011 OpenStack Foundation
# 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.
"""
Certificate interface.
"""
from novaclient.v1_1 import certs
class Certificate(certs.Certificate):
pass
class CertificateManager(certs.CertificateManager):
pass

View File

@ -1,196 +0,0 @@
# Copyright 2012 OpenStack Foundation
# Copyright 2013 IBM Corp.
#
# 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 novaclient import client
from novaclient.v3 import agents
from novaclient.v3 import aggregates
from novaclient.v3 import availability_zones
from novaclient.v3 import certs
from novaclient.v3 import flavor_access
from novaclient.v3 import flavors
from novaclient.v3 import hosts
from novaclient.v3 import hypervisors
from novaclient.v3 import images
from novaclient.v3 import keypairs
from novaclient.v3 import limits
from novaclient.v3 import list_extensions
from novaclient.v3 import quotas
from novaclient.v3 import servers
from novaclient.v3 import services
from novaclient.v3 import usage
from novaclient.v3 import volumes
class Client(object):
"""
Top-level object to access the OpenStack Compute API.
Create an instance with your creds::
>>> client = Client(USERNAME, PASSWORD, PROJECT_ID, AUTH_URL)
Or, alternatively, you can create a client instance using the
keystoneclient.session API::
>>> from keystoneclient.auth.identity import v2
>>> from keystoneclient import session
>>> from novaclient.client import Client
>>> auth = v2.Password(auth_url=AUTH_URL,
username=USERNAME,
password=PASSWORD,
tenant_name=PROJECT_ID)
>>> sess = session.Session(auth=auth)
>>> nova = client.Client(VERSION, session=sess)
Then call methods on its managers::
>>> client.servers.list()
...
>>> client.flavors.list()
...
It is also possible to use an instance as a context manager in which
case there will be a session kept alive for the duration of the with
statement::
>>> with Client(USERNAME, PASSWORD, PROJECT_ID, AUTH_URL) as client:
... client.servers.list()
... client.flavors.list()
...
It is also possible to have a permanent (process-long) connection pool,
by passing a connection_pool=True::
>>> client = Client(USERNAME, PASSWORD, PROJECT_ID,
... AUTH_URL, connection_pool=True)
"""
def __init__(self, username=None, password=None, project_id=None,
auth_url=None, insecure=False, timeout=None,
proxy_tenant_id=None, proxy_token=None, region_name=None,
endpoint_type='publicURL', extensions=None,
service_type='computev3', service_name=None,
volume_service_name=None, timings=False, bypass_url=None,
os_cache=False, no_cache=True, http_log_debug=False,
auth_system='keystone', auth_plugin=None, auth_token=None,
cacert=None, tenant_id=None, user_id=None,
connection_pool=False, session=None, auth=None,
completion_cache=None):
# NOTE(cyeoh): In the novaclient context (unlike Nova) the
# project_id is not the same as the tenant_id. Here project_id
# is a name (what the Nova API often refers to as a project or
# tenant name) and tenant_id is a UUID (what the Nova API
# often refers to as a project_id or tenant_id).
self.projectid = project_id
self.tenant_id = tenant_id
self.user_id = user_id
self.os_cache = os_cache or not no_cache
# TODO(bnemec): Add back in v3 extensions
self.agents = agents.AgentsManager(self)
self.aggregates = aggregates.AggregateManager(self)
self.availability_zones = \
availability_zones.AvailabilityZoneManager(self)
self.certs = certs.CertificateManager(self)
self.list_extensions = list_extensions.ListExtManager(self)
self.hosts = hosts.HostManager(self)
self.flavors = flavors.FlavorManager(self)
self.flavor_access = flavor_access.FlavorAccessManager(self)
self.hypervisors = hypervisors.HypervisorManager(self)
self.images = images.ImageManager(self)
self.keypairs = keypairs.KeypairManager(self)
self.limits = limits.LimitsManager(self)
self.quotas = quotas.QuotaSetManager(self)
self.servers = servers.ServerManager(self)
self.services = services.ServiceManager(self)
self.usage = usage.UsageManager(self)
self.volumes = volumes.VolumeManager(self)
# Add in any extensions...
if extensions:
for extension in extensions:
if extension.manager_class:
setattr(self, extension.name,
extension.manager_class(self))
self.client = client._construct_http_client(
username=username,
password=password,
user_id=user_id,
project_id=project_id,
tenant_id=tenant_id,
auth_url=auth_url,
auth_token=auth_token,
insecure=insecure,
timeout=timeout,
auth_system=auth_system,
auth_plugin=auth_plugin,
proxy_token=proxy_token,
proxy_tenant_id=proxy_tenant_id,
region_name=region_name,
endpoint_type=endpoint_type,
service_type=service_type,
service_name=service_name,
volume_service_name=volume_service_name,
timings=timings,
bypass_url=bypass_url,
os_cache=self.os_cache,
http_log_debug=http_log_debug,
cacert=cacert,
connection_pool=connection_pool,
session=session,
auth=auth)
self.completion_cache = completion_cache
def write_object_to_completion_cache(self, obj):
if self.completion_cache:
self.completion_cache.write_object(obj)
def clear_completion_cache_for_class(self, obj_class):
if self.completion_cache:
self.completion_cache.clear_class(obj_class)
@client._original_only
def __enter__(self):
self.client.open_session()
return self
@client._original_only
def __exit__(self, t, v, tb):
self.client.close_session()
@client._original_only
def set_management_url(self, url):
self.client.set_management_url(url)
def get_timings(self):
return self.client.get_timings()
def reset_timings(self):
self.client.reset_timings()
@client._original_only
def authenticate(self):
"""
Authenticate against the server.
Normally this is called automatically when you first access the API,
but you can call this method to force authentication right now.
Returns on success; raises :exc:`exceptions.Unauthorized` if the
credentials are wrong.
"""
self.client.authenticate()

View File

@ -1,45 +0,0 @@
# Copyright 2012 OpenStack Foundation
# Copyright 2013 IBM Corp.
# 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.
"""Flavor access interface."""
from novaclient import base
from novaclient.v1_1 import flavor_access
class FlavorAccess(flavor_access.FlavorAccess):
pass
class FlavorAccessManager(flavor_access.FlavorAccessManager):
"""
Manage :class:`FlavorAccess` resources.
"""
resource_class = FlavorAccess
def _list_by_flavor(self, flavor):
return self._list('/flavors/%s/flavor-access' % base.getid(flavor),
'flavor_access')
def add_tenant_access(self, flavor, tenant):
"""Add a tenant to the given flavor access list."""
info = {'tenant_id': tenant}
return self._action('add_tenant_access', flavor, info)
def remove_tenant_access(self, flavor, tenant):
"""Remove a tenant from the given flavor access list."""
info = {'tenant_id': tenant}
return self._action('remove_tenant_access', flavor, info)

View File

@ -1,100 +0,0 @@
# Copyright 2010 Jacob Kaplan-Moss
# Copyright 2013 IBM Corp.
#
# 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.
"""
Flavor interface.
"""
from novaclient import base
from novaclient import utils
from novaclient.v1_1 import flavors
class Flavor(base.Resource):
"""
A flavor is an available hardware configuration for a server.
"""
HUMAN_ID = True
def __repr__(self):
return "<Flavor: %s>" % self.name
@property
def is_public(self):
"""
Provide a user-friendly accessor to flavor-access:is_public
"""
return self._info.get("flavor-access:is_public", 'N/A')
def get_keys(self):
"""
Get extra specs from a flavor.
:param flavor: The :class:`Flavor` to get extra specs from
"""
_resp, body = self.manager.api.client.get(
"/flavors/%s/flavor-extra-specs" % base.getid(self))
return body["extra_specs"]
def set_keys(self, metadata):
"""
Set extra specs on a flavor.
:param flavor: The :class:`Flavor` to set extra spec on
:param metadata: A dict of key/value pairs to be set
"""
utils.validate_flavor_metadata_keys(metadata.keys())
body = {'extra_specs': metadata}
return self.manager._create(
"/flavors/%s/flavor-extra-specs" % base.getid(self), body,
"extra_specs", return_raw=True)
def unset_keys(self, keys):
"""
Unset extra specs on a flavor.
:param flavor: The :class:`Flavor` to unset extra spec on
:param keys: A list of keys to be unset
"""
for k in keys:
self.manager._delete(
"/flavors/%s/flavor-extra-specs/%s" % (base.getid(self), k))
def delete(self):
"""
Delete this flavor.
"""
self.manager.delete(self)
class FlavorManager(flavors.FlavorManager):
resource_class = Flavor
def _build_body(self, name, ram, vcpus, disk, id, swap,
ephemeral, rxtx_factor, is_public):
return {
"flavor": {
"name": name,
"ram": ram,
"vcpus": vcpus,
"disk": disk,
"id": id,
"swap": swap,
"ephemeral": ephemeral,
"os-flavor-rxtx:rxtx_factor": rxtx_factor,
"flavor-access:is_public": is_public,
}
}

View File

@ -1,51 +0,0 @@
# Copyright 2013 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.
""" V3 API versions of the Hosts interface.
Inherits from the 1.1 code because a lot of the functionality is shared.
"""
from novaclient.v1_1 import hosts
Host = hosts.Host
class HostManager(hosts.HostManager):
def update(self, host, values):
"""Update status or maintenance mode for the host."""
body = dict(host=values)
return self._update("/os-hosts/%s" % host, body, response_key='host')
def host_action(self, host, action):
"""Perform an action on a host."""
url = '/os-hosts/{0}/{1}'.format(host, action)
return self._get(url, response_key='host')
def list(self, zone=None, service=None):
"""List cloud hosts."""
filters = []
if zone:
filters.append('zone=%s' % zone)
if service:
filters.append('service=%s' % service)
if filters:
url = '/os-hosts?%s' % '&'.join(filters)
else:
url = '/os-hosts'
return self._list(url, "hosts")

View File

@ -1,49 +0,0 @@
# Copyright 2013 IBM Corp
# 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.
"""
Hypervisors interface
"""
from six.moves.urllib import parse
from novaclient.v1_1 import hypervisors
class Hypervisor(hypervisors.Hypervisor):
pass
class HypervisorManager(hypervisors.HypervisorManager):
resource_class = Hypervisor
def search(self, hypervisor_match):
"""
Get a list of matching hypervisors.
:param servers: If True, server information is also retrieved.
"""
url = ('/os-hypervisors/search?query=%s' %
parse.quote(hypervisor_match, safe=''))
return self._list(url, 'hypervisors')
def servers(self, hypervisor):
"""
Get servers for a specific hypervisor
:param hypervisor: ID of hypervisor to get list of servers for.
"""
return self._get('/os-hypervisors/%s/servers' % hypervisor,
'hypervisor')

View File

@ -1,107 +0,0 @@
# Copyright 2010 Jacob Kaplan-Moss
# Copyright 2013 IBM Corp.
#
# 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.
"""
Image interface.
"""
from oslo.utils import encodeutils
from oslo.utils import strutils
from six.moves.urllib import parse
from novaclient import base
class Image(base.Resource):
"""
An image is a collection of files used to create or rebuild a server.
"""
HUMAN_ID = True
def __repr__(self):
return "<Image: %s>" % self.name
def delete(self):
"""
Delete this image.
"""
self.manager.delete(self)
class ImageManager(base.ManagerWithFind):
"""
Manage :class:`Image` resources.
"""
resource_class = Image
# NOTE(cyeoh): Eventually we'll want novaclient to be smart
# enough to do version discovery, but for now we just request
# the v1 image API
image_api_prefix = '/v1'
def _image_meta_from_headers(self, headers):
meta = {'properties': {}}
safe_decode = encodeutils.safe_decode
for key, value in headers.items():
value = safe_decode(value, incoming='utf-8')
if key.startswith('x-image-meta-property-'):
_key = safe_decode(key[22:], incoming='utf-8')
meta['properties'][_key] = value
elif key.startswith('x-image-meta-'):
_key = safe_decode(key[13:], incoming='utf-8')
meta[_key] = value
for key in ['is_public', 'protected', 'deleted']:
if key in meta:
meta[key] = strutils.bool_from_string(meta[key])
return self._format_image_meta_for_user(meta)
@staticmethod
def _format_image_meta_for_user(meta):
for key in ['size', 'min_ram', 'min_disk']:
if key in meta:
try:
meta[key] = int(meta[key])
except ValueError:
pass
return meta
def get(self, image):
"""
Get an image.
:param image: The ID of the image to get.
:rtype: :class:`Image`
"""
url = "%s/images/%s" % (self.image_api_prefix, base.getid(image))
resp, _ = self.api.client._cs_request(url, 'HEAD')
foo = self._image_meta_from_headers(resp.headers)
return Image(self, foo)
def list(self, detailed=True, limit=None):
"""
Get a list of all images.
:rtype: list of :class:`Image`
:param limit: maximum number of images to return.
"""
params = {}
detail = ''
if detailed:
detail = '/detail'
if limit:
params['limit'] = int(limit)
query = '?%s' % parse.urlencode(params) if params else ''
return self._list('/v1/images%s%s' % (detail, query), 'images')

View File

@ -1,28 +0,0 @@
# Copyright 2013 IBM Corp.
#
# 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.
"""
Keypair interface
"""
from novaclient.v1_1 import keypairs
class Keypair(keypairs.Keypair):
pass
class KeypairManager(keypairs.KeypairManager):
resource_class = Keypair
keypair_prefix = "keypairs"

View File

@ -1,50 +0,0 @@
# 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.
from six.moves.urllib import parse
from novaclient.v1_1 import limits
class Limits(limits.Limits):
pass
class RateLimit(limits.RateLimit):
pass
class AbsoluteLimit(limits.AbsoluteLimit):
pass
class LimitsManager(limits.LimitsManager):
"""Manager object used to interact with limits resource."""
resource_class = Limits
def get(self, reserved=False, tenant_id=None):
"""
Get a specific extension.
:rtype: :class:`Limits`
"""
opts = {}
if reserved:
opts['reserved'] = 1
if tenant_id:
opts['tenant_id'] = tenant_id
query_string = "?%s" % parse.urlencode(opts) if opts else ""
return self._get("/limits%s" % query_string, "limits")

View File

@ -1,26 +0,0 @@
# Copyright 2014 NEC Corporation. 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.
"""
extension interface
"""
from novaclient.v1_1.contrib import list_extensions
class ListExtResource(list_extensions.ListExtResource):
pass
class ListExtManager(list_extensions.ListExtManager):
pass

View File

@ -1,42 +0,0 @@
# Copyright 2011 OpenStack Foundation
# 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.
from novaclient.v1_1 import quotas
class QuotaSet(quotas.QuotaSet):
pass
class QuotaSetManager(quotas.QuotaSetManager):
resource_class = QuotaSet
def get(self, tenant_id, user_id=None, detail=False):
if detail:
detail_string = '/detail'
else:
detail_string = ''
if hasattr(tenant_id, 'tenant_id'):
tenant_id = tenant_id.tenant_id
if user_id:
url = '/os-quota-sets/%s%s?user_id=%s' % (tenant_id, detail_string,
user_id)
else:
url = '/os-quota-sets/%s%s' % (tenant_id, detail_string)
return self._get(url, "quota_set")
def _update_body(self, tenant_id, **kwargs):
return {'quota_set': kwargs}

File diff suppressed because it is too large Load Diff

View File

@ -1,34 +0,0 @@
# Copyright 2013 IBM Corp.
#
# 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.
"""
service interface
"""
from novaclient.v1_1 import services
class Service(services.Service):
pass
class ServiceManager(services.ServiceManager):
resource_class = Service
def _update_body(self, host, binary, disabled_reason=None):
body = {"service":
{"host": host,
"binary": binary}}
if disabled_reason is not None:
body["service"]["disabled_reason"] = disabled_reason
return body

File diff suppressed because it is too large Load Diff

View File

@ -1,27 +0,0 @@
# Copyright 2013 IBM Corp.
#
# 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.
"""
Usage interface.
"""
from novaclient.v1_1 import usage
class Usage(usage.Usage):
pass
class UsageManager(usage.UsageManager):
pass

View File

@ -1,75 +0,0 @@
# Copyright 2013 IBM Corp.
#
# 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.
"""
Volume interface
"""
from novaclient import base
class VolumeManager(base.Manager):
"""
Manage :class:`Volume` resources.
"""
def attach_server_volume(self, server, volume_id, device,
disk_bus=None, device_type=None):
"""
Attach a volume identified by the volume ID to the given server ID
:param server: The server (or it's ID)
:param volume_id: The ID of the volume to attach.
:param device: The device name
:param disk_bus: The disk bus of the volume
:param device_type: The device type of the volume
:rtype: :class:`Volume`
"""
body = {'volume_id': volume_id, 'device': device}
if disk_bus:
body['disk_bus'] = disk_bus
if device_type:
body['device_type'] = device_type
return self._action('attach', server, body)
def update_server_volume(self, server, old_volume_id, new_volume_id):
"""
Update the volume identified by the attachment ID, that is attached to
the given server ID
:param server_id: The server (or it's ID)
:param old_volume_id: The ID of the attachment
:param new_volume_id: The ID of the new volume to attach
:rtype: :class:`Volume`
"""
body = {'new_volume_id': new_volume_id, 'old_volume_id': old_volume_id}
return self._action('swap_volume_attachment', server, body)
def delete_server_volume(self, server, volume_id):
"""
Detach a volume identified by the attachment ID from the given server
:param server_id: The ID of the server
:param volume_id: The ID of the attachment
"""
return self._action('detach', server, {'volume_id': volume_id})
def _action(self, action, server, info=None, **kwargs):
"""
Perform a server "action" -- reboot/rebuild/resize/etc.
"""
body = {action: info}
self.run_hooks('modify_body_for_action', body, **kwargs)
url = '/servers/%s/action' % base.getid(server)
return self.api.client.post(url, body=body)