summaryrefslogtreecommitdiff
path: root/openstackclient/tests/unit/compute
diff options
context:
space:
mode:
authorSteve Martinelli <s.martinelli@gmail.com>2016-09-05 22:14:33 -0700
committerSteve Martinelli <s.martinelli@gmail.com>2016-09-08 15:19:50 -0700
commit39839def2e356e8d145be89380c73a71423cf06d (patch)
treef53c090b7ded46554866436e9d687492d2cef487 /openstackclient/tests/unit/compute
parent7d1a5d0854c732681a130255ddc6abb2e9721a7a (diff)
downloadpython-openstackclient-39839def2e356e8d145be89380c73a71423cf06d.tar.gz
move unit tests to new "unit" test module
this will better isolate the unit tests from the functional tests. unfortunately, the "integration" tests had to be lumped into the "unit" tests since we need the separation in testr.conf Change-Id: Ifd12198c1f90e4e3c951c73bfa1884ab300d8ded
Diffstat (limited to 'openstackclient/tests/unit/compute')
-rw-r--r--openstackclient/tests/unit/compute/__init__.py0
-rw-r--r--openstackclient/tests/unit/compute/v2/__init__.py0
-rw-r--r--openstackclient/tests/unit/compute/v2/fakes.py1249
-rw-r--r--openstackclient/tests/unit/compute/v2/test_agent.py320
-rw-r--r--openstackclient/tests/unit/compute/v2/test_aggregate.py447
-rw-r--r--openstackclient/tests/unit/compute/v2/test_console.py196
-rw-r--r--openstackclient/tests/unit/compute/v2/test_flavor.py825
-rw-r--r--openstackclient/tests/unit/compute/v2/test_host.py179
-rw-r--r--openstackclient/tests/unit/compute/v2/test_hypervisor.py221
-rw-r--r--openstackclient/tests/unit/compute/v2/test_hypervisor_stats.py79
-rw-r--r--openstackclient/tests/unit/compute/v2/test_keypair.py312
-rw-r--r--openstackclient/tests/unit/compute/v2/test_server.py1844
-rw-r--r--openstackclient/tests/unit/compute/v2/test_server_backup.py271
-rw-r--r--openstackclient/tests/unit/compute/v2/test_server_group.py284
-rw-r--r--openstackclient/tests/unit/compute/v2/test_server_image.py228
-rw-r--r--openstackclient/tests/unit/compute/v2/test_service.py410
16 files changed, 6865 insertions, 0 deletions
diff --git a/openstackclient/tests/unit/compute/__init__.py b/openstackclient/tests/unit/compute/__init__.py
new file mode 100644
index 00000000..e69de29b
--- /dev/null
+++ b/openstackclient/tests/unit/compute/__init__.py
diff --git a/openstackclient/tests/unit/compute/v2/__init__.py b/openstackclient/tests/unit/compute/v2/__init__.py
new file mode 100644
index 00000000..e69de29b
--- /dev/null
+++ b/openstackclient/tests/unit/compute/v2/__init__.py
diff --git a/openstackclient/tests/unit/compute/v2/fakes.py b/openstackclient/tests/unit/compute/v2/fakes.py
new file mode 100644
index 00000000..0e3d47ba
--- /dev/null
+++ b/openstackclient/tests/unit/compute/v2/fakes.py
@@ -0,0 +1,1249 @@
+# Copyright 2013 Nebula Inc.
+#
+# 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 copy
+import mock
+import uuid
+
+from openstackclient.tests.unit import fakes
+from openstackclient.tests.unit.identity.v2_0 import fakes as identity_fakes
+from openstackclient.tests.unit.image.v2 import fakes as image_fakes
+from openstackclient.tests.unit.network.v2 import fakes as network_fakes
+from openstackclient.tests.unit import utils
+from openstackclient.tests.unit.volume.v2 import fakes as volume_fakes
+
+floating_ip_num = 100
+fix_ip_num = 100
+injected_file_num = 100
+injected_file_size_num = 10240
+injected_path_size_num = 255
+key_pair_num = 100
+core_num = 20
+ram_num = 51200
+instance_num = 10
+property_num = 128
+secgroup_rule_num = 20
+secgroup_num = 10
+servgroup_num = 10
+servgroup_members_num = 10
+project_name = 'project_test'
+QUOTA = {
+ 'project': project_name,
+ 'floating-ips': floating_ip_num,
+ 'fix-ips': fix_ip_num,
+ 'injected-files': injected_file_num,
+ 'injected-file-size': injected_file_size_num,
+ 'injected-path-size': injected_path_size_num,
+ 'key-pairs': key_pair_num,
+ 'cores': core_num,
+ 'ram': ram_num,
+ 'instances': instance_num,
+ 'properties': property_num,
+ 'secgroup_rules': secgroup_rule_num,
+ 'secgroups': secgroup_num,
+ 'server-groups': servgroup_num,
+ 'server-group-members': servgroup_members_num
+}
+
+QUOTA_columns = tuple(sorted(QUOTA))
+QUOTA_data = tuple(QUOTA[x] for x in sorted(QUOTA))
+
+
+class FakeAggregate(object):
+ """Fake one aggregate."""
+
+ @staticmethod
+ def create_one_aggregate(attrs=None):
+ """Create a fake aggregate.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :return:
+ A FakeResource object, with id and other attributes
+ """
+ attrs = attrs or {}
+
+ # Set default attribute
+ aggregate_info = {
+ "name": "aggregate-name-" + uuid.uuid4().hex,
+ "availability_zone": "ag_zone",
+ "hosts": [],
+ "id": "aggregate-id-" + uuid.uuid4().hex,
+ "metadata": {
+ "availability_zone": "ag_zone",
+ }
+ }
+
+ # Overwrite default attributes.
+ aggregate_info.update(attrs)
+
+ aggregate = fakes.FakeResource(
+ info=copy.deepcopy(aggregate_info),
+ loaded=True)
+ return aggregate
+
+ @staticmethod
+ def create_aggregates(attrs=None, count=2):
+ """Create multiple fake aggregates.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :param int count:
+ The number of aggregates to fake
+ :return:
+ A list of FakeResource objects faking the aggregates
+ """
+ aggregates = []
+ for i in range(0, count):
+ aggregates.append(FakeAggregate.create_one_aggregate(attrs))
+
+ return aggregates
+
+ @staticmethod
+ def get_aggregates(aggregates=None, count=2):
+ """Get an iterable MagicMock object with a list of faked aggregates.
+
+ If aggregates list is provided, then initialize the Mock object
+ with the list. Otherwise create one.
+
+ :param List aggregates:
+ A list of FakeResource objects faking aggregates
+ :param int count:
+ The number of aggregates to fake
+ :return:
+ An iterable Mock object with side_effect set to a list of faked
+ aggregates
+ """
+ if aggregates is None:
+ aggregates = FakeAggregate.create_aggregates(count)
+ return mock.MagicMock(side_effect=aggregates)
+
+
+class FakeComputev2Client(object):
+
+ def __init__(self, **kwargs):
+ self.agents = mock.Mock()
+ self.agents.resource_class = fakes.FakeResource(None, {})
+
+ self.aggregates = mock.Mock()
+ self.aggregates.resource_class = fakes.FakeResource(None, {})
+
+ self.availability_zones = mock.Mock()
+ self.availability_zones.resource_class = fakes.FakeResource(None, {})
+
+ self.images = mock.Mock()
+ self.images.resource_class = fakes.FakeResource(None, {})
+
+ self.servers = mock.Mock()
+ self.servers.resource_class = fakes.FakeResource(None, {})
+
+ self.services = mock.Mock()
+ self.services.resource_class = fakes.FakeResource(None, {})
+
+ self.extensions = mock.Mock()
+ self.extensions.resource_class = fakes.FakeResource(None, {})
+
+ self.flavors = mock.Mock()
+ self.flavors.resource_class = fakes.FakeResource(None, {})
+
+ self.flavor_access = mock.Mock()
+ self.flavor_access.resource_class = fakes.FakeResource(None, {})
+
+ self.quotas = mock.Mock()
+ self.quotas.resource_class = fakes.FakeResource(None, {})
+
+ self.quota_classes = mock.Mock()
+ self.quota_classes.resource_class = fakes.FakeResource(None, {})
+
+ self.volumes = mock.Mock()
+ self.volumes.resource_class = fakes.FakeResource(None, {})
+
+ self.hypervisors = mock.Mock()
+ self.hypervisors.resource_class = fakes.FakeResource(None, {})
+
+ self.hypervisors_stats = mock.Mock()
+ self.hypervisors_stats.resource_class = fakes.FakeResource(None, {})
+
+ self.security_groups = mock.Mock()
+ self.security_groups.resource_class = fakes.FakeResource(None, {})
+
+ self.security_group_rules = mock.Mock()
+ self.security_group_rules.resource_class = fakes.FakeResource(None, {})
+
+ self.floating_ips = mock.Mock()
+ self.floating_ips.resource_class = fakes.FakeResource(None, {})
+
+ self.floating_ip_pools = mock.Mock()
+ self.floating_ip_pools.resource_class = fakes.FakeResource(None, {})
+
+ self.networks = mock.Mock()
+ self.networks.resource_class = fakes.FakeResource(None, {})
+
+ self.keypairs = mock.Mock()
+ self.keypairs.resource_class = fakes.FakeResource(None, {})
+
+ self.hosts = mock.Mock()
+ self.hosts.resource_class = fakes.FakeResource(None, {})
+
+ self.server_groups = mock.Mock()
+ self.server_groups.resource_class = fakes.FakeResource(None, {})
+
+ self.auth_token = kwargs['token']
+
+ self.management_url = kwargs['endpoint']
+
+
+class TestComputev2(utils.TestCommand):
+
+ def setUp(self):
+ super(TestComputev2, self).setUp()
+
+ self.app.client_manager.compute = FakeComputev2Client(
+ endpoint=fakes.AUTH_URL,
+ token=fakes.AUTH_TOKEN,
+ )
+
+ self.app.client_manager.identity = identity_fakes.FakeIdentityv2Client(
+ endpoint=fakes.AUTH_URL,
+ token=fakes.AUTH_TOKEN,
+ )
+
+ self.app.client_manager.image = image_fakes.FakeImagev2Client(
+ endpoint=fakes.AUTH_URL,
+ token=fakes.AUTH_TOKEN,
+ )
+
+ self.app.client_manager.network = network_fakes.FakeNetworkV2Client(
+ endpoint=fakes.AUTH_URL,
+ token=fakes.AUTH_TOKEN,
+ )
+
+ self.app.client_manager.volume = volume_fakes.FakeVolumeClient(
+ endpoint=fakes.AUTH_URL,
+ token=fakes.AUTH_TOKEN,
+ )
+
+
+class FakeAgent(object):
+ """Fake one or more agent."""
+
+ @staticmethod
+ def create_one_agent(attrs=None):
+ """Create a fake agent.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :return:
+ A FakeResource object, with agent_id, os, and so on
+ """
+
+ attrs = attrs or {}
+
+ # set default attributes.
+ agent_info = {
+ 'agent_id': 'agent-id-' + uuid.uuid4().hex,
+ 'os': 'agent-os-' + uuid.uuid4().hex,
+ 'architecture': 'agent-architecture',
+ 'version': '8.0',
+ 'url': 'http://127.0.0.1',
+ 'md5hash': 'agent-md5hash',
+ 'hypervisor': 'hypervisor',
+ }
+
+ # Overwrite default attributes.
+ agent_info.update(attrs)
+
+ agent = fakes.FakeResource(info=copy.deepcopy(agent_info),
+ loaded=True)
+ return agent
+
+ @staticmethod
+ def create_agents(attrs=None, count=2):
+ """Create multiple fake agents.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :param int count:
+ The number of agents to fake
+ :return:
+ A list of FakeResource objects faking the agents
+ """
+ agents = []
+ for i in range(0, count):
+ agents.append(FakeAgent.create_one_agent(attrs))
+
+ return agents
+
+
+class FakeExtension(object):
+ """Fake one or more extension."""
+
+ @staticmethod
+ def create_one_extension(attrs=None):
+ """Create a fake extension.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :return:
+ A FakeResource object with name, namespace, etc.
+ """
+ attrs = attrs or {}
+
+ # Set default attributes.
+ extension_info = {
+ 'name': 'name-' + uuid.uuid4().hex,
+ 'namespace': (
+ 'http://docs.openstack.org/compute/ext/multinic/api/v1.1'),
+ 'description': 'description-' + uuid.uuid4().hex,
+ 'updated': '2014-01-07T12:00:0-00:00',
+ 'alias': 'NMN',
+ 'links': ('[{"href":'
+ '"https://github.com/openstack/compute-api", "type":'
+ ' "text/html", "rel": "describedby"}]')
+ }
+
+ # Overwrite default attributes.
+ extension_info.update(attrs)
+
+ extension = fakes.FakeResource(
+ info=copy.deepcopy(extension_info),
+ loaded=True)
+ return extension
+
+
+class FakeHypervisor(object):
+ """Fake one or more hypervisor."""
+
+ @staticmethod
+ def create_one_hypervisor(attrs=None):
+ """Create a fake hypervisor.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :return:
+ A FakeResource object, with id, hypervisor_hostname, and so on
+ """
+ attrs = attrs or {}
+
+ # Set default attributes.
+ hypervisor_info = {
+ 'id': 'hypervisor-id-' + uuid.uuid4().hex,
+ 'hypervisor_hostname': 'hypervisor-hostname-' + uuid.uuid4().hex,
+ 'status': 'enabled',
+ 'host_ip': '192.168.0.10',
+ 'cpu_info': {
+ 'aaa': 'aaa',
+ },
+ 'free_disk_gb': 50,
+ 'hypervisor_version': 2004001,
+ 'disk_available_least': 50,
+ 'local_gb': 50,
+ 'free_ram_mb': 1024,
+ 'service': {
+ 'host': 'aaa',
+ 'disabled_reason': None,
+ 'id': 1,
+ },
+ 'vcpus_used': 0,
+ 'hypervisor_type': 'QEMU',
+ 'local_gb_used': 0,
+ 'vcpus': 4,
+ 'memory_mb_used': 512,
+ 'memory_mb': 1024,
+ 'current_workload': 0,
+ 'state': 'up',
+ 'running_vms': 0,
+ }
+
+ # Overwrite default attributes.
+ hypervisor_info.update(attrs)
+
+ hypervisor = fakes.FakeResource(info=copy.deepcopy(hypervisor_info),
+ loaded=True)
+ return hypervisor
+
+ @staticmethod
+ def create_hypervisors(attrs=None, count=2):
+ """Create multiple fake hypervisors.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :param int count:
+ The number of hypervisors to fake
+ :return:
+ A list of FakeResource objects faking the hypervisors
+ """
+ hypervisors = []
+ for i in range(0, count):
+ hypervisors.append(FakeHypervisor.create_one_hypervisor(attrs))
+
+ return hypervisors
+
+
+class FakeHypervisorStats(object):
+ """Fake one or more hypervisor stats."""
+
+ @staticmethod
+ def create_one_hypervisor_stats(attrs=None):
+ """Create a fake hypervisor stats.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :return:
+ A FakeResource object, with count, current_workload, and so on
+ """
+ attrs = attrs or {}
+
+ # Set default attributes.
+ stats_info = {
+ 'count': 2,
+ 'current_workload': 0,
+ 'disk_available_least': 50,
+ 'free_disk_gb': 100,
+ 'free_ram_mb': 23000,
+ 'local_gb': 100,
+ 'local_gb_used': 0,
+ 'memory_mb': 23800,
+ 'memory_mb_used': 1400,
+ 'running_vms': 3,
+ 'vcpus': 8,
+ 'vcpus_used': 3,
+ }
+
+ # Overwrite default attributes.
+ stats_info.update(attrs)
+
+ # Set default method.
+ hypervisor_stats_method = {'to_dict': stats_info}
+
+ hypervisor_stats = fakes.FakeResource(
+ info=copy.deepcopy(stats_info),
+ methods=copy.deepcopy(hypervisor_stats_method),
+ loaded=True)
+ return hypervisor_stats
+
+ @staticmethod
+ def create_hypervisors_stats(attrs=None, count=2):
+ """Create multiple fake hypervisors stats.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :param int count:
+ The number of hypervisors to fake
+ :return:
+ A list of FakeResource objects faking the hypervisors
+ """
+ hypervisors = []
+ for i in range(0, count):
+ hypervisors.append(
+ FakeHypervisorStats.create_one_hypervisor_stats(attrs))
+
+ return hypervisors
+
+
+class FakeSecurityGroup(object):
+ """Fake one or more security groups."""
+
+ @staticmethod
+ def create_one_security_group(attrs=None):
+ """Create a fake security group.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :return:
+ A FakeResource object, with id, name, etc.
+ """
+ attrs = attrs or {}
+
+ # Set default attributes.
+ security_group_attrs = {
+ 'id': 'security-group-id-' + uuid.uuid4().hex,
+ 'name': 'security-group-name-' + uuid.uuid4().hex,
+ 'description': 'security-group-description-' + uuid.uuid4().hex,
+ 'tenant_id': 'project-id-' + uuid.uuid4().hex,
+ 'rules': [],
+ }
+
+ # Overwrite default attributes.
+ security_group_attrs.update(attrs)
+
+ security_group = fakes.FakeResource(
+ info=copy.deepcopy(security_group_attrs),
+ loaded=True)
+ return security_group
+
+ @staticmethod
+ def create_security_groups(attrs=None, count=2):
+ """Create multiple fake security groups.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :param int count:
+ The number of security groups to fake
+ :return:
+ A list of FakeResource objects faking the security groups
+ """
+ security_groups = []
+ for i in range(0, count):
+ security_groups.append(
+ FakeSecurityGroup.create_one_security_group(attrs))
+
+ return security_groups
+
+ @staticmethod
+ def get_security_groups(security_groups=None, count=2):
+ """Get an iterable MagicMock object with a list of faked security groups.
+
+ If security groups list is provided, then initialize the Mock object
+ with the list. Otherwise create one.
+
+ :param List security groups:
+ A list of FakeResource objects faking security groups
+ :param int count:
+ The number of security groups to fake
+ :return:
+ An iterable Mock object with side_effect set to a list of faked
+ security groups
+ """
+ if security_groups is None:
+ security_groups = FakeSecurityGroup.create_security_groups(count)
+ return mock.MagicMock(side_effect=security_groups)
+
+
+class FakeSecurityGroupRule(object):
+ """Fake one or more security group rules."""
+
+ @staticmethod
+ def create_one_security_group_rule(attrs=None):
+ """Create a fake security group rule.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :return:
+ A FakeResource object, with id, etc.
+ """
+ attrs = attrs or {}
+
+ # Set default attributes.
+ security_group_rule_attrs = {
+ 'from_port': 0,
+ 'group': {},
+ 'id': 'security-group-rule-id-' + uuid.uuid4().hex,
+ 'ip_protocol': 'tcp',
+ 'ip_range': {'cidr': '0.0.0.0/0'},
+ 'parent_group_id': 'security-group-id-' + uuid.uuid4().hex,
+ 'to_port': 0,
+ }
+
+ # Overwrite default attributes.
+ security_group_rule_attrs.update(attrs)
+
+ security_group_rule = fakes.FakeResource(
+ info=copy.deepcopy(security_group_rule_attrs),
+ loaded=True)
+ return security_group_rule
+
+ @staticmethod
+ def create_security_group_rules(attrs=None, count=2):
+ """Create multiple fake security group rules.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :param int count:
+ The number of security group rules to fake
+ :return:
+ A list of FakeResource objects faking the security group rules
+ """
+ security_group_rules = []
+ for i in range(0, count):
+ security_group_rules.append(
+ FakeSecurityGroupRule.create_one_security_group_rule(attrs))
+
+ return security_group_rules
+
+
+class FakeServer(object):
+ """Fake one or more compute servers."""
+
+ @staticmethod
+ def create_one_server(attrs=None, methods=None):
+ """Create a fake server.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :param Dictionary methods:
+ A dictionary with all methods
+ :return:
+ A FakeResource object, with id, name, metadata, and so on
+ """
+ attrs = attrs or {}
+ methods = methods or {}
+
+ # Set default attributes.
+ server_info = {
+ 'id': 'server-id-' + uuid.uuid4().hex,
+ 'name': 'server-name-' + uuid.uuid4().hex,
+ 'metadata': {},
+ 'image': {
+ 'id': 'image-id-' + uuid.uuid4().hex,
+ },
+ 'flavor': {
+ 'id': 'flavor-id-' + uuid.uuid4().hex,
+ },
+ 'OS-EXT-STS:power_state': 1,
+ }
+
+ # Overwrite default attributes.
+ server_info.update(attrs)
+
+ server = fakes.FakeResource(info=copy.deepcopy(server_info),
+ methods=methods,
+ loaded=True)
+ return server
+
+ @staticmethod
+ def create_servers(attrs=None, methods=None, count=2):
+ """Create multiple fake servers.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :param Dictionary methods:
+ A dictionary with all methods
+ :param int count:
+ The number of servers to fake
+ :return:
+ A list of FakeResource objects faking the servers
+ """
+ servers = []
+ for i in range(0, count):
+ servers.append(FakeServer.create_one_server(attrs, methods))
+
+ return servers
+
+ @staticmethod
+ def get_servers(servers=None, count=2):
+ """Get an iterable MagicMock object with a list of faked servers.
+
+ If servers list is provided, then initialize the Mock object with the
+ list. Otherwise create one.
+
+ :param List servers:
+ A list of FakeResource objects faking servers
+ :param int count:
+ The number of servers to fake
+ :return:
+ An iterable Mock object with side_effect set to a list of faked
+ servers
+ """
+ if servers is None:
+ servers = FakeServer.create_servers(count)
+ return mock.MagicMock(side_effect=servers)
+
+
+class FakeService(object):
+ """Fake one or more services."""
+
+ @staticmethod
+ def create_one_service(attrs=None):
+ """Create a fake service.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :return:
+ A FakeResource object, with id, host, binary, and so on
+ """
+ attrs = attrs or {}
+
+ # Set default attributes.
+ service_info = {
+ 'id': 'id-' + uuid.uuid4().hex,
+ 'host': 'host-' + uuid.uuid4().hex,
+ 'binary': 'binary-' + uuid.uuid4().hex,
+ 'status': 'enabled',
+ 'zone': 'zone-' + uuid.uuid4().hex,
+ 'state': 'state-' + uuid.uuid4().hex,
+ 'updated_at': 'time-' + uuid.uuid4().hex,
+ 'disabled_reason': 'earthquake',
+ }
+
+ # Overwrite default attributes.
+ service_info.update(attrs)
+
+ service = fakes.FakeResource(info=copy.deepcopy(service_info),
+ loaded=True)
+
+ return service
+
+ @staticmethod
+ def create_services(attrs=None, count=2):
+ """Create multiple fake services.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :param int count:
+ The number of services to fake
+ :return:
+ A list of FakeResource objects faking the services
+ """
+ services = []
+ for i in range(0, count):
+ services.append(FakeService.create_one_service(attrs))
+
+ return services
+
+
+class FakeFlavor(object):
+ """Fake one or more flavors."""
+
+ @staticmethod
+ def create_one_flavor(attrs=None):
+ """Create a fake flavor.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :return:
+ A FakeResource object, with id, name, ram, vcpus, and so on
+ """
+ attrs = attrs or {}
+
+ # Set default attributes.
+ flavor_info = {
+ 'id': 'flavor-id-' + uuid.uuid4().hex,
+ 'name': 'flavor-name-' + uuid.uuid4().hex,
+ 'ram': 8192,
+ 'vcpus': 4,
+ 'disk': 128,
+ 'swap': 0,
+ 'rxtx_factor': 1.0,
+ 'OS-FLV-DISABLED:disabled': False,
+ 'os-flavor-access:is_public': True,
+ 'OS-FLV-EXT-DATA:ephemeral': 0,
+ 'properties': {'property': 'value'},
+ }
+
+ # Overwrite default attributes.
+ flavor_info.update(attrs)
+
+ # Set default methods.
+ flavor_methods = {
+ 'set_keys': None,
+ 'unset_keys': None,
+ 'get_keys': {'property': 'value'},
+ }
+
+ flavor = fakes.FakeResource(info=copy.deepcopy(flavor_info),
+ methods=flavor_methods,
+ loaded=True)
+
+ # Set attributes with special mappings in nova client.
+ flavor.disabled = flavor_info['OS-FLV-DISABLED:disabled']
+ flavor.is_public = flavor_info['os-flavor-access:is_public']
+ flavor.ephemeral = flavor_info['OS-FLV-EXT-DATA:ephemeral']
+
+ return flavor
+
+ @staticmethod
+ def create_flavors(attrs=None, count=2):
+ """Create multiple fake flavors.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :param int count:
+ The number of flavors to fake
+ :return:
+ A list of FakeResource objects faking the flavors
+ """
+ flavors = []
+ for i in range(0, count):
+ flavors.append(FakeFlavor.create_one_flavor(attrs))
+
+ return flavors
+
+ @staticmethod
+ def get_flavors(flavors=None, count=2):
+ """Get an iterable MagicMock object with a list of faked flavors.
+
+ If flavors list is provided, then initialize the Mock object with the
+ list. Otherwise create one.
+
+ :param List flavors:
+ A list of FakeResource objects faking flavors
+ :param int count:
+ The number of flavors to fake
+ :return:
+ An iterable Mock object with side_effect set to a list of faked
+ flavors
+ """
+ if flavors is None:
+ flavors = FakeFlavor.create_flavors(count)
+ return mock.MagicMock(side_effect=flavors)
+
+
+class FakeFlavorAccess(object):
+ """Fake one or more flavor accesses."""
+
+ @staticmethod
+ def create_one_flavor_access(attrs=None):
+ """Create a fake flavor access.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :return:
+ A FakeResource object, with flavor_id, tenat_id
+ """
+ attrs = attrs or {}
+
+ # Set default attributes.
+ flavor_access_info = {
+ 'flavor_id': 'flavor-id-' + uuid.uuid4().hex,
+ 'tenant_id': 'tenant-id-' + uuid.uuid4().hex,
+ }
+
+ # Overwrite default attributes.
+ flavor_access_info.update(attrs)
+
+ flavor_access = fakes.FakeResource(
+ info=copy.deepcopy(flavor_access_info), loaded=True)
+
+ return flavor_access
+
+
+class FakeKeypair(object):
+ """Fake one or more keypairs."""
+
+ @staticmethod
+ def create_one_keypair(attrs=None, no_pri=False):
+ """Create a fake keypair
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :return:
+ A FakeResource object, name, fingerprint, and so on
+ """
+ attrs = attrs or {}
+
+ # Set default attributes.
+ keypair_info = {
+ 'name': 'keypair-name-' + uuid.uuid4().hex,
+ 'fingerprint': 'dummy',
+ 'public_key': 'dummy',
+ 'user_id': 'user'
+ }
+ if not no_pri:
+ keypair_info['private_key'] = 'private_key'
+
+ # Overwrite default attributes.
+ keypair_info.update(attrs)
+
+ keypair = fakes.FakeResource(info=copy.deepcopy(keypair_info),
+ loaded=True)
+
+ return keypair
+
+ @staticmethod
+ def create_keypairs(attrs=None, count=2):
+ """Create multiple fake keypairs.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :param int count:
+ The number of keypairs to fake
+ :return:
+ A list of FakeResource objects faking the keypairs
+ """
+
+ keypairs = []
+ for i in range(0, count):
+ keypairs.append(FakeKeypair.create_one_keypair(attrs))
+
+ return keypairs
+
+ @staticmethod
+ def get_keypairs(keypairs=None, count=2):
+ """Get an iterable MagicMock object with a list of faked keypairs.
+
+ If keypairs list is provided, then initialize the Mock object with the
+ list. Otherwise create one.
+
+ :param List keypairs:
+ A list of FakeResource objects faking keypairs
+ :param int count:
+ The number of keypairs to fake
+ :return:
+ An iterable Mock object with side_effect set to a list of faked
+ keypairs
+ """
+ if keypairs is None:
+ keypairs = FakeKeypair.create_keypairs(count)
+ return mock.MagicMock(side_effect=keypairs)
+
+
+class FakeAvailabilityZone(object):
+ """Fake one or more compute availability zones (AZs)."""
+
+ @staticmethod
+ def create_one_availability_zone(attrs=None):
+ """Create a fake AZ.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :return:
+ A FakeResource object with zoneName, zoneState, etc.
+ """
+ attrs = attrs or {}
+
+ # Set default attributes.
+ host_name = uuid.uuid4().hex
+ service_name = uuid.uuid4().hex
+ service_updated_at = uuid.uuid4().hex
+ availability_zone = {
+ 'zoneName': uuid.uuid4().hex,
+ 'zoneState': {'available': True},
+ 'hosts': {host_name: {service_name: {
+ 'available': True,
+ 'active': True,
+ 'updated_at': service_updated_at,
+ }}},
+ }
+
+ # Overwrite default attributes.
+ availability_zone.update(attrs)
+
+ availability_zone = fakes.FakeResource(
+ info=copy.deepcopy(availability_zone),
+ loaded=True)
+ return availability_zone
+
+ @staticmethod
+ def create_availability_zones(attrs=None, count=2):
+ """Create multiple fake AZs.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :param int count:
+ The number of AZs to fake
+ :return:
+ A list of FakeResource objects faking the AZs
+ """
+ availability_zones = []
+ for i in range(0, count):
+ availability_zone = \
+ FakeAvailabilityZone.create_one_availability_zone(attrs)
+ availability_zones.append(availability_zone)
+
+ return availability_zones
+
+
+class FakeFloatingIP(object):
+ """Fake one or more floating ip."""
+
+ @staticmethod
+ def create_one_floating_ip(attrs=None):
+ """Create a fake floating ip.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :return:
+ A FakeResource object, with id, ip, and so on
+ """
+ attrs = attrs or {}
+
+ # Set default attributes.
+ floating_ip_attrs = {
+ 'id': 'floating-ip-id-' + uuid.uuid4().hex,
+ 'ip': '1.0.9.0',
+ 'fixed_ip': '2.0.9.0',
+ 'instance_id': 'server-id-' + uuid.uuid4().hex,
+ 'pool': 'public',
+ }
+
+ # Overwrite default attributes.
+ floating_ip_attrs.update(attrs)
+
+ floating_ip = fakes.FakeResource(
+ info=copy.deepcopy(floating_ip_attrs),
+ loaded=True)
+
+ return floating_ip
+
+ @staticmethod
+ def create_floating_ips(attrs=None, count=2):
+ """Create multiple fake floating ips.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :param int count:
+ The number of floating ips to fake
+ :return:
+ A list of FakeResource objects faking the floating ips
+ """
+ floating_ips = []
+ for i in range(0, count):
+ floating_ips.append(FakeFloatingIP.create_one_floating_ip(attrs))
+ return floating_ips
+
+ @staticmethod
+ def get_floating_ips(floating_ips=None, count=2):
+ """Get an iterable MagicMock object with a list of faked floating ips.
+
+ If floating_ips list is provided, then initialize the Mock object
+ with the list. Otherwise create one.
+
+ :param List floating ips:
+ A list of FakeResource objects faking floating ips
+ :param int count:
+ The number of floating ips to fake
+ :return:
+ An iterable Mock object with side_effect set to a list of faked
+ floating ips
+ """
+ if floating_ips is None:
+ floating_ips = FakeFloatingIP.create_floating_ips(count)
+ return mock.MagicMock(side_effect=floating_ips)
+
+
+class FakeFloatingIPPool(object):
+ """Fake one or more floating ip pools."""
+
+ @staticmethod
+ def create_one_floating_ip_pool(attrs=None):
+ """Create a fake floating ip pool.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :return:
+ A FakeResource object, with name, etc
+ """
+ if attrs is None:
+ attrs = {}
+
+ # Set default attributes.
+ floating_ip_pool_attrs = {
+ 'name': 'floating-ip-pool-name-' + uuid.uuid4().hex,
+ }
+
+ # Overwrite default attributes.
+ floating_ip_pool_attrs.update(attrs)
+
+ floating_ip_pool = fakes.FakeResource(
+ info=copy.deepcopy(floating_ip_pool_attrs),
+ loaded=True)
+
+ return floating_ip_pool
+
+ @staticmethod
+ def create_floating_ip_pools(attrs=None, count=2):
+ """Create multiple fake floating ip pools.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :param int count:
+ The number of floating ip pools to fake
+ :return:
+ A list of FakeResource objects faking the floating ip pools
+ """
+ floating_ip_pools = []
+ for i in range(0, count):
+ floating_ip_pools.append(
+ FakeFloatingIPPool.create_one_floating_ip_pool(attrs)
+ )
+ return floating_ip_pools
+
+
+class FakeNetwork(object):
+ """Fake one or more networks."""
+
+ @staticmethod
+ def create_one_network(attrs=None):
+ """Create a fake network.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :return:
+ A FakeResource object, with id, label, cidr and so on
+ """
+ attrs = attrs or {}
+
+ # Set default attributes.
+ network_attrs = {
+ 'bridge': 'br100',
+ 'bridge_interface': None,
+ 'broadcast': '10.0.0.255',
+ 'cidr': '10.0.0.0/24',
+ 'cidr_v6': None,
+ 'created_at': '2016-02-11T11:17:37.000000',
+ 'deleted': False,
+ 'deleted_at': None,
+ 'dhcp_server': '10.0.0.1',
+ 'dhcp_start': '10.0.0.2',
+ 'dns1': '8.8.4.4',
+ 'dns2': None,
+ 'enable_dhcp': True,
+ 'gateway': '10.0.0.1',
+ 'gateway_v6': None,
+ 'host': None,
+ 'id': 'network-id-' + uuid.uuid4().hex,
+ 'injected': False,
+ 'label': 'network-label-' + uuid.uuid4().hex,
+ 'mtu': None,
+ 'multi_host': False,
+ 'netmask': '255.255.255.0',
+ 'netmask_v6': None,
+ 'priority': None,
+ 'project_id': 'project-id-' + uuid.uuid4().hex,
+ 'rxtx_base': None,
+ 'share_address': False,
+ 'updated_at': None,
+ 'vlan': None,
+ 'vpn_private_address': None,
+ 'vpn_public_address': None,
+ 'vpn_public_port': None,
+ }
+
+ # Overwrite default attributes.
+ network_attrs.update(attrs)
+
+ network = fakes.FakeResource(info=copy.deepcopy(network_attrs),
+ loaded=True)
+
+ return network
+
+ @staticmethod
+ def create_networks(attrs=None, count=2):
+ """Create multiple fake networks.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :param int count:
+ The number of networks to fake
+ :return:
+ A list of FakeResource objects faking the networks
+ """
+ networks = []
+ for i in range(0, count):
+ networks.append(FakeNetwork.create_one_network(attrs))
+
+ return networks
+
+ @staticmethod
+ def get_networks(networks=None, count=2):
+ """Get an iterable MagicMock object with a list of faked networks.
+
+ If networks list is provided, then initialize the Mock object with the
+ list. Otherwise create one.
+
+ :param List networks:
+ A list of FakeResource objects faking networks
+ :param int count:
+ The number of networks to fake
+ :return:
+ An iterable Mock object with side_effect set to a list of faked
+ networks
+ """
+ if networks is None:
+ networks = FakeNetwork.create_networks(count=count)
+ return mock.Mock(side_effect=networks)
+
+
+class FakeHost(object):
+ """Fake one host."""
+
+ @staticmethod
+ def create_one_host(attrs=None):
+ """Create a fake host.
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :return:
+ A FakeResource object, with uuid and other attributes
+ """
+ attrs = attrs or {}
+
+ # Set default attributes.
+ host_info = {
+ "service_id": 1,
+ "host": "host1",
+ "uuid": 'host-id-' + uuid.uuid4().hex,
+ "vcpus": 10,
+ "memory_mb": 100,
+ "local_gb": 100,
+ "vcpus_used": 5,
+ "memory_mb_used": 50,
+ "local_gb_used": 10,
+ "hypervisor_type": "xen",
+ "hypervisor_version": 1,
+ "hypervisor_hostname": "devstack1",
+ "free_ram_mb": 50,
+ "free_disk_gb": 50,
+ "current_workload": 10,
+ "running_vms": 1,
+ "cpu_info": "",
+ "disk_available_least": 1,
+ "host_ip": "10.10.10.10",
+ "supported_instances": "",
+ "metrics": "",
+ "pci_stats": "",
+ "extra_resources": "",
+ "stats": "",
+ "numa_topology": "",
+ "ram_allocation_ratio": 1.0,
+ "cpu_allocation_ratio": 1.0,
+ "zone": 'zone-' + uuid.uuid4().hex,
+ "host_name": 'name-' + uuid.uuid4().hex,
+ "service": 'service-' + uuid.uuid4().hex,
+ "cpu": 4,
+ "disk_gb": 100,
+ 'project': 'project-' + uuid.uuid4().hex,
+ }
+ host_info.update(attrs)
+ host = fakes.FakeResource(
+ info=copy.deepcopy(host_info),
+ loaded=True)
+ return host
+
+
+class FakeServerGroup(object):
+ """Fake one server group"""
+
+ @staticmethod
+ def create_one_server_group(attrs=None):
+ """Create a fake server group
+
+ :param Dictionary attrs:
+ A dictionary with all attributes
+ :return:
+ A FakeResource object, with id and other attributes
+ """
+ if attrs is None:
+ attrs = {}
+
+ # Set default attributes.
+ server_group_info = {
+ 'id': 'server-group-id-' + uuid.uuid4().hex,
+ 'members': [],
+ 'metadata': {},
+ 'name': 'server-group-name-' + uuid.uuid4().hex,
+ 'policies': [],
+ 'project_id': 'server-group-project-id-' + uuid.uuid4().hex,
+ 'user_id': 'server-group-user-id-' + uuid.uuid4().hex,
+ }
+
+ # Overwrite default attributes.
+ server_group_info.update(attrs)
+
+ server_group = fakes.FakeResource(
+ info=copy.deepcopy(server_group_info),
+ loaded=True)
+ return server_group
diff --git a/openstackclient/tests/unit/compute/v2/test_agent.py b/openstackclient/tests/unit/compute/v2/test_agent.py
new file mode 100644
index 00000000..169940e2
--- /dev/null
+++ b/openstackclient/tests/unit/compute/v2/test_agent.py
@@ -0,0 +1,320 @@
+# Copyright 2016 Easystack. 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 mock
+from mock import call
+
+from osc_lib import exceptions
+
+from openstackclient.compute.v2 import agent
+from openstackclient.tests.unit.compute.v2 import fakes as compute_fakes
+from openstackclient.tests.unit import utils as tests_utils
+
+
+class TestAgent(compute_fakes.TestComputev2):
+
+ attr = {}
+ attr['agent_id'] = 1
+ fake_agent = compute_fakes.FakeAgent.create_one_agent(attr)
+
+ columns = (
+ 'agent_id',
+ 'architecture',
+ 'hypervisor',
+ 'md5hash',
+ 'os',
+ 'url',
+ 'version',
+ )
+
+ data = (
+ fake_agent.agent_id,
+ fake_agent.architecture,
+ fake_agent.hypervisor,
+ fake_agent.md5hash,
+ fake_agent.os,
+ fake_agent.url,
+ fake_agent.version,
+ )
+
+ def setUp(self):
+ super(TestAgent, self).setUp()
+
+ self.agents_mock = self.app.client_manager.compute.agents
+ self.agents_mock.reset_mock()
+
+
+class TestAgentCreate(TestAgent):
+
+ def setUp(self):
+ super(TestAgentCreate, self).setUp()
+
+ self.agents_mock.create.return_value = self.fake_agent
+ self.cmd = agent.CreateAgent(self.app, None)
+
+ def test_agent_create(self):
+ arglist = [
+ self.fake_agent.os,
+ self.fake_agent.architecture,
+ self.fake_agent.version,
+ self.fake_agent.url,
+ self.fake_agent.md5hash,
+ self.fake_agent.hypervisor,
+ ]
+
+ verifylist = [
+ ('os', self.fake_agent.os),
+ ('architecture', self.fake_agent.architecture),
+ ('version', self.fake_agent.version),
+ ('url', self.fake_agent.url),
+ ('md5hash', self.fake_agent.md5hash),
+ ('hypervisor', self.fake_agent.hypervisor),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ columns, data = self.cmd.take_action(parsed_args)
+ self.agents_mock.create.assert_called_with(parsed_args.os,
+ parsed_args.architecture,
+ parsed_args.version,
+ parsed_args.url,
+ parsed_args.md5hash,
+ parsed_args.hypervisor)
+
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(self.data, data)
+
+
+class TestAgentDelete(TestAgent):
+
+ fake_agents = compute_fakes.FakeAgent.create_agents(count=2)
+
+ def setUp(self):
+ super(TestAgentDelete, self).setUp()
+
+ self.agents_mock.get.return_value = self.fake_agents
+ self.cmd = agent.DeleteAgent(self.app, None)
+
+ def test_delete_one_agent(self):
+ arglist = [
+ self.fake_agents[0].agent_id
+ ]
+
+ verifylist = [
+ ('id', [self.fake_agents[0].agent_id]),
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ result = self.cmd.take_action(parsed_args)
+ self.agents_mock.delete.assert_called_with(
+ self.fake_agents[0].agent_id)
+ self.assertIsNone(result)
+
+ def test_delete_multiple_agents(self):
+ arglist = []
+ for n in self.fake_agents:
+ arglist.append(n.agent_id)
+ verifylist = [
+ ('id', arglist),
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ result = self.cmd.take_action(parsed_args)
+
+ calls = []
+ for n in self.fake_agents:
+ calls.append(call(n.agent_id))
+ self.agents_mock.delete.assert_has_calls(calls)
+ self.assertIsNone(result)
+
+ def test_delete_multiple_agents_exception(self):
+ arglist = [
+ self.fake_agents[0].agent_id,
+ self.fake_agents[1].agent_id,
+ 'x-y-z',
+ ]
+ verifylist = [
+ ('id', arglist),
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ ret_delete = [
+ None,
+ None,
+ exceptions.NotFound('404')
+ ]
+ self.agents_mock.delete = mock.Mock(side_effect=ret_delete)
+
+ self.assertRaises(exceptions.CommandError, self.cmd.take_action,
+ parsed_args)
+ calls = [
+ call(self.fake_agents[0].agent_id),
+ call(self.fake_agents[1].agent_id),
+ ]
+ self.agents_mock.delete.assert_has_calls(calls)
+
+ def test_agent_delete_no_input(self):
+ arglist = []
+ verifylist = None
+ self.assertRaises(tests_utils.ParserException,
+ self.check_parser,
+ self.cmd,
+ arglist,
+ verifylist)
+
+
+class TestAgentList(TestAgent):
+
+ agents = compute_fakes.FakeAgent.create_agents(count=3)
+ list_columns = (
+ "Agent ID",
+ "Hypervisor",
+ "OS",
+ "Architecture",
+ "Version",
+ "Md5Hash",
+ "URL",
+ )
+
+ list_data = []
+ for _agent in agents:
+ list_data.append((
+ _agent.agent_id,
+ _agent.hypervisor,
+ _agent.os,
+ _agent.architecture,
+ _agent.version,
+ _agent.md5hash,
+ _agent.url,
+ ))
+
+ def setUp(self):
+
+ super(TestAgentList, self).setUp()
+
+ self.agents_mock.list.return_value = self.agents
+ self.cmd = agent.ListAgent(self.app, None)
+
+ def test_agent_list(self):
+
+ arglist = []
+ verifylist = []
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ columns, data = self.cmd.take_action(parsed_args)
+
+ self.assertEqual(self.list_columns, columns)
+ self.assertEqual(self.list_data, list(data))
+
+ def test_agent_list_with_hypervisor(self):
+
+ arglist = [
+ '--hypervisor',
+ 'hypervisor',
+ ]
+ verifylist = [
+ ('hypervisor', 'hypervisor'),
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ columns, data = self.cmd.take_action(parsed_args)
+
+ self.assertEqual(self.list_columns, columns)
+ self.assertEqual(self.list_data, list(data))
+
+
+class TestAgentSet(TestAgent):
+
+ def setUp(self):
+ super(TestAgentSet, self).setUp()
+
+ self.agents_mock.update.return_value = self.fake_agent
+ self.agents_mock.list.return_value = [self.fake_agent]
+ self.cmd = agent.SetAgent(self.app, None)
+
+ def test_agent_set_nothing(self):
+ arglist = [
+ '1',
+ ]
+ verifylist = [
+ ('id', '1'),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ result = self.cmd.take_action(parsed_args)
+
+ self.agents_mock.update.assert_called_with(parsed_args.id,
+ self.fake_agent.version,
+ self.fake_agent.url,
+ self.fake_agent.md5hash)
+ self.assertIsNone(result)
+
+ def test_agent_set_version(self):
+ arglist = [
+ '1',
+ '--agent-version', 'new-version',
+ ]
+
+ verifylist = [
+ ('id', '1'),
+ ('version', 'new-version'),
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ result = self.cmd.take_action(parsed_args)
+
+ self.agents_mock.update.assert_called_with(parsed_args.id,
+ parsed_args.version,
+ self.fake_agent.url,
+ self.fake_agent.md5hash)
+ self.assertIsNone(result)
+
+ def test_agent_set_url(self):
+ arglist = [
+ '1',
+ '--url', 'new-url',
+ ]
+
+ verifylist = [
+ ('id', '1'),
+ ('url', 'new-url'),
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ result = self.cmd.take_action(parsed_args)
+
+ self.agents_mock.update.assert_called_with(parsed_args.id,
+ self.fake_agent.version,
+ parsed_args.url,
+ self.fake_agent.md5hash)
+ self.assertIsNone(result)
+
+ def test_agent_set_md5hash(self):
+ arglist = [
+ '1',
+ '--md5hash', 'new-md5hash',
+ ]
+
+ verifylist = [
+ ('id', '1'),
+ ('md5hash', 'new-md5hash'),
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ result = self.cmd.take_action(parsed_args)
+
+ self.agents_mock.update.assert_called_with(parsed_args.id,
+ self.fake_agent.version,
+ self.fake_agent.url,
+ parsed_args.md5hash)
+ self.assertIsNone(result)
diff --git a/openstackclient/tests/unit/compute/v2/test_aggregate.py b/openstackclient/tests/unit/compute/v2/test_aggregate.py
new file mode 100644
index 00000000..c636d3de
--- /dev/null
+++ b/openstackclient/tests/unit/compute/v2/test_aggregate.py
@@ -0,0 +1,447 @@
+# Copyright 2016 Huawei, Inc. 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 mock
+from mock import call
+
+from osc_lib import exceptions
+from osc_lib import utils
+
+from openstackclient.compute.v2 import aggregate
+from openstackclient.tests.unit.compute.v2 import fakes as compute_fakes
+from openstackclient.tests.unit import utils as tests_utils
+
+
+class TestAggregate(compute_fakes.TestComputev2):
+
+ fake_ag = compute_fakes.FakeAggregate.create_one_aggregate()
+
+ columns = (
+ 'availability_zone',
+ 'hosts',
+ 'id',
+ 'metadata',
+ 'name',
+ )
+
+ data = (
+ fake_ag.availability_zone,
+ fake_ag.hosts,
+ fake_ag.id,
+ fake_ag.metadata,
+ fake_ag.name,
+ )
+
+ def setUp(self):
+ super(TestAggregate, self).setUp()
+
+ # Get a shortcut to the AggregateManager Mock
+ self.aggregate_mock = self.app.client_manager.compute.aggregates
+ self.aggregate_mock.reset_mock()
+
+
+class TestAggregateAddHost(TestAggregate):
+
+ def setUp(self):
+ super(TestAggregateAddHost, self).setUp()
+
+ self.aggregate_mock.get.return_value = self.fake_ag
+ self.aggregate_mock.add_host.return_value = self.fake_ag
+ self.cmd = aggregate.AddAggregateHost(self.app, None)
+
+ def test_aggregate_add_host(self):
+ arglist = [
+ 'ag1',
+ 'host1',
+ ]
+ verifylist = [
+ ('aggregate', 'ag1'),
+ ('host', 'host1'),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ columns, data = self.cmd.take_action(parsed_args)
+ self.aggregate_mock.get.assert_called_once_with(parsed_args.aggregate)
+ self.aggregate_mock.add_host.assert_called_once_with(self.fake_ag,
+ parsed_args.host)
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(self.data, data)
+
+
+class TestAggregateCreate(TestAggregate):
+
+ def setUp(self):
+ super(TestAggregateCreate, self).setUp()
+
+ self.aggregate_mock.create.return_value = self.fake_ag
+ self.aggregate_mock.set_metadata.return_value = self.fake_ag
+ self.cmd = aggregate.CreateAggregate(self.app, None)
+
+ def test_aggregate_create(self):
+ arglist = [
+ 'ag1',
+ ]
+ verifylist = [
+ ('name', 'ag1'),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ columns, data = self.cmd.take_action(parsed_args)
+ self.aggregate_mock.create.assert_called_once_with(parsed_args.name,
+ None)
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(self.data, data)
+
+ def test_aggregate_create_with_zone(self):
+ arglist = [
+ '--zone', 'zone1',
+ 'ag1',
+ ]
+ verifylist = [
+ ('zone', 'zone1'),
+ ('name', 'ag1'),
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ columns, data = self.cmd.take_action(parsed_args)
+ self.aggregate_mock.create.assert_called_once_with(parsed_args.name,
+ parsed_args.zone)
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(self.data, data)
+
+ def test_aggregate_create_with_property(self):
+ arglist = [
+ '--property', 'key1=value1',
+ '--property', 'key2=value2',
+ 'ag1',
+ ]
+ verifylist = [
+ ('property', {'key1': 'value1', 'key2': 'value2'}),
+ ('name', 'ag1'),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ columns, data = self.cmd.take_action(parsed_args)
+ self.aggregate_mock.create.assert_called_once_with(parsed_args.name,
+ None)
+ self.aggregate_mock.set_metadata.assert_called_once_with(
+ self.fake_ag, parsed_args.property)
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(self.data, data)
+
+
+class TestAggregateDelete(TestAggregate):
+
+ fake_ags = compute_fakes.FakeAggregate.create_aggregates(count=2)
+
+ def setUp(self):
+ super(TestAggregateDelete, self).setUp()
+
+ self.aggregate_mock.get = (
+ compute_fakes.FakeAggregate.get_aggregates(self.fake_ags))
+ self.cmd = aggregate.DeleteAggregate(self.app, None)
+
+ def test_aggregate_delete(self):
+ arglist = [
+ self.fake_ags[0].id
+ ]
+ verifylist = [
+ ('aggregate', [self.fake_ags[0].id]),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ result = self.cmd.take_action(parsed_args)
+ self.aggregate_mock.get.assert_called_once_with(self.fake_ags[0].id)
+ self.aggregate_mock.delete.assert_called_once_with(self.fake_ags[0].id)
+ self.assertIsNone(result)
+
+ def test_delete_multiple_aggregates(self):
+ arglist = []
+ for a in self.fake_ags:
+ arglist.append(a.id)
+ verifylist = [
+ ('aggregate', arglist),
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ result = self.cmd.take_action(parsed_args)
+
+ calls = []
+ for a in self.fake_ags:
+ calls.append(call(a.id))
+ self.aggregate_mock.delete.assert_has_calls(calls)
+ self.assertIsNone(result)
+
+ def test_delete_multiple_agggregates_with_exception(self):
+ arglist = [
+ self.fake_ags[0].id,
+ 'unexist_aggregate',
+ ]
+ verifylist = [
+ ('aggregate', arglist),
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ find_mock_result = [self.fake_ags[0], exceptions.CommandError]
+ with mock.patch.object(utils, 'find_resource',
+ side_effect=find_mock_result) as find_mock:
+ try:
+ self.cmd.take_action(parsed_args)
+ self.fail('CommandError should be raised.')
+ except exceptions.CommandError as e:
+ self.assertEqual('1 of 2 aggregates failed to delete.',
+ str(e))
+
+ find_mock.assert_any_call(self.aggregate_mock, self.fake_ags[0].id)
+ find_mock.assert_any_call(self.aggregate_mock, 'unexist_aggregate')
+
+ self.assertEqual(2, find_mock.call_count)
+ self.aggregate_mock.delete.assert_called_once_with(
+ self.fake_ags[0].id
+ )
+
+
+class TestAggregateList(TestAggregate):
+
+ list_columns = (
+ "ID",
+ "Name",
+ "Availability Zone",
+ )
+
+ list_columns_long = (
+ "ID",
+ "Name",
+ "Availability Zone",
+ "Properties",
+ )
+
+ list_data = ((
+ TestAggregate.fake_ag.id,
+ TestAggregate.fake_ag.name,
+ TestAggregate.fake_ag.availability_zone,
+ ), )
+
+ list_data_long = ((
+ TestAggregate.fake_ag.id,
+ TestAggregate.fake_ag.name,
+ TestAggregate.fake_ag.availability_zone,
+ {},
+ ), )
+
+ def setUp(self):
+ super(TestAggregateList, self).setUp()
+
+ self.aggregate_mock.list.return_value = [self.fake_ag]
+ self.cmd = aggregate.ListAggregate(self.app, None)
+
+ def test_aggregate_list(self):
+
+ parsed_args = self.check_parser(self.cmd, [], [])
+ columns, data = self.cmd.take_action(parsed_args)
+
+ self.assertEqual(self.list_columns, columns)
+ self.assertEqual(self.list_data, tuple(data))
+
+ def test_aggregate_list_with_long(self):
+ arglist = [
+ '--long',
+ ]
+ vertifylist = [
+ ('long', True),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, vertifylist)
+ columns, data = self.cmd.take_action(parsed_args)
+
+ self.assertEqual(self.list_columns_long, columns)
+ self.assertEqual(self.list_data_long, tuple(data))
+
+
+class TestAggregateRemoveHost(TestAggregate):
+
+ def setUp(self):
+ super(TestAggregateRemoveHost, self).setUp()
+
+ self.aggregate_mock.get.return_value = self.fake_ag
+ self.aggregate_mock.remove_host.return_value = self.fake_ag
+ self.cmd = aggregate.RemoveAggregateHost(self.app, None)
+
+ def test_aggregate_add_host(self):
+ arglist = [
+ 'ag1',
+ 'host1',
+ ]
+ verifylist = [
+ ('aggregate', 'ag1'),
+ ('host', 'host1'),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ columns, data = self.cmd.take_action(parsed_args)
+ self.aggregate_mock.get.assert_called_once_with(parsed_args.aggregate)
+ self.aggregate_mock.remove_host.assert_called_once_with(
+ self.fake_ag, parsed_args.host)
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(self.data, data)
+
+
+class TestAggregateSet(TestAggregate):
+
+ def setUp(self):
+ super(TestAggregateSet, self).setUp()
+
+ self.aggregate_mock.get.return_value = self.fake_ag
+ self.cmd = aggregate.SetAggregate(self.app, None)
+
+ def test_aggregate_set_no_option(self):
+ arglist = [
+ 'ag1',
+ ]
+ verifylist = [
+ ('aggregate', 'ag1'),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ result = self.cmd.take_action(parsed_args)
+
+ self.aggregate_mock.get.assert_called_once_with(parsed_args.aggregate)
+ self.assertNotCalled(self.aggregate_mock.update)
+ self.assertNotCalled(self.aggregate_mock.set_metadata)
+ self.assertIsNone(result)
+
+ def test_aggregate_set_with_name(self):
+ arglist = [
+ '--name', 'new_name',
+ 'ag1',
+ ]
+ verifylist = [
+ ('name', 'new_name'),
+ ('aggregate', 'ag1'),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ result = self.cmd.take_action(parsed_args)
+
+ self.aggregate_mock.get.assert_called_once_with(parsed_args.aggregate)
+ self.aggregate_mock.update.assert_called_once_with(
+ self.fake_ag, {'name': parsed_args.name})
+ self.assertNotCalled(self.aggregate_mock.set_metadata)
+ self.assertIsNone(result)
+
+ def test_aggregate_set_with_zone(self):
+ arglist = [
+ '--zone', 'new_zone',
+ 'ag1',
+ ]
+ verifylist = [
+ ('zone', 'new_zone'),
+ ('aggregate', 'ag1'),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ result = self.cmd.take_action(parsed_args)
+
+ self.aggregate_mock.get.assert_called_once_with(parsed_args.aggregate)
+ self.aggregate_mock.update.assert_called_once_with(
+ self.fake_ag, {'availability_zone': parsed_args.zone})
+ self.assertNotCalled(self.aggregate_mock.set_metadata)
+ self.assertIsNone(result)
+
+ def test_aggregate_set_with_property(self):
+ arglist = [
+ '--property', 'key1=value1',
+ '--property', 'key2=value2',
+ 'ag1',
+ ]
+ verifylist = [
+ ('property', {'key1': 'value1', 'key2': 'value2'}),
+ ('aggregate', 'ag1'),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ result = self.cmd.take_action(parsed_args)
+
+ self.aggregate_mock.get.assert_called_once_with(parsed_args.aggregate)
+ self.assertNotCalled(self.aggregate_mock.update)
+ self.aggregate_mock.set_metadata.assert_called_once_with(
+ self.fake_ag, parsed_args.property)
+ self.assertIsNone(result)
+
+
+class TestAggregateShow(TestAggregate):
+
+ columns = (
+ 'availability_zone',
+ 'hosts',
+ 'id',
+ 'name',
+ 'properties',
+ )
+
+ data = (
+ TestAggregate.fake_ag.availability_zone,
+ TestAggregate.fake_ag.hosts,
+ TestAggregate.fake_ag.id,
+ TestAggregate.fake_ag.name,
+ '',
+ )
+
+ def setUp(self):
+ super(TestAggregateShow, self).setUp()
+
+ self.aggregate_mock.get.return_value = self.fake_ag
+ self.cmd = aggregate.ShowAggregate(self.app, None)
+
+ def test_aggregate_show(self):
+ arglist = [
+ 'ag1',
+ ]
+ verifylist = [
+ ('aggregate', 'ag1'),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ columns, data = self.cmd.take_action(parsed_args)
+ self.aggregate_mock.get.assert_called_once_with(parsed_args.aggregate)
+
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(self.data, data)
+
+
+class TestAggregateUnset(TestAggregate):
+
+ def setUp(self):
+ super(TestAggregateUnset, self).setUp()
+
+ self.aggregate_mock.get.return_value = self.fake_ag
+ self.cmd = aggregate.UnsetAggregate(self.app, None)
+
+ def test_aggregate_unset(self):
+ arglist = [
+ '--property', 'unset_key',
+ 'ag1',
+ ]
+ verifylist = [
+ ('property', ['unset_key']),
+ ('aggregate', 'ag1'),
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ result = self.cmd.take_action(parsed_args)
+ self.aggregate_mock.set_metadata.assert_called_once_with(
+ self.fake_ag, {'unset_key': None})
+ self.assertIsNone(result)
+
+ def test_aggregate_unset_no_property(self):
+ arglist = [
+ 'ag1',
+ ]
+ verifylist = None
+ self.assertRaises(tests_utils.ParserException,
+ self.check_parser,
+ self.cmd,
+ arglist,
+ verifylist)
diff --git a/openstackclient/tests/unit/compute/v2/test_console.py b/openstackclient/tests/unit/compute/v2/test_console.py
new file mode 100644
index 00000000..d53d241e
--- /dev/null
+++ b/openstackclient/tests/unit/compute/v2/test_console.py
@@ -0,0 +1,196 @@
+# Copyright 2016 Huawei, Inc. 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 mock
+
+from openstackclient.compute.v2 import console
+from openstackclient.tests.unit.compute.v2 import fakes as compute_fakes
+
+
+class TestConsole(compute_fakes.TestComputev2):
+
+ def setUp(self):
+ super(TestConsole, self).setUp()
+ self.servers_mock = self.app.client_manager.compute.servers
+ self.servers_mock.reset_mock()
+
+
+class TestConsoleUrlShow(TestConsole):
+
+ def setUp(self):
+ super(TestConsoleUrlShow, self).setUp()
+ fake_console_data = {'remote_console': {'url': 'http://localhost',
+ 'protocol': 'fake_protocol',
+ 'type': 'fake_type'}}
+ methods = {
+ 'get_vnc_console': fake_console_data,
+ 'get_spice_console': fake_console_data,
+ 'get_serial_console': fake_console_data,
+ 'get_rdp_console': fake_console_data,
+ 'get_mks_console': fake_console_data,
+ }
+ self.fake_server = compute_fakes.FakeServer.create_one_server(
+ methods=methods)
+ self.servers_mock.get.return_value = self.fake_server
+
+ self.columns = (
+ 'protocol',
+ 'type',
+ 'url',
+ )
+ self.data = (
+ fake_console_data['remote_console']['protocol'],
+ fake_console_data['remote_console']['type'],
+ fake_console_data['remote_console']['url']
+ )
+
+ self.cmd = console.ShowConsoleURL(self.app, None)
+
+ def test_console_url_show_by_default(self):
+ arglist = [
+ 'foo_vm',
+ ]
+ verifylist = [
+ ('url_type', 'novnc'),
+ ('server', 'foo_vm'),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ columns, data = self.cmd.take_action(parsed_args)
+ self.fake_server.get_vnc_console.assert_called_once_with('novnc')
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(self.data, data)
+
+ def test_console_url_show_with_novnc(self):
+ arglist = [
+ '--novnc',
+ 'foo_vm',
+ ]
+ verifylist = [
+ ('url_type', 'novnc'),
+ ('server', 'foo_vm'),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ columns, data = self.cmd.take_action(parsed_args)
+ self.fake_server.get_vnc_console.assert_called_once_with('novnc')
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(self.data, data)
+
+ def test_console_url_show_with_xvpvnc(self):
+ arglist = [
+ '--xvpvnc',
+ 'foo_vm',
+ ]
+ verifylist = [
+ ('url_type', 'xvpvnc'),
+ ('server', 'foo_vm'),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ columns, data = self.cmd.take_action(parsed_args)
+ self.fake_server.get_vnc_console.assert_called_once_with('xvpvnc')
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(self.data, data)
+
+ def test_console_url_show_with_spice(self):
+ arglist = [
+ '--spice',
+ 'foo_vm',
+ ]
+ verifylist = [
+ ('url_type', 'spice-html5'),
+ ('server', 'foo_vm'),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ columns, data = self.cmd.take_action(parsed_args)
+ self.fake_server.get_spice_console.assert_called_once_with(
+ 'spice-html5')
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(self.data, data)
+
+ def test_console_url_show_compatible(self):
+ methods = {
+ 'get_vnc_console': {'console': {'url': 'http://localhost',
+ 'type': 'fake_type'}},
+ }
+ old_fake_server = compute_fakes.FakeServer.create_one_server(
+ methods=methods)
+ old_columns = (
+ 'type',
+ 'url',
+ )
+ old_data = (
+ methods['get_vnc_console']['console']['type'],
+ methods['get_vnc_console']['console']['url']
+ )
+ arglist = [
+ 'foo_vm',
+ ]
+ verifylist = [
+ ('url_type', 'novnc'),
+ ('server', 'foo_vm'),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ with mock.patch.object(self.servers_mock, 'get',
+ return_value=old_fake_server):
+ columns, data = self.cmd.take_action(parsed_args)
+ old_fake_server.get_vnc_console.assert_called_once_with('novnc')
+ self.assertEqual(old_columns, columns)
+ self.assertEqual(old_data, data)
+
+ def test_console_url_show_with_rdp(self):
+ arglist = [
+ '--rdp',
+ 'foo_vm',
+ ]
+ verifylist = [
+ ('url_type', 'rdp-html5'),
+ ('server', 'foo_vm'),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ columns, data = self.cmd.take_action(parsed_args)
+ self.fake_server.get_rdp_console.assert_called_once_with(
+ 'rdp-html5')
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(self.data, data)
+
+ def test_console_url_show_with_serial(self):
+ arglist = [
+ '--serial',
+ 'foo_vm',
+ ]
+ verifylist = [
+ ('url_type', 'serial'),
+ ('server', 'foo_vm'),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ columns, data = self.cmd.take_action(parsed_args)
+ self.fake_server.get_serial_console.assert_called_once_with(
+ 'serial')
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(self.data, data)
+
+ def test_console_url_show_with_mks(self):
+ arglist = [
+ '--mks',
+ 'foo_vm',
+ ]
+ verifylist = [
+ ('url_type', 'webmks'),
+ ('server', 'foo_vm'),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ columns, data = self.cmd.take_action(parsed_args)
+ self.fake_server.get_mks_console.assert_called_once_with()
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(self.data, data)
diff --git a/openstackclient/tests/unit/compute/v2/test_flavor.py b/openstackclient/tests/unit/compute/v2/test_flavor.py
new file mode 100644
index 00000000..ace650eb
--- /dev/null
+++ b/openstackclient/tests/unit/compute/v2/test_flavor.py
@@ -0,0 +1,825 @@
+# Copyright 2015 Symantec Corporation
+#
+# 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 mock import call
+
+from osc_lib import exceptions
+from osc_lib import utils
+
+from openstackclient.compute.v2 import flavor
+from openstackclient.tests.unit.compute.v2 import fakes as compute_fakes
+from openstackclient.tests.unit.identity.v3 import fakes as identity_fakes
+from openstackclient.tests.unit import utils as tests_utils
+
+
+class TestFlavor(compute_fakes.TestComputev2):
+
+ def setUp(self):
+ super(TestFlavor, self).setUp()
+
+ # Get a shortcut to the FlavorManager Mock
+ self.flavors_mock = self.app.client_manager.compute.flavors
+ self.flavors_mock.reset_mock()
+
+ # Get a shortcut to the FlavorAccessManager Mock
+ self.flavor_access_mock = self.app.client_manager.compute.flavor_access
+ self.flavor_access_mock.reset_mock()
+
+ self.projects_mock = self.app.client_manager.identity.projects
+ self.projects_mock.reset_mock()
+
+
+class TestFlavorCreate(TestFlavor):
+
+ flavor = compute_fakes.FakeFlavor.create_one_flavor(
+ attrs={'links': 'flavor-links'})
+ project = identity_fakes.FakeProject.create_one_project()
+ columns = (
+ 'OS-FLV-DISABLED:disabled',
+ 'OS-FLV-EXT-DATA:ephemeral',
+ 'disk',
+ 'id',
+ 'name',
+ 'os-flavor-access:is_public',
+ 'properties',
+ 'ram',
+ 'rxtx_factor',
+ 'swap',
+ 'vcpus',
+ )
+ data = (
+ flavor.disabled,
+ flavor.ephemeral,
+ flavor.disk,
+ flavor.id,
+ flavor.name,
+ flavor.is_public,
+ utils.format_dict(flavor.properties),
+ flavor.ram,
+ flavor.rxtx_factor,
+ flavor.swap,
+ flavor.vcpus,
+ )
+
+ def setUp(self):
+ super(TestFlavorCreate, self).setUp()
+
+ # Return a project
+ self.projects_mock.get.return_value = self.project
+ self.flavors_mock.create.return_value = self.flavor
+ self.cmd = flavor.CreateFlavor(self.app, None)
+
+ def test_flavor_create_default_options(self):
+
+ arglist = [
+ self.flavor.name
+ ]
+ verifylist = [
+ ('name', self.flavor.name),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ default_args = (
+ self.flavor.name,
+ 256,
+ 1,
+ 0,
+ 'auto',
+ 0,
+ 0,
+ 1.0,
+ True
+ )
+ columns, data = self.cmd.take_action(parsed_args)
+ self.flavors_mock.create.assert_called_once_with(*default_args)
+
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(self.data, data)
+
+ def test_flavor_create_all_options(self):
+
+ arglist = [
+ '--id', self.flavor.id,
+ '--ram', str(self.flavor.ram),
+ '--disk', str(self.flavor.disk),
+ '--ephemeral', str(self.flavor.ephemeral),
+ '--swap', str(self.flavor.swap),
+ '--vcpus', str(self.flavor.vcpus),
+ '--rxtx-factor', str(self.flavor.rxtx_factor),
+ '--public',
+ '--property', 'property=value',
+ self.flavor.name,
+ ]
+ verifylist = [
+ ('id', self.flavor.id),
+ ('ram', self.flavor.ram),
+ ('disk', self.flavor.disk),
+ ('ephemeral', self.flavor.ephemeral),
+ ('swap', self.flavor.swap),
+ ('vcpus', self.flavor.vcpus),
+ ('rxtx_factor', self.flavor.rxtx_factor),
+ ('public', True),
+ ('property', {'property': 'value'}),
+ ('name', self.flavor.name),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ args = (
+ self.flavor.name,
+ self.flavor.ram,
+ self.flavor.vcpus,
+ self.flavor.disk,
+ self.flavor.id,
+ self.flavor.ephemeral,
+ self.flavor.swap,
+ self.flavor.rxtx_factor,
+ self.flavor.is_public,
+ )
+ columns, data = self.cmd.take_action(parsed_args)
+ self.flavors_mock.create.assert_called_once_with(*args)
+ self.flavor.set_keys.assert_called_once_with({'property': 'value'})
+ self.flavor.get_keys.assert_called_once_with()
+
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(self.data, data)
+
+ def test_flavor_create_other_options(self):
+
+ self.flavor.is_public = False
+ arglist = [
+ '--id', self.flavor.id,
+ '--ram', str(self.flavor.ram),
+ '--disk', str(self.flavor.disk),
+ '--ephemeral', str(self.flavor.ephemeral),
+ '--swap', str(self.flavor.swap),
+ '--vcpus', str(self.flavor.vcpus),
+ '--rxtx-factor', str(self.flavor.rxtx_factor),
+ '--private',
+ '--project', self.project.id,
+ '--property', 'key1=value1',
+ '--property', 'key2=value2',
+ self.flavor.name,
+ ]
+ verifylist = [
+ ('id', self.flavor.id),
+ ('ram', self.flavor.ram),
+ ('disk', self.flavor.disk),
+ ('ephemeral', self.flavor.ephemeral),
+ ('swap', self.flavor.swap),
+ ('vcpus', self.flavor.vcpus),
+ ('rxtx_factor', self.flavor.rxtx_factor),
+ ('public', False),
+ ('project', self.project.id),
+ ('property', {'key1': 'value1', 'key2': 'value2'}),
+ ('name', self.flavor.name),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ args = (
+ self.flavor.name,
+ self.flavor.ram,
+ self.flavor.vcpus,
+ self.flavor.disk,
+ self.flavor.id,
+ self.flavor.ephemeral,
+ self.flavor.swap,
+ self.flavor.rxtx_factor,
+ self.flavor.is_public,
+ )
+ columns, data = self.cmd.take_action(parsed_args)
+ self.flavors_mock.create.assert_called_once_with(*args)
+ self.flavor_access_mock.add_tenant_access.assert_called_with(
+ self.flavor.id,
+ self.project.id,
+ )
+ self.flavor.set_keys.assert_called_with(
+ {'key1': 'value1', 'key2': 'value2'})
+ self.flavor.get_keys.assert_called_with()
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(self.data, data)
+
+ def test_public_flavor_create_with_project(self):
+ arglist = [
+ '--project', self.project.id,
+ self.flavor.name,
+ ]
+ verifylist = [
+ ('project', self.project.id),
+ ('name', self.flavor.name),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ self.assertRaises(exceptions.CommandError,
+ self.cmd.take_action,
+ parsed_args)
+
+ def test_flavor_create_no_options(self):
+ arglist = []
+ verifylist = None
+ self.assertRaises(tests_utils.ParserException,
+ self.check_parser,
+ self.cmd,
+ arglist,
+ verifylist)
+
+
+class TestFlavorDelete(TestFlavor):
+
+ flavors = compute_fakes.FakeFlavor.create_flavors(count=2)
+
+ def setUp(self):
+ super(TestFlavorDelete, self).setUp()
+
+ self.flavors_mock.get = (
+ compute_fakes.FakeFlavor.get_flavors(self.flavors))
+ self.flavors_mock.delete.return_value = None
+
+ self.cmd = flavor.DeleteFlavor(self.app, None)
+
+ def test_flavor_delete(self):
+ arglist = [
+ self.flavors[0].id
+ ]
+ verifylist = [
+ ('flavor', [self.flavors[0].id]),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ result = self.cmd.take_action(parsed_args)
+
+ self.flavors_mock.delete.assert_called_with(self.flavors[0].id)
+ self.assertIsNone(result)
+
+ def test_delete_multiple_flavors(self):
+ arglist = []
+ for f in self.flavors:
+ arglist.append(f.id)
+ verifylist = [
+ ('flavor', arglist),
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ result = self.cmd.take_action(parsed_args)
+
+ calls = []
+ for f in self.flavors:
+ calls.append(call(f.id))
+ self.flavors_mock.delete.assert_has_calls(calls)
+ self.assertIsNone(result)
+
+ def test_multi_flavors_delete_with_exception(self):
+ arglist = [
+ self.flavors[0].id,
+ 'unexist_flavor',
+ ]
+ verifylist = [
+ ('flavor', [self.flavors[0].id, 'unexist_flavor'])
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ find_mock_result = [self.flavors[0], exceptions.CommandError]
+ self.flavors_mock.get = (
+ mock.MagicMock(side_effect=find_mock_result)
+ )
+ self.flavors_mock.find.side_effect = exceptions.NotFound(None)
+
+ try:
+ self.cmd.take_action(parsed_args)
+ self.fail('CommandError should be raised.')
+ except exceptions.CommandError as e:
+ self.assertEqual('1 of 2 flavors failed to delete.', str(e))
+
+ self.flavors_mock.get.assert_any_call(self.flavors[0].id)
+ self.flavors_mock.get.assert_any_call('unexist_flavor')
+ self.flavors_mock.delete.assert_called_once_with(self.flavors[0].id)
+
+
+class TestFlavorList(TestFlavor):
+
+ # Return value of self.flavors_mock.list().
+ flavors = compute_fakes.FakeFlavor.create_flavors(count=1)
+
+ columns = (
+ 'ID',
+ 'Name',
+ 'RAM',
+ 'Disk',
+ 'Ephemeral',
+ 'VCPUs',
+ 'Is Public',
+ )
+ columns_long = columns + (
+ 'Swap',
+ 'RXTX Factor',
+ 'Properties'
+ )
+
+ data = ((
+ flavors[0].id,
+ flavors[0].name,
+ flavors[0].ram,
+ flavors[0].disk,
+ flavors[0].ephemeral,
+ flavors[0].vcpus,
+ flavors[0].is_public,
+ ), )
+ data_long = (data[0] + (
+ flavors[0].swap,
+ flavors[0].rxtx_factor,
+ u'property=\'value\''
+ ), )
+
+ def setUp(self):
+ super(TestFlavorList, self).setUp()
+
+ self.flavors_mock.list.return_value = self.flavors
+
+ # Get the command object to test
+ self.cmd = flavor.ListFlavor(self.app, None)
+
+ def test_flavor_list_no_options(self):
+ arglist = []
+ verifylist = [
+ ('public', True),
+ ('all', False),
+ ('long', False),
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ # In base command class Lister in cliff, abstract method take_action()
+ # returns a tuple containing the column names and an iterable
+ # containing the data to be listed.
+ columns, data = self.cmd.take_action(parsed_args)
+
+ # Set expected values
+ kwargs = {
+ 'is_public': True,
+ 'limit': None,
+ 'marker': None
+ }
+
+ self.flavors_mock.list.assert_called_with(
+ **kwargs
+ )
+
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(tuple(self.data), tuple(data))
+
+ def test_flavor_list_all_flavors(self):
+ arglist = [
+ '--all',
+ ]
+ verifylist = [
+ ('all', True),
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ # In base command class Lister in cliff, abstract method take_action()
+ # returns a tuple containing the column names and an iterable
+ # containing the data to be listed.
+ columns, data = self.cmd.take_action(parsed_args)
+
+ # Set expected values
+ kwargs = {
+ 'is_public': None,
+ 'limit': None,
+ 'marker': None
+ }
+
+ self.flavors_mock.list.assert_called_with(
+ **kwargs
+ )
+
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(tuple(self.data), tuple(data))
+
+ def test_flavor_list_private_flavors(self):
+ arglist = [
+ '--private',
+ ]
+ verifylist = [
+ ('public', False),
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ # In base command class Lister in cliff, abstract method take_action()
+ # returns a tuple containing the column names and an iterable
+ # containing the data to be listed.
+ columns, data = self.cmd.take_action(parsed_args)
+
+ # Set expected values
+ kwargs = {
+ 'is_public': False,
+ 'limit': None,
+ 'marker': None
+ }
+
+ self.flavors_mock.list.assert_called_with(
+ **kwargs
+ )
+
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(tuple(self.data), tuple(data))
+
+ def test_flavor_list_public_flavors(self):
+ arglist = [
+ '--public',
+ ]
+ verifylist = [
+ ('public', True),
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ # In base command class Lister in cliff, abstract method take_action()
+ # returns a tuple containing the column names and an iterable
+ # containing the data to be listed.
+ columns, data = self.cmd.take_action(parsed_args)
+
+ # Set expected values
+ kwargs = {
+ 'is_public': True,
+ 'limit': None,
+ 'marker': None
+ }
+
+ self.flavors_mock.list.assert_called_with(
+ **kwargs
+ )
+
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(tuple(self.data), tuple(data))
+
+ def test_flavor_list_long(self):
+ arglist = [
+ '--long',
+ ]
+ verifylist = [
+ ('long', True),
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ # In base command class Lister in cliff, abstract method take_action()
+ # returns a tuple containing the column names and an iterable
+ # containing the data to be listed.
+ columns, data = self.cmd.take_action(parsed_args)
+
+ # Set expected values
+ kwargs = {
+ 'is_public': True,
+ 'limit': None,
+ 'marker': None
+ }
+
+ self.flavors_mock.list.assert_called_with(
+ **kwargs
+ )
+
+ self.assertEqual(self.columns_long, columns)
+ self.assertEqual(tuple(self.data_long), tuple(data))
+
+
+class TestFlavorSet(TestFlavor):
+
+ # Return value of self.flavors_mock.find().
+ flavor = compute_fakes.FakeFlavor.create_one_flavor(
+ attrs={'os-flavor-access:is_public': False})
+ project = identity_fakes.FakeProject.create_one_project()
+
+ def setUp(self):
+ super(TestFlavorSet, self).setUp()
+
+ self.flavors_mock.find.return_value = self.flavor
+ self.flavors_mock.get.side_effect = exceptions.NotFound(None)
+ # Return a project
+ self.projects_mock.get.return_value = self.project
+ self.cmd = flavor.SetFlavor(self.app, None)
+
+ def test_flavor_set_property(self):
+ arglist = [
+ '--property', 'FOO="B A R"',
+ 'baremetal'
+ ]
+ verifylist = [
+ ('property', {'FOO': '"B A R"'}),
+ ('flavor', 'baremetal')
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ result = self.cmd.take_action(parsed_args)
+ self.flavors_mock.find.assert_called_with(name=parsed_args.flavor,
+ is_public=None)
+ self.flavor.set_keys.assert_called_with({'FOO': '"B A R"'})
+ self.assertIsNone(result)
+
+ def test_flavor_set_project(self):
+ arglist = [
+ '--project', self.project.id,
+ self.flavor.id,
+ ]
+ verifylist = [
+ ('project', self.project.id),
+ ('flavor', self.flavor.id),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ result = self.cmd.take_action(parsed_args)
+
+ self.flavors_mock.find.assert_called_with(name=parsed_args.flavor,
+ is_public=None)
+ self.flavor_access_mock.add_tenant_access.assert_called_with(
+ self.flavor.id,
+ self.project.id,
+ )
+ self.flavor.set_keys.assert_not_called()
+ self.assertIsNone(result)
+
+ def test_flavor_set_no_project(self):
+ arglist = [
+ '--project',
+ self.flavor.id,
+ ]
+ verifylist = [
+ ('project', None),
+ ('flavor', self.flavor.id),
+ ]
+ self.assertRaises(tests_utils.ParserException, self.check_parser,
+ self.cmd, arglist, verifylist)
+
+ def test_flavor_set_no_flavor(self):
+ arglist = [
+ '--project', self.project.id,
+ ]
+ verifylist = [
+ ('project', self.project.id),
+ ]
+ self.assertRaises(tests_utils.ParserException, self.check_parser,
+ self.cmd, arglist, verifylist)
+
+ def test_flavor_set_with_unexist_flavor(self):
+ self.flavors_mock.get.side_effect = exceptions.NotFound(None)
+ self.flavors_mock.find.side_effect = exceptions.NotFound(None)
+
+ arglist = [
+ '--project', self.project.id,
+ 'unexist_flavor',
+ ]
+ verifylist = [
+ ('project', self.project.id),
+ ('flavor', 'unexist_flavor'),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ self.assertRaises(exceptions.CommandError,
+ self.cmd.take_action,
+ parsed_args)
+
+ def test_flavor_set_nothing(self):
+ arglist = [
+ self.flavor.id,
+ ]
+ verifylist = [
+ ('flavor', self.flavor.id),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ result = self.cmd.take_action(parsed_args)
+
+ self.flavors_mock.find.assert_called_with(name=parsed_args.flavor,
+ is_public=None)
+ self.flavor_access_mock.add_tenant_access.assert_not_called()
+ self.assertIsNone(result)
+
+
+class TestFlavorShow(TestFlavor):
+
+ # Return value of self.flavors_mock.find().
+ flavor_access = compute_fakes.FakeFlavorAccess.create_one_flavor_access()
+ flavor = compute_fakes.FakeFlavor.create_one_flavor()
+
+ columns = (
+ 'OS-FLV-DISABLED:disabled',
+ 'OS-FLV-EXT-DATA:ephemeral',
+ 'access_project_ids',
+ 'disk',
+ 'id',
+ 'name',
+ 'os-flavor-access:is_public',
+ 'properties',
+ 'ram',
+ 'rxtx_factor',
+ 'swap',
+ 'vcpus',
+ )
+
+ data = (
+ flavor.disabled,
+ flavor.ephemeral,
+ None,
+ flavor.disk,
+ flavor.id,
+ flavor.name,
+ flavor.is_public,
+ utils.format_dict(flavor.get_keys()),
+ flavor.ram,
+ flavor.rxtx_factor,
+ flavor.swap,
+ flavor.vcpus,
+ )
+
+ def setUp(self):
+ super(TestFlavorShow, self).setUp()
+
+ # Return value of _find_resource()
+ self.flavors_mock.find.return_value = self.flavor
+ self.flavors_mock.get.side_effect = exceptions.NotFound(None)
+ self.flavor_access_mock.list.return_value = [self.flavor_access]
+ self.cmd = flavor.ShowFlavor(self.app, None)
+
+ def test_show_no_options(self):
+ arglist = []
+ verifylist = []
+
+ # Missing required args should boil here
+ self.assertRaises(tests_utils.ParserException, self.check_parser,
+ self.cmd, arglist, verifylist)
+
+ def test_public_flavor_show(self):
+ arglist = [
+ self.flavor.name,
+ ]
+ verifylist = [
+ ('flavor', self.flavor.name),
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ columns, data = self.cmd.take_action(parsed_args)
+
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(self.data, data)
+
+ def test_private_flavor_show(self):
+ private_flavor = compute_fakes.FakeFlavor.create_one_flavor(
+ attrs={
+ 'os-flavor-access:is_public': False,
+ }
+ )
+ self.flavors_mock.find.return_value = private_flavor
+
+ arglist = [
+ private_flavor.name,
+ ]
+ verifylist = [
+ ('flavor', private_flavor.name),
+ ]
+
+ data_with_project = (
+ private_flavor.disabled,
+ private_flavor.ephemeral,
+ self.flavor_access.tenant_id,
+ private_flavor.disk,
+ private_flavor.id,
+ private_flavor.name,
+ private_flavor.is_public,
+ utils.format_dict(private_flavor.get_keys()),
+ private_flavor.ram,
+ private_flavor.rxtx_factor,
+ private_flavor.swap,
+ private_flavor.vcpus,
+ )
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ columns, data = self.cmd.take_action(parsed_args)
+
+ self.flavor_access_mock.list.assert_called_with(
+ flavor=private_flavor.id)
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(data_with_project, data)
+
+
+class TestFlavorUnset(TestFlavor):
+
+ # Return value of self.flavors_mock.find().
+ flavor = compute_fakes.FakeFlavor.create_one_flavor(
+ attrs={'os-flavor-access:is_public': False})
+ project = identity_fakes.FakeProject.create_one_project()
+
+ def setUp(self):
+ super(TestFlavorUnset, self).setUp()
+
+ self.flavors_mock.find.return_value = self.flavor
+ self.flavors_mock.get.side_effect = exceptions.NotFound(None)
+ # Return a project
+ self.projects_mock.get.return_value = self.project
+ self.cmd = flavor.UnsetFlavor(self.app, None)
+
+ def test_flavor_unset_property(self):
+ arglist = [
+ '--property', 'property',
+ 'baremetal'
+ ]
+ verifylist = [
+ ('property', ['property']),
+ ('flavor', 'baremetal'),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ result = self.cmd.take_action(parsed_args)
+ self.flavors_mock.find.assert_called_with(name=parsed_args.flavor,
+ is_public=None)
+ self.flavor.unset_keys.assert_called_with(['property'])
+ self.flavor_access_mock.remove_tenant_access.assert_not_called()
+ self.assertIsNone(result)
+
+ def test_flavor_unset_project(self):
+ arglist = [
+ '--project', self.project.id,
+ self.flavor.id,
+ ]
+ verifylist = [
+ ('project', self.project.id),
+ ('flavor', self.flavor.id),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ result = self.cmd.take_action(parsed_args)
+ self.assertIsNone(result)
+
+ self.flavors_mock.find.assert_called_with(name=parsed_args.flavor,
+ is_public=None)
+ self.flavor_access_mock.remove_tenant_access.assert_called_with(
+ self.flavor.id,
+ self.project.id,
+ )
+ self.flavor.unset_keys.assert_not_called()
+ self.assertIsNone(result)
+
+ def test_flavor_unset_no_project(self):
+ arglist = [
+ '--project',
+ self.flavor.id,
+ ]
+ verifylist = [
+ ('project', None),
+ ('flavor', self.flavor.id),
+ ]
+ self.assertRaises(tests_utils.ParserException, self.check_parser,
+ self.cmd, arglist, verifylist)
+
+ def test_flavor_unset_no_flavor(self):
+ arglist = [
+ '--project', self.project.id,
+ ]
+ verifylist = [
+ ('project', self.project.id),
+ ]
+ self.assertRaises(tests_utils.ParserException, self.check_parser,
+ self.cmd, arglist, verifylist)
+
+ def test_flavor_unset_with_unexist_flavor(self):
+ self.flavors_mock.get.side_effect = exceptions.NotFound(None)
+ self.flavors_mock.find.side_effect = exceptions.NotFound(None)
+
+ arglist = [
+ '--project', self.project.id,
+ 'unexist_flavor',
+ ]
+ verifylist = [
+ ('project', self.project.id),
+ ('flavor', 'unexist_flavor'),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ self.assertRaises(exceptions.CommandError, self.cmd.take_action,
+ parsed_args)
+
+ def test_flavor_unset_nothing(self):
+ arglist = [
+ self.flavor.id,
+ ]
+ verifylist = [
+ ('flavor', self.flavor.id),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ result = self.cmd.take_action(parsed_args)
+ self.assertIsNone(result)
+
+ self.flavor_access_mock.remove_tenant_access.assert_not_called()
diff --git a/openstackclient/tests/unit/compute/v2/test_host.py b/openstackclient/tests/unit/compute/v2/test_host.py
new file mode 100644
index 00000000..a388172f
--- /dev/null
+++ b/openstackclient/tests/unit/compute/v2/test_host.py
@@ -0,0 +1,179 @@
+# Copyright 2016 IBM Corporation
+#
+# 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 openstackclient.compute.v2 import host
+from openstackclient.tests.unit.compute.v2 import fakes as compute_fakes
+from openstackclient.tests.unit import utils as tests_utils
+
+
+class TestHost(compute_fakes.TestComputev2):
+
+ def setUp(self):
+ super(TestHost, self).setUp()
+
+ # Get a shortcut to the FlavorManager Mock
+ self.host_mock = self.app.client_manager.compute.hosts
+ self.host_mock.reset_mock()
+
+
+class TestHostList(TestHost):
+
+ host = compute_fakes.FakeHost.create_one_host()
+
+ columns = (
+ 'Host Name',
+ 'Service',
+ 'Zone',
+ )
+
+ data = [(
+ host.host_name,
+ host.service,
+ host.zone,
+ )]
+
+ def setUp(self):
+ super(TestHostList, self).setUp()
+
+ self.host_mock.list_all.return_value = [self.host]
+
+ self.cmd = host.ListHost(self.app, None)
+
+ def test_host_list_no_option(self):
+ arglist = []
+ verifylist = []
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ columns, data = self.cmd.take_action(parsed_args)
+
+ self.host_mock.list_all.assert_called_with(None)
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(self.data, list(data))
+
+ def test_host_list_with_option(self):
+ arglist = [
+ '--zone', self.host.zone,
+ ]
+ verifylist = [
+ ('zone', self.host.zone),
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ columns, data = self.cmd.take_action(parsed_args)
+
+ self.host_mock.list_all.assert_called_with(self.host.zone)
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(self.data, list(data))
+
+
+class TestHostSet(TestHost):
+
+ def setUp(self):
+ super(TestHostSet, self).setUp()
+
+ self.host = compute_fakes.FakeHost.create_one_host()
+ self.host_mock.get.return_value = self.host
+ self.host_mock.update.return_value = None
+
+ self.cmd = host.SetHost(self.app, None)
+
+ def test_host_set_no_option(self):
+ arglist = [
+ self.host.host
+ ]
+ verifylist = [
+ ('host', self.host.host)
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ result = self.cmd.take_action(parsed_args)
+ self.assertIsNone(result)
+
+ body = {}
+ self.host_mock.update.assert_called_with(self.host.host, body)
+
+ def test_host_set(self):
+ arglist = [
+ '--enable',
+ '--disable-maintenance',
+ self.host.host
+ ]
+ verifylist = [
+ ('enable', True),
+ ('enable_maintenance', False),
+ ('host', self.host.host)
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ result = self.cmd.take_action(parsed_args)
+ self.assertIsNone(result)
+
+ body = {'status': 'enable', 'maintenance_mode': 'disable'}
+ self.host_mock.update.assert_called_with(self.host.host, body)
+
+
+class TestHostShow(TestHost):
+
+ host = compute_fakes.FakeHost.create_one_host()
+
+ columns = (
+ 'Host',
+ 'Project',
+ 'CPU',
+ 'Memory MB',
+ 'Disk GB',
+ )
+ data = [(
+ host.host,
+ host.project,
+ host.cpu,
+ host.memory_mb,
+ host.disk_gb,
+ )]
+
+ def setUp(self):
+ super(TestHostShow, self).setUp()
+
+ self.host_mock.get.return_value = [self.host]
+
+ self.cmd = host.ShowHost(self.app, None)
+
+ def test_host_show_no_option(self):
+ arglist = []
+ verifylist = []
+
+ # Missing required args should bail here
+ self.assertRaises(tests_utils.ParserException, self.check_parser,
+ self.cmd, arglist, verifylist)
+
+ def test_host_show_with_option(self):
+ arglist = [
+ self.host.host_name,
+ ]
+ verifylist = [
+ ('host', self.host.host_name),
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ columns, data = self.cmd.take_action(parsed_args)
+
+ self.host_mock.get.assert_called_with(self.host.host_name)
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(self.data, list(data))
diff --git a/openstackclient/tests/unit/compute/v2/test_hypervisor.py b/openstackclient/tests/unit/compute/v2/test_hypervisor.py
new file mode 100644
index 00000000..d94a107c
--- /dev/null
+++ b/openstackclient/tests/unit/compute/v2/test_hypervisor.py
@@ -0,0 +1,221 @@
+# Copyright 2016 EasyStack Corporation
+#
+# 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 copy
+
+from osc_lib import exceptions
+
+from openstackclient.compute.v2 import hypervisor
+from openstackclient.tests.unit.compute.v2 import fakes as compute_fakes
+from openstackclient.tests.unit import fakes
+
+
+class TestHypervisor(compute_fakes.TestComputev2):
+
+ def setUp(self):
+ super(TestHypervisor, self).setUp()
+
+ # Get a shortcut to the compute client hypervisors mock
+ self.hypervisors_mock = self.app.client_manager.compute.hypervisors
+ self.hypervisors_mock.reset_mock()
+
+ # Get a shortcut to the compute client aggregates mock
+ self.aggregates_mock = self.app.client_manager.compute.aggregates
+ self.aggregates_mock.reset_mock()
+
+
+class TestHypervisorList(TestHypervisor):
+
+ def setUp(self):
+ super(TestHypervisorList, self).setUp()
+
+ # Fake hypervisors to be listed up
+ self.hypervisors = compute_fakes.FakeHypervisor.create_hypervisors()
+ self.hypervisors_mock.list.return_value = self.hypervisors
+
+ self.columns = (
+ "ID",
+ "Hypervisor Hostname"
+ )
+ self.data = (
+ (
+ self.hypervisors[0].id,
+ self.hypervisors[0].hypervisor_hostname,
+ ),
+ (
+ self.hypervisors[1].id,
+ self.hypervisors[1].hypervisor_hostname,
+ ),
+ )
+
+ # Get the command object to test
+ self.cmd = hypervisor.ListHypervisor(self.app, None)
+
+ def test_hypervisor_list_no_option(self):
+ arglist = []
+ verifylist = []
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ # In base command class Lister in cliff, abstract method take_action()
+ # returns a tuple containing the column names and an iterable
+ # containing the data to be listed.
+ columns, data = self.cmd.take_action(parsed_args)
+
+ self.hypervisors_mock.list.assert_called_with()
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(self.data, tuple(data))
+
+ def test_hypervisor_list_matching_option_found(self):
+ arglist = [
+ '--matching', self.hypervisors[0].hypervisor_hostname,
+ ]
+ verifylist = [
+ ('matching', self.hypervisors[0].hypervisor_hostname),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ # Fake the return value of search()
+ self.hypervisors_mock.search.return_value = [self.hypervisors[0]]
+ self.data = (
+ (
+ self.hypervisors[0].id,
+ self.hypervisors[0].hypervisor_hostname,
+ ),
+ )
+
+ # In base command class Lister in cliff, abstract method take_action()
+ # returns a tuple containing the column names and an iterable
+ # containing the data to be listed.
+ columns, data = self.cmd.take_action(parsed_args)
+
+ self.hypervisors_mock.search.assert_called_with(
+ self.hypervisors[0].hypervisor_hostname
+ )
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(self.data, tuple(data))
+
+ def test_hypervisor_list_matching_option_not_found(self):
+ arglist = [
+ '--matching', 'xxx',
+ ]
+ verifylist = [
+ ('matching', 'xxx'),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ # Fake exception raised from search()
+ self.hypervisors_mock.search.side_effect = exceptions.NotFound(None)
+
+ self.assertRaises(exceptions.NotFound,
+ self.cmd.take_action,
+ parsed_args)
+
+
+class TestHypervisorShow(TestHypervisor):
+
+ def setUp(self):
+ super(TestHypervisorShow, self).setUp()
+
+ # Fake hypervisors to be listed up
+ self.hypervisor = compute_fakes.FakeHypervisor.create_one_hypervisor()
+
+ # Return value of utils.find_resource()
+ self.hypervisors_mock.get.return_value = self.hypervisor
+
+ # Return value of compute_client.aggregates.list()
+ self.aggregates_mock.list.return_value = []
+
+ # Return value of compute_client.hypervisors.uptime()
+ uptime_info = {
+ 'status': self.hypervisor.status,
+ 'state': self.hypervisor.state,
+ 'id': self.hypervisor.id,
+ 'hypervisor_hostname': self.hypervisor.hypervisor_hostname,
+ 'uptime': ' 01:28:24 up 3 days, 11:15, 1 user, '
+ ' load average: 0.94, 0.62, 0.50\n',
+ }
+ self.hypervisors_mock.uptime.return_value = fakes.FakeResource(
+ info=copy.deepcopy(uptime_info),
+ loaded=True
+ )
+
+ self.columns = (
+ 'aggregates',
+ 'cpu_info',
+ 'current_workload',
+ 'disk_available_least',
+ 'free_disk_gb',
+ 'free_ram_mb',
+ 'host_ip',
+ 'hypervisor_hostname',
+ 'hypervisor_type',
+ 'hypervisor_version',
+ 'id',
+ 'local_gb',
+ 'local_gb_used',
+ 'memory_mb',
+ 'memory_mb_used',
+ 'running_vms',
+ 'service_host',
+ 'service_id',
+ 'state',
+ 'status',
+ 'vcpus',
+ 'vcpus_used',
+ )
+ self.data = (
+ [],
+ {'aaa': 'aaa'},
+ 0,
+ 50,
+ 50,
+ 1024,
+ '192.168.0.10',
+ self.hypervisor.hypervisor_hostname,
+ 'QEMU',
+ 2004001,
+ self.hypervisor.id,
+ 50,
+ 0,
+ 1024,
+ 512,
+ 0,
+ 'aaa',
+ 1,
+ 'up',
+ 'enabled',
+ 4,
+ 0,
+ )
+
+ # Get the command object to test
+ self.cmd = hypervisor.ShowHypervisor(self.app, None)
+
+ def test_hypervisor_show(self):
+ arglist = [
+ self.hypervisor.hypervisor_hostname,
+ ]
+ verifylist = [
+ ('hypervisor', self.hypervisor.hypervisor_hostname),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ # In base command class ShowOne in cliff, abstract method take_action()
+ # returns a two-part tuple with a tuple of column names and a tuple of
+ # data to be shown.
+ columns, data = self.cmd.take_action(parsed_args)
+
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(self.data, data)
diff --git a/openstackclient/tests/unit/compute/v2/test_hypervisor_stats.py b/openstackclient/tests/unit/compute/v2/test_hypervisor_stats.py
new file mode 100644
index 00000000..40086f9b
--- /dev/null
+++ b/openstackclient/tests/unit/compute/v2/test_hypervisor_stats.py
@@ -0,0 +1,79 @@
+# Copyright 2016 EasyStack Corporation
+#
+# 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 openstackclient.compute.v2 import hypervisor_stats
+from openstackclient.tests.unit.compute.v2 import fakes as compute_fakes
+
+
+class TestHypervisorStats(compute_fakes.TestComputev2):
+
+ def setUp(self):
+ super(TestHypervisorStats, self).setUp()
+
+ # Get a shortcut to the compute client hypervisors mock
+ self.hypervisors_mock = self.app.client_manager.compute.hypervisors
+ self.hypervisors_mock.reset_mock()
+
+
+class TestHypervisorStatsShow(TestHypervisorStats):
+
+ def setUp(self):
+ super(TestHypervisorStatsShow, self).setUp()
+
+ self.hypervisor_stats = \
+ compute_fakes.FakeHypervisorStats.create_one_hypervisor_stats()
+
+ self.hypervisors_mock.statistics.return_value =\
+ self.hypervisor_stats
+
+ self.cmd = hypervisor_stats.ShowHypervisorStats(self.app, None)
+
+ self.columns = (
+ 'count',
+ 'current_workload',
+ 'disk_available_least',
+ 'free_disk_gb',
+ 'free_ram_mb',
+ 'local_gb',
+ 'local_gb_used',
+ 'memory_mb',
+ 'memory_mb_used',
+ 'running_vms',
+ 'vcpus',
+ 'vcpus_used',
+ )
+
+ self.data = (
+ 2,
+ 0,
+ 50,
+ 100,
+ 23000,
+ 100,
+ 0,
+ 23800,
+ 1400,
+ 3,
+ 8,
+ 3,
+ )
+
+ def test_hypervisor_show_stats(self):
+ arglist = []
+ verifylist = []
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ columns, data = self.cmd.take_action(parsed_args)
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(self.data, data)
diff --git a/openstackclient/tests/unit/compute/v2/test_keypair.py b/openstackclient/tests/unit/compute/v2/test_keypair.py
new file mode 100644
index 00000000..cb008545
--- /dev/null
+++ b/openstackclient/tests/unit/compute/v2/test_keypair.py
@@ -0,0 +1,312 @@
+# Copyright 2016 IBM
+#
+# 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 mock import call
+
+from osc_lib import exceptions
+from osc_lib import utils
+
+from openstackclient.compute.v2 import keypair
+from openstackclient.tests.unit.compute.v2 import fakes as compute_fakes
+from openstackclient.tests.unit import utils as tests_utils
+
+
+class TestKeypair(compute_fakes.TestComputev2):
+
+ def setUp(self):
+ super(TestKeypair, self).setUp()
+
+ # Get a shortcut to the KeypairManager Mock
+ self.keypairs_mock = self.app.client_manager.compute.keypairs
+ self.keypairs_mock.reset_mock()
+
+
+class TestKeypairCreate(TestKeypair):
+
+ keypair = compute_fakes.FakeKeypair.create_one_keypair()
+
+ def setUp(self):
+ super(TestKeypairCreate, self).setUp()
+
+ self.columns = (
+ 'fingerprint',
+ 'name',
+ 'user_id'
+ )
+ self.data = (
+ self.keypair.fingerprint,
+ self.keypair.name,
+ self.keypair.user_id
+ )
+
+ # Get the command object to test
+ self.cmd = keypair.CreateKeypair(self.app, None)
+
+ self.keypairs_mock.create.return_value = self.keypair
+
+ def test_key_pair_create_no_options(self):
+
+ arglist = [
+ self.keypair.name,
+ ]
+ verifylist = [
+ ('name', self.keypair.name),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ columns, data = self.cmd.take_action(parsed_args)
+
+ self.keypairs_mock.create.assert_called_with(
+ self.keypair.name,
+ public_key=None
+ )
+
+ self.assertEqual({}, columns)
+ self.assertEqual({}, data)
+
+ def test_keypair_create_public_key(self):
+ # overwrite the setup one because we want to omit private_key
+ self.keypair = compute_fakes.FakeKeypair.create_one_keypair(
+ no_pri=True)
+ self.keypairs_mock.create.return_value = self.keypair
+
+ self.data = (
+ self.keypair.fingerprint,
+ self.keypair.name,
+ self.keypair.user_id
+ )
+
+ arglist = [
+ '--public-key', self.keypair.public_key,
+ self.keypair.name,
+ ]
+ verifylist = [
+ ('public_key', self.keypair.public_key),
+ ('name', self.keypair.name)
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ with mock.patch('io.open') as mock_open:
+ mock_open.return_value = mock.MagicMock()
+ m_file = mock_open.return_value.__enter__.return_value
+ m_file.read.return_value = 'dummy'
+
+ columns, data = self.cmd.take_action(parsed_args)
+
+ self.keypairs_mock.create.assert_called_with(
+ self.keypair.name,
+ public_key=self.keypair.public_key
+ )
+
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(self.data, data)
+
+
+class TestKeypairDelete(TestKeypair):
+
+ keypairs = compute_fakes.FakeKeypair.create_keypairs(count=2)
+
+ def setUp(self):
+ super(TestKeypairDelete, self).setUp()
+
+ self.keypairs_mock.get = compute_fakes.FakeKeypair.get_keypairs(
+ self.keypairs)
+ self.keypairs_mock.delete.return_value = None
+
+ self.cmd = keypair.DeleteKeypair(self.app, None)
+
+ def test_keypair_delete(self):
+ arglist = [
+ self.keypairs[0].name
+ ]
+ verifylist = [
+ ('name', [self.keypairs[0].name]),
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ ret = self.cmd.take_action(parsed_args)
+
+ self.assertIsNone(ret)
+ self.keypairs_mock.delete.assert_called_with(self.keypairs[0].name)
+
+ def test_delete_multiple_keypairs(self):
+ arglist = []
+ for k in self.keypairs:
+ arglist.append(k.name)
+ verifylist = [
+ ('name', arglist),
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ result = self.cmd.take_action(parsed_args)
+
+ calls = []
+ for k in self.keypairs:
+ calls.append(call(k.name))
+ self.keypairs_mock.delete.assert_has_calls(calls)
+ self.assertIsNone(result)
+
+ def test_delete_multiple_keypairs_with_exception(self):
+ arglist = [
+ self.keypairs[0].name,
+ 'unexist_keypair',
+ ]
+ verifylist = [
+ ('name', arglist),
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ find_mock_result = [self.keypairs[0], exceptions.CommandError]
+ with mock.patch.object(utils, 'find_resource',
+ side_effect=find_mock_result) as find_mock:
+ try:
+ self.cmd.take_action(parsed_args)
+ self.fail('CommandError should be raised.')
+ except exceptions.CommandError as e:
+ self.assertEqual('1 of 2 public keys failed to delete.',
+ str(e))
+
+ find_mock.assert_any_call(
+ self.keypairs_mock, self.keypairs[0].name)
+ find_mock.assert_any_call(self.keypairs_mock, 'unexist_keypair')
+
+ self.assertEqual(2, find_mock.call_count)
+ self.keypairs_mock.delete.assert_called_once_with(
+ self.keypairs[0].name
+ )
+
+
+class TestKeypairList(TestKeypair):
+
+ # Return value of self.keypairs_mock.list().
+ keypairs = compute_fakes.FakeKeypair.create_keypairs(count=1)
+
+ columns = (
+ "Name",
+ "Fingerprint"
+ )
+
+ data = ((
+ keypairs[0].name,
+ keypairs[0].fingerprint
+ ), )
+
+ def setUp(self):
+ super(TestKeypairList, self).setUp()
+
+ self.keypairs_mock.list.return_value = self.keypairs
+
+ # Get the command object to test
+ self.cmd = keypair.ListKeypair(self.app, None)
+
+ def test_keypair_list_no_options(self):
+ arglist = []
+ verifylist = [
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ # In base command class Lister in cliff, abstract method take_action()
+ # returns a tuple containing the column names and an iterable
+ # containing the data to be listed.
+ columns, data = self.cmd.take_action(parsed_args)
+
+ # Set expected values
+
+ self.keypairs_mock.list.assert_called_with()
+
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(tuple(self.data), tuple(data))
+
+
+class TestKeypairShow(TestKeypair):
+
+ keypair = compute_fakes.FakeKeypair.create_one_keypair()
+
+ def setUp(self):
+ super(TestKeypairShow, self).setUp()
+
+ self.keypairs_mock.get.return_value = self.keypair
+
+ self.cmd = keypair.ShowKeypair(self.app, None)
+
+ self.columns = (
+ "fingerprint",
+ "name",
+ "user_id"
+ )
+
+ self.data = (
+ self.keypair.fingerprint,
+ self.keypair.name,
+ self.keypair.user_id
+ )
+
+ def test_show_no_options(self):
+
+ arglist = []
+ verifylist = []
+
+ # Missing required args should boil here
+ self.assertRaises(tests_utils.ParserException, self.check_parser,
+ self.cmd, arglist, verifylist)
+
+ def test_keypair_show(self):
+ # overwrite the setup one because we want to omit private_key
+ self.keypair = compute_fakes.FakeKeypair.create_one_keypair(
+ no_pri=True)
+ self.keypairs_mock.get.return_value = self.keypair
+
+ self.data = (
+ self.keypair.fingerprint,
+ self.keypair.name,
+ self.keypair.user_id
+ )
+
+ arglist = [
+ self.keypair.name
+ ]
+ verifylist = [
+ ('name', self.keypair.name)
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ columns, data = self.cmd.take_action(parsed_args)
+
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(self.data, data)
+
+ def test_keypair_show_public(self):
+
+ arglist = [
+ '--public-key',
+ self.keypair.name
+ ]
+ verifylist = [
+ ('public_key', True),
+ ('name', self.keypair.name)
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ columns, data = self.cmd.take_action(parsed_args)
+
+ self.assertEqual({}, columns)
+ self.assertEqual({}, data)
diff --git a/openstackclient/tests/unit/compute/v2/test_server.py b/openstackclient/tests/unit/compute/v2/test_server.py
new file mode 100644
index 00000000..d4843f51
--- /dev/null
+++ b/openstackclient/tests/unit/compute/v2/test_server.py
@@ -0,0 +1,1844 @@
+# Copyright 2013 Nebula Inc.
+#
+# 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 collections
+import getpass
+import mock
+from mock import call
+
+from osc_lib import exceptions
+from osc_lib import utils as common_utils
+
+from openstackclient.compute.v2 import server
+from openstackclient.tests.unit.compute.v2 import fakes as compute_fakes
+from openstackclient.tests.unit.image.v2 import fakes as image_fakes
+from openstackclient.tests.unit import utils
+from openstackclient.tests.unit.volume.v2 import fakes as volume_fakes
+
+
+class TestServer(compute_fakes.TestComputev2):
+
+ def setUp(self):
+ super(TestServer, self).setUp()
+
+ # Get a shortcut to the compute client ServerManager Mock
+ self.servers_mock = self.app.client_manager.compute.servers
+ self.servers_mock.reset_mock()
+
+ # Get a shortcut to the compute client ImageManager Mock
+ self.cimages_mock = self.app.client_manager.compute.images
+ self.cimages_mock.reset_mock()
+
+ # Get a shortcut to the compute client FlavorManager Mock
+ self.flavors_mock = self.app.client_manager.compute.flavors
+ self.flavors_mock.reset_mock()
+
+ # Get a shortcut to the compute client SecurityGroupManager Mock
+ self.security_groups_mock = \
+ self.app.client_manager.compute.security_groups
+ self.security_groups_mock.reset_mock()
+
+ # Get a shortcut to the image client ImageManager Mock
+ self.images_mock = self.app.client_manager.image.images
+ self.images_mock.reset_mock()
+
+ # Get a shortcut to the volume client VolumeManager Mock
+ self.volumes_mock = self.app.client_manager.volume.volumes
+ self.volumes_mock.reset_mock()
+
+ # Set object attributes to be tested. Could be overwritten in subclass.
+ self.attrs = {}
+
+ # Set object methods to be tested. Could be overwritten in subclass.
+ self.methods = {}
+
+ def setup_servers_mock(self, count):
+ servers = compute_fakes.FakeServer.create_servers(attrs=self.attrs,
+ methods=self.methods,
+ count=count)
+
+ # This is the return value for utils.find_resource()
+ self.servers_mock.get = compute_fakes.FakeServer.get_servers(servers,
+ 0)
+ return servers
+
+ def run_method_with_servers(self, method_name, server_count):
+ servers = self.setup_servers_mock(server_count)
+
+ arglist = []
+ verifylist = []
+
+ for s in servers:
+ arglist.append(s.id)
+ verifylist = [
+ ('server', arglist),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ result = self.cmd.take_action(parsed_args)
+
+ for s in servers:
+ method = getattr(s, method_name)
+ method.assert_called_with()
+ self.assertIsNone(result)
+
+
+class TestServerAddFixedIP(TestServer):
+
+ def setUp(self):
+ super(TestServerAddFixedIP, self).setUp()
+
+ # Get a shortcut to the compute client ServerManager Mock
+ self.networks_mock = self.app.client_manager.compute.networks
+
+ # Get the command object to test
+ self.cmd = server.AddFixedIP(self.app, None)
+
+ # Set add_fixed_ip method to be tested.
+ self.methods = {
+ 'add_fixed_ip': None,
+ }
+
+ def test_server_add_fixed_ip(self):
+ servers = self.setup_servers_mock(count=1)
+ network = compute_fakes.FakeNetwork.create_one_network()
+ self.networks_mock.get.return_value = network
+
+ arglist = [
+ servers[0].id,
+ network.id,
+ ]
+ verifylist = [
+ ('server', servers[0].id),
+ ('network', network.id)
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ result = self.cmd.take_action(parsed_args)
+
+ servers[0].add_fixed_ip.assert_called_once_with(
+ network.id,
+ )
+ self.assertIsNone(result)
+
+
+class TestServerAddFloatingIP(TestServer):
+
+ def setUp(self):
+ super(TestServerAddFloatingIP, self).setUp()
+
+ # Get a shortcut to the compute client ServerManager Mock
+ self.networks_mock = self.app.client_manager.compute.networks
+
+ # Get the command object to test
+ self.cmd = server.AddFloatingIP(self.app, None)
+
+ # Set add_floating_ip method to be tested.
+ self.methods = {
+ 'add_floating_ip': None,
+ }
+
+ def test_server_add_floating_ip(self):
+ servers = self.setup_servers_mock(count=1)
+
+ arglist = [
+ servers[0].id,
+ '1.2.3.4',
+ ]
+ verifylist = [
+ ('server', servers[0].id),
+ ('ip_address', '1.2.3.4'),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ result = self.cmd.take_action(parsed_args)
+
+ servers[0].add_floating_ip.assert_called_once_with('1.2.3.4')
+ self.assertIsNone(result)
+
+
+class TestServerAddSecurityGroup(TestServer):
+
+ def setUp(self):
+ super(TestServerAddSecurityGroup, self).setUp()
+
+ self.security_group = \
+ compute_fakes.FakeSecurityGroup.create_one_security_group()
+ # This is the return value for utils.find_resource() for security group
+ self.security_groups_mock.get.return_value = self.security_group
+
+ attrs = {
+ 'security_groups': [{'name': self.security_group.id}]
+ }
+ methods = {
+ 'add_security_group': None,
+ }
+
+ self.server = compute_fakes.FakeServer.create_one_server(
+ attrs=attrs,
+ methods=methods
+ )
+ # This is the return value for utils.find_resource() for server
+ self.servers_mock.get.return_value = self.server
+
+ # Get the command object to test
+ self.cmd = server.AddServerSecurityGroup(self.app, None)
+
+ def test_server_add_security_group(self):
+ arglist = [
+ self.server.id,
+ self.security_group.id
+ ]
+ verifylist = [
+ ('server', self.server.id),
+ ('group', self.security_group.id),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ result = self.cmd.take_action(parsed_args)
+ self.security_groups_mock.get.assert_called_with(
+ self.security_group.id,
+ )
+ self.servers_mock.get.assert_called_with(self.server.id)
+ self.server.add_security_group.assert_called_with(
+ self.security_group.id,
+ )
+ self.assertIsNone(result)
+
+
+class TestServerCreate(TestServer):
+
+ columns = (
+ 'OS-EXT-STS:power_state',
+ 'addresses',
+ 'flavor',
+ 'id',
+ 'image',
+ 'name',
+ 'networks',
+ 'properties',
+ )
+
+ def datalist(self):
+ datalist = (
+ server._format_servers_list_power_state(
+ getattr(self.new_server, 'OS-EXT-STS:power_state')),
+ '',
+ self.flavor.name + ' (' + self.new_server.flavor.get('id') + ')',
+ self.new_server.id,
+ self.image.name + ' (' + self.new_server.image.get('id') + ')',
+ self.new_server.name,
+ self.new_server.networks,
+ '',
+ )
+ return datalist
+
+ def setUp(self):
+ super(TestServerCreate, self).setUp()
+
+ attrs = {
+ 'networks': {},
+ }
+ self.new_server = compute_fakes.FakeServer.create_one_server(
+ attrs=attrs)
+
+ # This is the return value for utils.find_resource().
+ # This is for testing --wait option.
+ self.servers_mock.get.return_value = self.new_server
+
+ self.servers_mock.create.return_value = self.new_server
+
+ self.image = image_fakes.FakeImage.create_one_image()
+ self.cimages_mock.get.return_value = self.image
+
+ self.flavor = compute_fakes.FakeFlavor.create_one_flavor()
+ self.flavors_mock.get.return_value = self.flavor
+
+ self.volume = volume_fakes.FakeVolume.create_one_volume()
+ self.volumes_mock.get.return_value = self.volume
+ self.block_device_mapping = 'vda=' + self.volume.name + ':::0'
+
+ # Get the command object to test
+ self.cmd = server.CreateServer(self.app, None)
+
+ def test_server_create_no_options(self):
+ arglist = [
+ self.new_server.name,
+ ]
+ verifylist = [
+ ('server_name', self.new_server.name),
+ ]
+
+ self.assertRaises(utils.ParserException, self.check_parser,
+ self.cmd, arglist, verifylist)
+
+ def test_server_create_minimal(self):
+ arglist = [
+ '--image', 'image1',
+ '--flavor', 'flavor1',
+ self.new_server.name,
+ ]
+ verifylist = [
+ ('image', 'image1'),
+ ('flavor', 'flavor1'),
+ ('config_drive', False),
+ ('server_name', self.new_server.name),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ # In base command class ShowOne in cliff, abstract method take_action()
+ # returns a two-part tuple with a tuple of column names and a tuple of
+ # data to be shown.
+ columns, data = self.cmd.take_action(parsed_args)
+
+ # Set expected values
+ kwargs = dict(
+ meta=None,
+ files={},
+ reservation_id=None,
+ min_count=1,
+ max_count=1,
+ security_groups=[],
+ userdata=None,
+ key_name=None,
+ availability_zone=None,
+ block_device_mapping={},
+ nics=[],
+ scheduler_hints={},
+ config_drive=None,
+ )
+ # ServerManager.create(name, image, flavor, **kwargs)
+ self.servers_mock.create.assert_called_with(
+ self.new_server.name,
+ self.image,
+ self.flavor,
+ **kwargs
+ )
+
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(self.datalist(), data)
+
+ def test_server_create_with_network(self):
+ arglist = [
+ '--image', 'image1',
+ '--flavor', 'flavor1',
+ '--nic', 'net-id=net1',
+ '--nic', 'port-id=port1',
+ self.new_server.name,
+ ]
+ verifylist = [
+ ('image', 'image1'),
+ ('flavor', 'flavor1'),
+ ('nic', ['net-id=net1', 'port-id=port1']),
+ ('config_drive', False),
+ ('server_name', self.new_server.name),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ get_endpoints = mock.Mock()
+ get_endpoints.return_value = {'network': []}
+ self.app.client_manager.auth_ref = mock.Mock()
+ self.app.client_manager.auth_ref.service_catalog = mock.Mock()
+ self.app.client_manager.auth_ref.service_catalog.get_endpoints = (
+ get_endpoints)
+
+ find_network = mock.Mock()
+ find_port = mock.Mock()
+ network_client = self.app.client_manager.network
+ network_client.find_network = find_network
+ network_client.find_port = find_port
+ network_resource = mock.Mock()
+ network_resource.id = 'net1_uuid'
+ port_resource = mock.Mock()
+ port_resource.id = 'port1_uuid'
+ find_network.return_value = network_resource
+ find_port.return_value = port_resource
+
+ # Mock sdk APIs.
+ _network = mock.Mock()
+ _network.id = 'net1_uuid'
+ _port = mock.Mock()
+ _port.id = 'port1_uuid'
+ find_network = mock.Mock()
+ find_port = mock.Mock()
+ find_network.return_value = _network
+ find_port.return_value = _port
+ self.app.client_manager.network.find_network = find_network
+ self.app.client_manager.network.find_port = find_port
+
+ # In base command class ShowOne in cliff, abstract method take_action()
+ # returns a two-part tuple with a tuple of column names and a tuple of
+ # data to be shown.
+ columns, data = self.cmd.take_action(parsed_args)
+
+ # Set expected values
+ kwargs = dict(
+ meta=None,
+ files={},
+ reservation_id=None,
+ min_count=1,
+ max_count=1,
+ security_groups=[],
+ userdata=None,
+ key_name=None,
+ availability_zone=None,
+ block_device_mapping={},
+ nics=[{'net-id': 'net1_uuid',
+ 'v4-fixed-ip': '',
+ 'v6-fixed-ip': '',
+ 'port-id': ''},
+ {'net-id': '',
+ 'v4-fixed-ip': '',
+ 'v6-fixed-ip': '',
+ 'port-id': 'port1_uuid'}],
+ scheduler_hints={},
+ config_drive=None,
+ )
+ # ServerManager.create(name, image, flavor, **kwargs)
+ self.servers_mock.create.assert_called_with(
+ self.new_server.name,
+ self.image,
+ self.flavor,
+ **kwargs
+ )
+
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(self.datalist(), data)
+
+ @mock.patch('openstackclient.compute.v2.server.io.open')
+ def test_server_create_userdata(self, mock_open):
+ mock_file = mock.MagicMock(name='File')
+ mock_open.return_value = mock_file
+ mock_open.read.return_value = '#!/bin/sh'
+
+ arglist = [
+ '--image', 'image1',
+ '--flavor', 'flavor1',
+ '--user-data', 'userdata.sh',
+ self.new_server.name,
+ ]
+ verifylist = [
+ ('image', 'image1'),
+ ('flavor', 'flavor1'),
+ ('user_data', 'userdata.sh'),
+ ('config_drive', False),
+ ('server_name', self.new_server.name),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ # In base command class ShowOne in cliff, abstract method take_action()
+ # returns a two-part tuple with a tuple of column names and a tuple of
+ # data to be shown.
+ columns, data = self.cmd.take_action(parsed_args)
+
+ # Ensure the userdata file is opened
+ mock_open.assert_called_with('userdata.sh')
+
+ # Ensure the userdata file is closed
+ mock_file.close.assert_called_with()
+
+ # Set expected values
+ kwargs = dict(
+ meta=None,
+ files={},
+ reservation_id=None,
+ min_count=1,
+ max_count=1,
+ security_groups=[],
+ userdata=mock_file,
+ key_name=None,
+ availability_zone=None,
+ block_device_mapping={},
+ nics=[],
+ scheduler_hints={},
+ config_drive=None,
+ )
+ # ServerManager.create(name, image, flavor, **kwargs)
+ self.servers_mock.create.assert_called_with(
+ self.new_server.name,
+ self.image,
+ self.flavor,
+ **kwargs
+ )
+
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(self.datalist(), data)
+
+ def test_server_create_with_block_device_mapping(self):
+ arglist = [
+ '--image', 'image1',
+ '--flavor', self.flavor.id,
+ '--block-device-mapping', self.block_device_mapping,
+ self.new_server.name,
+ ]
+ verifylist = [
+ ('image', 'image1'),
+ ('flavor', self.flavor.id),
+ ('block_device_mapping', [self.block_device_mapping]),
+ ('config_drive', False),
+ ('server_name', self.new_server.name),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ # CreateServer.take_action() returns two tuples
+ columns, data = self.cmd.take_action(parsed_args)
+
+ real_volume_mapping = (
+ (self.block_device_mapping.split('=', 1)[1]).replace(
+ self.volume.name,
+ self.volume.id))
+
+ # Set expected values
+ kwargs = dict(
+ meta=None,
+ files={},
+ reservation_id=None,
+ min_count=1,
+ max_count=1,
+ security_groups=[],
+ userdata=None,
+ key_name=None,
+ availability_zone=None,
+ block_device_mapping={
+ 'vda': real_volume_mapping
+ },
+ nics=[],
+ scheduler_hints={},
+ config_drive=None,
+ )
+ # ServerManager.create(name, image, flavor, **kwargs)
+ self.servers_mock.create.assert_called_with(
+ self.new_server.name,
+ self.image,
+ self.flavor,
+ **kwargs
+ )
+
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(self.datalist(), data)
+
+
+class TestServerDelete(TestServer):
+
+ def setUp(self):
+ super(TestServerDelete, self).setUp()
+
+ self.servers_mock.delete.return_value = None
+
+ # Get the command object to test
+ self.cmd = server.DeleteServer(self.app, None)
+
+ def test_server_delete_no_options(self):
+ servers = self.setup_servers_mock(count=1)
+
+ arglist = [
+ servers[0].id,
+ ]
+ verifylist = [
+ ('server', [servers[0].id]),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ result = self.cmd.take_action(parsed_args)
+
+ self.servers_mock.delete.assert_called_with(servers[0].id)
+ self.assertIsNone(result)
+
+ def test_server_delete_multi_servers(self):
+ servers = self.setup_servers_mock(count=3)
+
+ arglist = []
+ verifylist = []
+
+ for s in servers:
+ arglist.append(s.id)
+ verifylist = [
+ ('server', arglist),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ result = self.cmd.take_action(parsed_args)
+
+ calls = []
+ for s in servers:
+ calls.append(call(s.id))
+ self.servers_mock.delete.assert_has_calls(calls)
+ self.assertIsNone(result)
+
+ @mock.patch.object(common_utils, 'wait_for_delete', return_value=True)
+ def test_server_delete_wait_ok(self, mock_wait_for_delete):
+ servers = self.setup_servers_mock(count=1)
+
+ arglist = [
+ servers[0].id, '--wait'
+ ]
+ verifylist = [
+ ('server', [servers[0].id]),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ result = self.cmd.take_action(parsed_args)
+
+ self.servers_mock.delete.assert_called_with(servers[0].id)
+ mock_wait_for_delete.assert_called_once_with(
+ self.servers_mock,
+ servers[0].id,
+ callback=server._show_progress
+ )
+ self.assertIsNone(result)
+
+ @mock.patch.object(common_utils, 'wait_for_delete', return_value=False)
+ def test_server_delete_wait_fails(self, mock_wait_for_delete):
+ servers = self.setup_servers_mock(count=1)
+
+ arglist = [
+ servers[0].id, '--wait'
+ ]
+ verifylist = [
+ ('server', [servers[0].id]),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ self.assertRaises(SystemExit, self.cmd.take_action, parsed_args)
+
+ self.servers_mock.delete.assert_called_with(servers[0].id)
+ mock_wait_for_delete.assert_called_once_with(
+ self.servers_mock,
+ servers[0].id,
+ callback=server._show_progress
+ )
+
+
+class TestServerDumpCreate(TestServer):
+
+ def setUp(self):
+ super(TestServerDumpCreate, self).setUp()
+
+ # Get the command object to test
+ self.cmd = server.CreateServerDump(self.app, None)
+
+ # Set methods to be tested.
+ self.methods = {
+ 'trigger_crash_dump': None,
+ }
+
+ def test_server_dump_one_server(self):
+ self.run_method_with_servers('trigger_crash_dump', 1)
+
+ def test_server_dump_multi_servers(self):
+ self.run_method_with_servers('trigger_crash_dump', 3)
+
+
+class TestServerList(TestServer):
+
+ # Columns to be listed up.
+ columns = (
+ 'ID',
+ 'Name',
+ 'Status',
+ 'Networks',
+ 'Image Name',
+ )
+ columns_long = (
+ 'ID',
+ 'Name',
+ 'Status',
+ 'Task State',
+ 'Power State',
+ 'Networks',
+ 'Image Name',
+ 'Image ID',
+ 'Availability Zone',
+ 'Host',
+ 'Properties',
+ )
+
+ def setUp(self):
+ super(TestServerList, self).setUp()
+
+ self.search_opts = {
+ 'reservation_id': None,
+ 'ip': None,
+ 'ip6': None,
+ 'name': None,
+ 'instance_name': None,
+ 'status': None,
+ 'flavor': None,
+ 'image': None,
+ 'host': None,
+ 'tenant_id': None,
+ 'all_tenants': False,
+ 'user_id': None,
+ }
+
+ # Default params of the core function of the command in the case of no
+ # commandline option specified.
+ self.kwargs = {
+ 'search_opts': self.search_opts,
+ 'marker': None,
+ 'limit': None,
+ }
+
+ # The fake servers' attributes. Use the original attributes names in
+ # nova, not the ones printed by "server list" command.
+ self.attrs = {
+ 'status': 'ACTIVE',
+ 'OS-EXT-STS:task_state': 'None',
+ 'OS-EXT-STS:power_state': 0x01, # Running
+ 'networks': {
+ u'public': [u'10.20.30.40', u'2001:db8::5']
+ },
+ 'OS-EXT-AZ:availability_zone': 'availability-zone-xxx',
+ 'OS-EXT-SRV-ATTR:host': 'host-name-xxx',
+ 'Metadata': '',
+ }
+
+ # The servers to be listed.
+ self.servers = self.setup_servers_mock(3)
+ self.servers_mock.list.return_value = self.servers
+
+ self.image = image_fakes.FakeImage.create_one_image()
+ self.cimages_mock.get.return_value = self.image
+
+ self.flavor = compute_fakes.FakeFlavor.create_one_flavor()
+ self.flavors_mock.get.return_value = self.flavor
+
+ # Get the command object to test
+ self.cmd = server.ListServer(self.app, None)
+
+ # Prepare data returned by fake Nova API.
+ self.data = []
+ self.data_long = []
+
+ Image = collections.namedtuple('Image', 'id name')
+ self.images_mock.list.return_value = [
+ Image(id=s.image['id'], name=self.image.name)
+ for s in self.servers
+ ]
+
+ for s in self.servers:
+ self.data.append((
+ s.id,
+ s.name,
+ s.status,
+ server._format_servers_list_networks(s.networks),
+ self.image.name,
+ ))
+ self.data_long.append((
+ s.id,
+ s.name,
+ s.status,
+ getattr(s, 'OS-EXT-STS:task_state'),
+ server._format_servers_list_power_state(
+ getattr(s, 'OS-EXT-STS:power_state')
+ ),
+ server._format_servers_list_networks(s.networks),
+ self.image.name,
+ s.image['id'],
+ getattr(s, 'OS-EXT-AZ:availability_zone'),
+ getattr(s, 'OS-EXT-SRV-ATTR:host'),
+ s.Metadata,
+ ))
+
+ def test_server_list_no_option(self):
+ arglist = []
+ verifylist = [
+ ('all_projects', False),
+ ('long', False),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ columns, data = self.cmd.take_action(parsed_args)
+
+ self.servers_mock.list.assert_called_with(**self.kwargs)
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(tuple(self.data), tuple(data))
+
+ def test_server_list_long_option(self):
+ arglist = [
+ '--long',
+ ]
+ verifylist = [
+ ('all_projects', False),
+ ('long', True),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ columns, data = self.cmd.take_action(parsed_args)
+
+ self.servers_mock.list.assert_called_with(**self.kwargs)
+ self.assertEqual(self.columns_long, columns)
+ self.assertEqual(tuple(self.data_long), tuple(data))
+
+ def test_server_list_with_image(self):
+
+ arglist = [
+ '--image', self.image.id
+ ]
+ verifylist = [
+ ('image', self.image.id)
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ columns, data = self.cmd.take_action(parsed_args)
+
+ self.cimages_mock.get.assert_any_call(self.image.id)
+
+ self.search_opts['image'] = self.image.id
+ self.servers_mock.list.assert_called_with(**self.kwargs)
+
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(tuple(self.data), tuple(data))
+
+ def test_server_list_with_flavor(self):
+
+ arglist = [
+ '--flavor', self.flavor.id
+ ]
+ verifylist = [
+ ('flavor', self.flavor.id)
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ columns, data = self.cmd.take_action(parsed_args)
+
+ self.flavors_mock.get.assert_called_with(self.flavor.id)
+
+ self.search_opts['flavor'] = self.flavor.id
+ self.servers_mock.list.assert_called_with(**self.kwargs)
+
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(tuple(self.data), tuple(data))
+
+
+class TestServerLock(TestServer):
+
+ def setUp(self):
+ super(TestServerLock, self).setUp()
+
+ # Get the command object to test
+ self.cmd = server.LockServer(self.app, None)
+
+ # Set methods to be tested.
+ self.methods = {
+ 'lock': None,
+ }
+
+ def test_server_lock_one_server(self):
+ self.run_method_with_servers('lock', 1)
+
+ def test_server_lock_multi_servers(self):
+ self.run_method_with_servers('lock', 3)
+
+
+class TestServerPause(TestServer):
+
+ def setUp(self):
+ super(TestServerPause, self).setUp()
+
+ # Get the command object to test
+ self.cmd = server.PauseServer(self.app, None)
+
+ # Set methods to be tested.
+ self.methods = {
+ 'pause': None,
+ }
+
+ def test_server_pause_one_server(self):
+ self.run_method_with_servers('pause', 1)
+
+ def test_server_pause_multi_servers(self):
+ self.run_method_with_servers('pause', 3)
+
+
+class TestServerRebuild(TestServer):
+
+ def setUp(self):
+ super(TestServerRebuild, self).setUp()
+
+ # Return value for utils.find_resource for image
+ self.image = image_fakes.FakeImage.create_one_image()
+ self.cimages_mock.get.return_value = self.image
+
+ # Fake the rebuilt new server.
+ new_server = compute_fakes.FakeServer.create_one_server()
+
+ # Fake the server to be rebuilt. The IDs of them should be the same.
+ attrs = {
+ 'id': new_server.id,
+ 'image': {
+ 'id': self.image.id
+ },
+ 'networks': {},
+ 'adminPass': 'passw0rd',
+ }
+ methods = {
+ 'rebuild': new_server,
+ }
+ self.server = compute_fakes.FakeServer.create_one_server(
+ attrs=attrs,
+ methods=methods
+ )
+
+ # Return value for utils.find_resource for server.
+ self.servers_mock.get.return_value = self.server
+
+ self.cmd = server.RebuildServer(self.app, None)
+
+ def test_rebuild_with_current_image(self):
+ arglist = [
+ self.server.id,
+ ]
+ verifylist = [
+ ('server', self.server.id)
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ # Get the command object to test.
+ self.cmd.take_action(parsed_args)
+
+ self.servers_mock.get.assert_called_with(self.server.id)
+ self.cimages_mock.get.assert_called_with(self.image.id)
+ self.server.rebuild.assert_called_with(self.image, None)
+
+ def test_rebuild_with_current_image_and_password(self):
+ password = 'password-xxx'
+ arglist = [
+ self.server.id,
+ '--password', password
+ ]
+ verifylist = [
+ ('server', self.server.id),
+ ('password', password)
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ # Get the command object to test
+ self.cmd.take_action(parsed_args)
+
+ self.servers_mock.get.assert_called_with(self.server.id)
+ self.cimages_mock.get.assert_called_with(self.image.id)
+ self.server.rebuild.assert_called_with(self.image, password)
+
+ @mock.patch.object(common_utils, 'wait_for_status', return_value=True)
+ def test_rebuild_with_wait_ok(self, mock_wait_for_status):
+ arglist = [
+ '--wait',
+ self.server.id,
+ ]
+ verifylist = [
+ ('wait', True),
+ ('server', self.server.id),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ # Get the command object to test.
+ self.cmd.take_action(parsed_args)
+
+ # kwargs = dict(success_status=['active', 'verify_resize'],)
+
+ mock_wait_for_status.assert_called_once_with(
+ self.servers_mock.get,
+ self.server.id,
+ callback=server._show_progress,
+ # **kwargs
+ )
+
+ self.servers_mock.get.assert_called_with(self.server.id)
+ self.cimages_mock.get.assert_called_with(self.image.id)
+ self.server.rebuild.assert_called_with(self.image, None)
+
+ @mock.patch.object(common_utils, 'wait_for_status', return_value=False)
+ def test_rebuild_with_wait_fails(self, mock_wait_for_status):
+ arglist = [
+ '--wait',
+ self.server.id,
+ ]
+ verifylist = [
+ ('wait', True),
+ ('server', self.server.id),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ self.assertRaises(SystemExit, self.cmd.take_action, parsed_args)
+
+ mock_wait_for_status.assert_called_once_with(
+ self.servers_mock.get,
+ self.server.id,
+ callback=server._show_progress
+ )
+
+ self.servers_mock.get.assert_called_with(self.server.id)
+ self.cimages_mock.get.assert_called_with(self.image.id)
+ self.server.rebuild.assert_called_with(self.image, None)
+
+
+class TestServerRemoveFixedIP(TestServer):
+
+ def setUp(self):
+ super(TestServerRemoveFixedIP, self).setUp()
+
+ # Get the command object to test
+ self.cmd = server.RemoveFixedIP(self.app, None)
+
+ # Set unshelve method to be tested.
+ self.methods = {
+ 'remove_fixed_ip': None,
+ }
+
+ def test_server_remove_fixed_ip(self):
+ servers = self.setup_servers_mock(count=1)
+
+ arglist = [
+ servers[0].id,
+ '1.2.3.4',
+ ]
+ verifylist = [
+ ('server', servers[0].id),
+ ('ip_address', '1.2.3.4'),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ result = self.cmd.take_action(parsed_args)
+
+ servers[0].remove_fixed_ip.assert_called_once_with('1.2.3.4')
+ self.assertIsNone(result)
+
+
+class TestServerRemoveFloatingIP(TestServer):
+
+ def setUp(self):
+ super(TestServerRemoveFloatingIP, self).setUp()
+
+ # Get the command object to test
+ self.cmd = server.RemoveFloatingIP(self.app, None)
+
+ # Set unshelve method to be tested.
+ self.methods = {
+ 'remove_floating_ip': None,
+ }
+
+ def test_server_remove_floating_ip(self):
+ servers = self.setup_servers_mock(count=1)
+
+ arglist = [
+ servers[0].id,
+ '1.2.3.4',
+ ]
+ verifylist = [
+ ('server', servers[0].id),
+ ('ip_address', '1.2.3.4'),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ result = self.cmd.take_action(parsed_args)
+
+ servers[0].remove_floating_ip.assert_called_once_with('1.2.3.4')
+ self.assertIsNone(result)
+
+
+class TestServerRemoveSecurityGroup(TestServer):
+
+ def setUp(self):
+ super(TestServerRemoveSecurityGroup, self).setUp()
+
+ self.security_group = \
+ compute_fakes.FakeSecurityGroup.create_one_security_group()
+ # This is the return value for utils.find_resource() for security group
+ self.security_groups_mock.get.return_value = self.security_group
+
+ attrs = {
+ 'security_groups': [{'name': self.security_group.id}]
+ }
+ methods = {
+ 'remove_security_group': None,
+ }
+
+ self.server = compute_fakes.FakeServer.create_one_server(
+ attrs=attrs,
+ methods=methods
+ )
+ # This is the return value for utils.find_resource() for server
+ self.servers_mock.get.return_value = self.server
+
+ # Get the command object to test
+ self.cmd = server.RemoveServerSecurityGroup(self.app, None)
+
+ def test_server_remove_security_group(self):
+ arglist = [
+ self.server.id,
+ self.security_group.id
+ ]
+ verifylist = [
+ ('server', self.server.id),
+ ('group', self.security_group.id),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ result = self.cmd.take_action(parsed_args)
+ self.security_groups_mock.get.assert_called_with(
+ self.security_group.id,
+ )
+ self.servers_mock.get.assert_called_with(self.server.id)
+ self.server.remove_security_group.assert_called_with(
+ self.security_group.id,
+ )
+ self.assertIsNone(result)
+
+
+class TestServerResize(TestServer):
+
+ def setUp(self):
+ super(TestServerResize, self).setUp()
+
+ self.server = compute_fakes.FakeServer.create_one_server()
+
+ # This is the return value for utils.find_resource()
+ self.servers_mock.get.return_value = self.server
+
+ self.servers_mock.resize.return_value = None
+ self.servers_mock.confirm_resize.return_value = None
+ self.servers_mock.revert_resize.return_value = None
+
+ # This is the return value for utils.find_resource()
+ self.flavors_get_return_value = \
+ compute_fakes.FakeFlavor.create_one_flavor()
+ self.flavors_mock.get.return_value = self.flavors_get_return_value
+
+ # Get the command object to test
+ self.cmd = server.ResizeServer(self.app, None)
+
+ def test_server_resize_no_options(self):
+ arglist = [
+ self.server.id,
+ ]
+ verifylist = [
+ ('confirm', False),
+ ('revert', False),
+ ('server', self.server.id),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ result = self.cmd.take_action(parsed_args)
+
+ self.servers_mock.get.assert_called_with(self.server.id)
+
+ self.assertNotCalled(self.servers_mock.resize)
+ self.assertNotCalled(self.servers_mock.confirm_resize)
+ self.assertNotCalled(self.servers_mock.revert_resize)
+ self.assertIsNone(result)
+
+ def test_server_resize(self):
+ arglist = [
+ '--flavor', self.flavors_get_return_value.id,
+ self.server.id,
+ ]
+ verifylist = [
+ ('flavor', self.flavors_get_return_value.id),
+ ('confirm', False),
+ ('revert', False),
+ ('server', self.server.id),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ result = self.cmd.take_action(parsed_args)
+
+ self.servers_mock.get.assert_called_with(self.server.id)
+ self.flavors_mock.get.assert_called_with(
+ self.flavors_get_return_value.id,
+ )
+ self.servers_mock.resize.assert_called_with(
+ self.server,
+ self.flavors_get_return_value,
+ )
+ self.assertNotCalled(self.servers_mock.confirm_resize)
+ self.assertNotCalled(self.servers_mock.revert_resize)
+ self.assertIsNone(result)
+
+ def test_server_resize_confirm(self):
+ arglist = [
+ '--confirm',
+ self.server.id,
+ ]
+ verifylist = [
+ ('confirm', True),
+ ('revert', False),
+ ('server', self.server.id),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ result = self.cmd.take_action(parsed_args)
+
+ self.servers_mock.get.assert_called_with(self.server.id)
+ self.assertNotCalled(self.servers_mock.resize)
+ self.servers_mock.confirm_resize.assert_called_with(self.server)
+ self.assertNotCalled(self.servers_mock.revert_resize)
+ self.assertIsNone(result)
+
+ def test_server_resize_revert(self):
+ arglist = [
+ '--revert',
+ self.server.id,
+ ]
+ verifylist = [
+ ('confirm', False),
+ ('revert', True),
+ ('server', self.server.id),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ result = self.cmd.take_action(parsed_args)
+
+ self.servers_mock.get.assert_called_with(self.server.id)
+ self.assertNotCalled(self.servers_mock.resize)
+ self.assertNotCalled(self.servers_mock.confirm_resize)
+ self.servers_mock.revert_resize.assert_called_with(self.server)
+ self.assertIsNone(result)
+
+ @mock.patch.object(common_utils, 'wait_for_status', return_value=True)
+ def test_server_resize_with_wait_ok(self, mock_wait_for_status):
+
+ arglist = [
+ '--flavor', self.flavors_get_return_value.id,
+ '--wait',
+ self.server.id,
+ ]
+
+ verifylist = [
+ ('flavor', self.flavors_get_return_value.id),
+ ('confirm', False),
+ ('revert', False),
+ ('wait', True),
+ ('server', self.server.id),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ self.cmd.take_action(parsed_args)
+
+ self.servers_mock.get.assert_called_with(
+ self.server.id,
+ )
+
+ kwargs = dict(success_status=['active', 'verify_resize'],)
+
+ mock_wait_for_status.assert_called_once_with(
+ self.servers_mock.get,
+ self.server.id,
+ callback=server._show_progress,
+ **kwargs
+ )
+
+ self.servers_mock.resize.assert_called_with(
+ self.server,
+ self.flavors_get_return_value
+ )
+ self.assertNotCalled(self.servers_mock.confirm_resize)
+ self.assertNotCalled(self.servers_mock.revert_resize)
+
+ @mock.patch.object(common_utils, 'wait_for_status', return_value=False)
+ def test_server_resize_with_wait_fails(self, mock_wait_for_status):
+
+ arglist = [
+ '--flavor', self.flavors_get_return_value.id,
+ '--wait',
+ self.server.id,
+ ]
+
+ verifylist = [
+ ('flavor', self.flavors_get_return_value.id),
+ ('confirm', False),
+ ('revert', False),
+ ('wait', True),
+ ('server', self.server.id),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ self.assertRaises(SystemExit, self.cmd.take_action, parsed_args)
+
+ self.servers_mock.get.assert_called_with(
+ self.server.id,
+ )
+
+ kwargs = dict(success_status=['active', 'verify_resize'],)
+
+ mock_wait_for_status.assert_called_once_with(
+ self.servers_mock.get,
+ self.server.id,
+ callback=server._show_progress,
+ **kwargs
+ )
+
+ self.servers_mock.resize.assert_called_with(
+ self.server,
+ self.flavors_get_return_value
+ )
+
+
+class TestServerRestore(TestServer):
+
+ def setUp(self):
+ super(TestServerRestore, self).setUp()
+
+ # Get the command object to test
+ self.cmd = server.RestoreServer(self.app, None)
+
+ # Set methods to be tested.
+ self.methods = {
+ 'restore': None,
+ }
+
+ def test_server_restore_one_server(self):
+ self.run_method_with_servers('restore', 1)
+
+ def test_server_restore_multi_servers(self):
+ self.run_method_with_servers('restore', 3)
+
+
+class TestServerResume(TestServer):
+
+ def setUp(self):
+ super(TestServerResume, self).setUp()
+
+ # Get the command object to test
+ self.cmd = server.ResumeServer(self.app, None)
+
+ # Set methods to be tested.
+ self.methods = {
+ 'resume': None,
+ }
+
+ def test_server_resume_one_server(self):
+ self.run_method_with_servers('resume', 1)
+
+ def test_server_resume_multi_servers(self):
+ self.run_method_with_servers('resume', 3)
+
+
+class TestServerSet(TestServer):
+
+ def setUp(self):
+ super(TestServerSet, self).setUp()
+
+ self.methods = {
+ 'update': None,
+ 'reset_state': None,
+ 'change_password': None,
+ }
+
+ self.fake_servers = self.setup_servers_mock(2)
+
+ # Get the command object to test
+ self.cmd = server.SetServer(self.app, None)
+
+ def test_server_set_no_option(self):
+ arglist = [
+ 'foo_vm'
+ ]
+ verifylist = [
+ ('server', 'foo_vm')
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ result = self.cmd.take_action(parsed_args)
+ self.assertNotCalled(self.fake_servers[0].update)
+ self.assertNotCalled(self.fake_servers[0].reset_state)
+ self.assertNotCalled(self.fake_servers[0].change_password)
+ self.assertNotCalled(self.servers_mock.set_meta)
+ self.assertIsNone(result)
+
+ def test_server_set_with_state(self):
+ for index, state in enumerate(['active', 'error']):
+ arglist = [
+ '--state', state,
+ 'foo_vm',
+ ]
+ verifylist = [
+ ('state', state),
+ ('server', 'foo_vm'),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ result = self.cmd.take_action(parsed_args)
+ self.fake_servers[index].reset_state.assert_called_once_with(
+ state=state)
+ self.assertIsNone(result)
+
+ def test_server_set_with_invalid_state(self):
+ arglist = [
+ '--state', 'foo_state',
+ 'foo_vm',
+ ]
+ verifylist = [
+ ('state', 'foo_state'),
+ ('server', 'foo_vm'),
+ ]
+ self.assertRaises(utils.ParserException,
+ self.check_parser,
+ self.cmd, arglist, verifylist)
+
+ def test_server_set_with_name(self):
+ arglist = [
+ '--name', 'foo_name',
+ 'foo_vm',
+ ]
+ verifylist = [
+ ('name', 'foo_name'),
+ ('server', 'foo_vm'),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ result = self.cmd.take_action(parsed_args)
+ self.fake_servers[0].update.assert_called_once_with(name='foo_name')
+ self.assertIsNone(result)
+
+ def test_server_set_with_property(self):
+ arglist = [
+ '--property', 'key1=value1',
+ '--property', 'key2=value2',
+ 'foo_vm',
+ ]
+ verifylist = [
+ ('property', {'key1': 'value1', 'key2': 'value2'}),
+ ('server', 'foo_vm'),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ result = self.cmd.take_action(parsed_args)
+ self.servers_mock.set_meta.assert_called_once_with(
+ self.fake_servers[0], parsed_args.property)
+ self.assertIsNone(result)
+
+ @mock.patch.object(getpass, 'getpass',
+ return_value=mock.sentinel.fake_pass)
+ def test_server_set_with_root_password(self, mock_getpass):
+ arglist = [
+ '--root-password',
+ 'foo_vm',
+ ]
+ verifylist = [
+ ('root_password', True),
+ ('server', 'foo_vm'),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ result = self.cmd.take_action(parsed_args)
+ self.fake_servers[0].change_password.assert_called_once_with(
+ mock.sentinel.fake_pass)
+ self.assertIsNone(result)
+
+
+class TestServerShelve(TestServer):
+
+ def setUp(self):
+ super(TestServerShelve, self).setUp()
+
+ # Get the command object to test
+ self.cmd = server.ShelveServer(self.app, None)
+
+ # Set shelve method to be tested.
+ self.methods = {
+ 'shelve': None,
+ }
+
+ def test_shelve_one_server(self):
+ self.run_method_with_servers('shelve', 1)
+
+ def test_shelve_multi_servers(self):
+ self.run_method_with_servers('shelve', 3)
+
+
+class TestServerShow(TestServer):
+
+ def setUp(self):
+ super(TestServerShow, self).setUp()
+
+ self.image = image_fakes.FakeImage.create_one_image()
+ self.flavor = compute_fakes.FakeFlavor.create_one_flavor()
+ server_info = {
+ 'image': {'id': self.image.id},
+ 'flavor': {'id': self.flavor.id},
+ 'tenant_id': 'tenant-id-xxx',
+ 'networks': {'public': ['10.20.30.40', '2001:db8::f']},
+ }
+ # Fake the server.diagnostics() method. The return value contains http
+ # response and data. The data is a dict. Sincce this method itself is
+ # faked, we don't need to fake everything of the return value exactly.
+ resp = mock.Mock()
+ resp.status_code = 200
+ server_method = {
+ 'diagnostics': (resp, {'test': 'test'}),
+ }
+ self.server = compute_fakes.FakeServer.create_one_server(
+ attrs=server_info, methods=server_method)
+
+ # This is the return value for utils.find_resource()
+ self.servers_mock.get.return_value = self.server
+ self.cimages_mock.get.return_value = self.image
+ self.flavors_mock.get.return_value = self.flavor
+
+ # Get the command object to test
+ self.cmd = server.ShowServer(self.app, None)
+
+ self.columns = (
+ 'OS-EXT-STS:power_state',
+ 'addresses',
+ 'flavor',
+ 'id',
+ 'image',
+ 'name',
+ 'networks',
+ 'project_id',
+ 'properties',
+ )
+
+ self.data = (
+ 'Running',
+ 'public=10.20.30.40, 2001:db8::f',
+ self.flavor.name + " (" + self.flavor.id + ")",
+ self.server.id,
+ self.image.name + " (" + self.image.id + ")",
+ self.server.name,
+ {'public': ['10.20.30.40', '2001:db8::f']},
+ 'tenant-id-xxx',
+ '',
+ )
+
+ def test_show_no_options(self):
+ arglist = []
+ verifylist = []
+
+ self.assertRaises(utils.ParserException, self.check_parser,
+ self.cmd, arglist, verifylist)
+
+ def test_show(self):
+ arglist = [
+ self.server.name,
+ ]
+ verifylist = [
+ ('diagnostics', False),
+ ('server', self.server.name),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ columns, data = self.cmd.take_action(parsed_args)
+
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(self.data, data)
+
+ def test_show_diagnostics(self):
+ arglist = [
+ '--diagnostics',
+ self.server.name,
+ ]
+ verifylist = [
+ ('diagnostics', True),
+ ('server', self.server.name),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ columns, data = self.cmd.take_action(parsed_args)
+
+ self.assertEqual(('test',), columns)
+ self.assertEqual(('test',), data)
+
+
+class TestServerStart(TestServer):
+
+ def setUp(self):
+ super(TestServerStart, self).setUp()
+
+ # Get the command object to test
+ self.cmd = server.StartServer(self.app, None)
+
+ # Set methods to be tested.
+ self.methods = {
+ 'start': None,
+ }
+
+ def test_server_start_one_server(self):
+ self.run_method_with_servers('start', 1)
+
+ def test_server_start_multi_servers(self):
+ self.run_method_with_servers('start', 3)
+
+
+class TestServerStop(TestServer):
+
+ def setUp(self):
+ super(TestServerStop, self).setUp()
+
+ # Get the command object to test
+ self.cmd = server.StopServer(self.app, None)
+
+ # Set methods to be tested.
+ self.methods = {
+ 'stop': None,
+ }
+
+ def test_server_stop_one_server(self):
+ self.run_method_with_servers('stop', 1)
+
+ def test_server_stop_multi_servers(self):
+ self.run_method_with_servers('stop', 3)
+
+
+class TestServerSuspend(TestServer):
+
+ def setUp(self):
+ super(TestServerSuspend, self).setUp()
+
+ # Get the command object to test
+ self.cmd = server.SuspendServer(self.app, None)
+
+ # Set methods to be tested.
+ self.methods = {
+ 'suspend': None,
+ }
+
+ def test_server_suspend_one_server(self):
+ self.run_method_with_servers('suspend', 1)
+
+ def test_server_suspend_multi_servers(self):
+ self.run_method_with_servers('suspend', 3)
+
+
+class TestServerUnlock(TestServer):
+
+ def setUp(self):
+ super(TestServerUnlock, self).setUp()
+
+ # Get the command object to test
+ self.cmd = server.UnlockServer(self.app, None)
+
+ # Set methods to be tested.
+ self.methods = {
+ 'unlock': None,
+ }
+
+ def test_server_unlock_one_server(self):
+ self.run_method_with_servers('unlock', 1)
+
+ def test_server_unlock_multi_servers(self):
+ self.run_method_with_servers('unlock', 3)
+
+
+class TestServerUnpause(TestServer):
+
+ def setUp(self):
+ super(TestServerUnpause, self).setUp()
+
+ # Get the command object to test
+ self.cmd = server.UnpauseServer(self.app, None)
+
+ # Set methods to be tested.
+ self.methods = {
+ 'unpause': None,
+ }
+
+ def test_server_unpause_one_server(self):
+ self.run_method_with_servers('unpause', 1)
+
+ def test_server_unpause_multi_servers(self):
+ self.run_method_with_servers('unpause', 3)
+
+
+class TestServerUnset(TestServer):
+
+ def setUp(self):
+ super(TestServerUnset, self).setUp()
+
+ self.fake_server = self.setup_servers_mock(1)[0]
+
+ # Get the command object to test
+ self.cmd = server.UnsetServer(self.app, None)
+
+ def test_server_unset_no_option(self):
+ arglist = [
+ 'foo_vm',
+ ]
+ verifylist = [
+ ('server', 'foo_vm'),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ result = self.cmd.take_action(parsed_args)
+ self.assertNotCalled(self.servers_mock.delete_meta)
+ self.assertIsNone(result)
+
+ def test_server_unset_with_property(self):
+ arglist = [
+ '--property', 'key1',
+ '--property', 'key2',
+ 'foo_vm',
+ ]
+ verifylist = [
+ ('property', ['key1', 'key2']),
+ ('server', 'foo_vm'),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ result = self.cmd.take_action(parsed_args)
+ self.servers_mock.delete_meta.assert_called_once_with(
+ self.fake_server, ['key1', 'key2'])
+ self.assertIsNone(result)
+
+
+class TestServerUnshelve(TestServer):
+
+ def setUp(self):
+ super(TestServerUnshelve, self).setUp()
+
+ # Get the command object to test
+ self.cmd = server.UnshelveServer(self.app, None)
+
+ # Set unshelve method to be tested.
+ self.methods = {
+ 'unshelve': None,
+ }
+
+ def test_unshelve_one_server(self):
+ self.run_method_with_servers('unshelve', 1)
+
+ def test_unshelve_multi_servers(self):
+ self.run_method_with_servers('unshelve', 3)
+
+
+class TestServerGeneral(TestServer):
+ OLD = {
+ 'private': [
+ {
+ 'addr': '192.168.0.3',
+ 'version': 4,
+ },
+ ]
+ }
+ NEW = {
+ 'foo': [
+ {
+ 'OS-EXT-IPS-MAC:mac_addr': 'fa:16:3e:93:b3:01',
+ 'version': 4,
+ 'addr': '10.10.1.2',
+ 'OS-EXT-IPS:type': 'fixed',
+ },
+ {
+ 'OS-EXT-IPS-MAC:mac_addr': 'fa:16:3e:93:b3:02',
+ 'version': 6,
+ 'addr': '0:0:0:0:0:ffff:a0a:103',
+ 'OS-EXT-IPS:type': 'floating',
+ },
+ ]
+ }
+ ODD = {'jenkins': ['10.3.3.18', '124.12.125.4']}
+
+ def test_get_ip_address(self):
+ self.assertEqual("192.168.0.3",
+ server._get_ip_address(self.OLD, 'private', [4, 6]))
+ self.assertEqual("10.10.1.2",
+ server._get_ip_address(self.NEW, 'fixed', [4, 6]))
+ self.assertEqual("10.10.1.2",
+ server._get_ip_address(self.NEW, 'private', [4, 6]))
+ self.assertEqual("0:0:0:0:0:ffff:a0a:103",
+ server._get_ip_address(self.NEW, 'public', [6]))
+ self.assertEqual("0:0:0:0:0:ffff:a0a:103",
+ server._get_ip_address(self.NEW, 'floating', [6]))
+ self.assertEqual("124.12.125.4",
+ server._get_ip_address(self.ODD, 'public', [4, 6]))
+ self.assertEqual("10.3.3.18",
+ server._get_ip_address(self.ODD, 'private', [4, 6]))
+ self.assertRaises(exceptions.CommandError,
+ server._get_ip_address, self.NEW, 'public', [4])
+ self.assertRaises(exceptions.CommandError,
+ server._get_ip_address, self.NEW, 'admin', [4])
+ self.assertRaises(exceptions.CommandError,
+ server._get_ip_address, self.OLD, 'public', [4, 6])
+ self.assertRaises(exceptions.CommandError,
+ server._get_ip_address, self.OLD, 'private', [6])
+
+ def test_format_servers_list_power_state(self):
+ self.assertEqual("NOSTATE",
+ server._format_servers_list_power_state(0x00))
+ self.assertEqual("Running",
+ server._format_servers_list_power_state(0x01))
+ self.assertEqual("",
+ server._format_servers_list_power_state(0x02))
+ self.assertEqual("Paused",
+ server._format_servers_list_power_state(0x03))
+ self.assertEqual("Shutdown",
+ server._format_servers_list_power_state(0x04))
+ self.assertEqual("",
+ server._format_servers_list_power_state(0x05))
+ self.assertEqual("Crashed",
+ server._format_servers_list_power_state(0x06))
+ self.assertEqual("Suspended",
+ server._format_servers_list_power_state(0x07))
+ self.assertEqual("N/A",
+ server._format_servers_list_power_state(0x08))
+
+ def test_format_servers_list_networks(self):
+ # Setup network info to test.
+ networks = {
+ u'public': [u'10.20.30.40', u'2001:db8::f'],
+ u'private': [u'2001:db8::f', u'10.20.30.40'],
+ }
+
+ # Prepare expected data.
+ # Since networks is a dict, whose items are in random order, there
+ # could be two results after formatted.
+ data_1 = (u'private=2001:db8::f, 10.20.30.40; '
+ u'public=10.20.30.40, 2001:db8::f')
+ data_2 = (u'public=10.20.30.40, 2001:db8::f; '
+ u'private=2001:db8::f, 10.20.30.40')
+
+ # Call _format_servers_list_networks().
+ networks_format = server._format_servers_list_networks(networks)
+
+ msg = ('Network string is not formatted correctly.\n'
+ 'reference = %s or %s\n'
+ 'actual = %s\n' %
+ (data_1, data_2, networks_format))
+ self.assertIn(networks_format, (data_1, data_2), msg)
+
+ @mock.patch('osc_lib.utils.find_resource')
+ def test_prep_server_detail(self, find_resource):
+ # Setup mock method return value. utils.find_resource() will be called
+ # three times in _prep_server_detail():
+ # - The first time, return server info.
+ # - The second time, return image info.
+ # - The third time, return flavor info.
+ _image = image_fakes.FakeImage.create_one_image()
+ _flavor = compute_fakes.FakeFlavor.create_one_flavor()
+ server_info = {
+ 'image': {u'id': _image.id},
+ 'flavor': {u'id': _flavor.id},
+ 'tenant_id': u'tenant-id-xxx',
+ 'networks': {u'public': [u'10.20.30.40', u'2001:db8::f']},
+ 'links': u'http://xxx.yyy.com',
+ }
+ _server = compute_fakes.FakeServer.create_one_server(attrs=server_info)
+ find_resource.side_effect = [_server, _image, _flavor]
+
+ # Prepare result data.
+ info = {
+ 'id': _server.id,
+ 'name': _server.name,
+ 'addresses': u'public=10.20.30.40, 2001:db8::f',
+ 'flavor': u'%s (%s)' % (_flavor.name, _flavor.id),
+ 'image': u'%s (%s)' % (_image.name, _image.id),
+ 'project_id': u'tenant-id-xxx',
+ 'properties': '',
+ 'OS-EXT-STS:power_state': server._format_servers_list_power_state(
+ getattr(_server, 'OS-EXT-STS:power_state')),
+ }
+
+ # Call _prep_server_detail().
+ server_detail = server._prep_server_detail(
+ self.app.client_manager.compute,
+ _server
+ )
+ # 'networks' is used to create _server. Remove it.
+ server_detail.pop('networks')
+
+ # Check the results.
+ self.assertEqual(info, server_detail)
diff --git a/openstackclient/tests/unit/compute/v2/test_server_backup.py b/openstackclient/tests/unit/compute/v2/test_server_backup.py
new file mode 100644
index 00000000..9aa63fc7
--- /dev/null
+++ b/openstackclient/tests/unit/compute/v2/test_server_backup.py
@@ -0,0 +1,271 @@
+# 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 osc_lib import exceptions
+from osc_lib import utils as common_utils
+
+from openstackclient.compute.v2 import server_backup
+from openstackclient.tests.unit.compute.v2 import fakes as compute_fakes
+from openstackclient.tests.unit.image.v2 import fakes as image_fakes
+
+
+class TestServerBackup(compute_fakes.TestComputev2):
+
+ def setUp(self):
+ super(TestServerBackup, self).setUp()
+
+ # Get a shortcut to the compute client ServerManager Mock
+ self.servers_mock = self.app.client_manager.compute.servers
+ self.servers_mock.reset_mock()
+
+ # Get a shortcut to the image client ImageManager Mock
+ self.images_mock = self.app.client_manager.image.images
+ self.images_mock.reset_mock()
+
+ # Set object attributes to be tested. Could be overwritten in subclass.
+ self.attrs = {}
+
+ # Set object methods to be tested. Could be overwritten in subclass.
+ self.methods = {}
+
+ def setup_servers_mock(self, count):
+ servers = compute_fakes.FakeServer.create_servers(
+ attrs=self.attrs,
+ methods=self.methods,
+ count=count,
+ )
+
+ # This is the return value for utils.find_resource()
+ self.servers_mock.get = compute_fakes.FakeServer.get_servers(
+ servers,
+ 0,
+ )
+ return servers
+
+
+class TestServerBackupCreate(TestServerBackup):
+
+ # Just return whatever Image is testing with these days
+ def image_columns(self, image):
+ columnlist = tuple(sorted(image.keys()))
+ return columnlist
+
+ def image_data(self, image):
+ datalist = (
+ image['id'],
+ image['name'],
+ image['owner'],
+ image['protected'],
+ 'active',
+ common_utils.format_list(image.get('tags')),
+ image['visibility'],
+ )
+ return datalist
+
+ def setUp(self):
+ super(TestServerBackupCreate, self).setUp()
+
+ # Get the command object to test
+ self.cmd = server_backup.CreateServerBackup(self.app, None)
+
+ self.methods = {
+ 'backup': None,
+ }
+
+ def setup_images_mock(self, count, servers=None):
+ if servers:
+ images = image_fakes.FakeImage.create_images(
+ attrs={
+ 'name': servers[0].name,
+ 'status': 'active',
+ },
+ count=count,
+ )
+ else:
+ images = image_fakes.FakeImage.create_images(
+ attrs={
+ 'status': 'active',
+ },
+ count=count,
+ )
+
+ self.images_mock.get = mock.MagicMock(side_effect=images)
+ return images
+
+ def test_server_backup_defaults(self):
+ servers = self.setup_servers_mock(count=1)
+ images = self.setup_images_mock(count=1, servers=servers)
+
+ arglist = [
+ servers[0].id,
+ ]
+ verifylist = [
+ ('name', None),
+ ('type', None),
+ ('rotate', None),
+ ('wait', False),
+ ('server', servers[0].id),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ # In base command class ShowOne in cliff, abstract method take_action()
+ # returns a two-part tuple with a tuple of column names and a tuple of
+ # data to be shown.
+ columns, data = self.cmd.take_action(parsed_args)
+
+ # ServerManager.backup(server, backup_name, backup_type, rotation)
+ self.servers_mock.backup.assert_called_with(
+ servers[0].id,
+ servers[0].name,
+ '',
+ 1,
+ )
+
+ self.assertEqual(self.image_columns(images[0]), columns)
+ self.assertEqual(self.image_data(images[0]), data)
+
+ def test_server_backup_create_options(self):
+ servers = self.setup_servers_mock(count=1)
+ images = self.setup_images_mock(count=1, servers=servers)
+
+ arglist = [
+ '--name', 'image',
+ '--type', 'daily',
+ '--rotate', '2',
+ servers[0].id,
+ ]
+ verifylist = [
+ ('name', 'image'),
+ ('type', 'daily'),
+ ('rotate', 2),
+ ('server', servers[0].id),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ # In base command class ShowOne in cliff, abstract method take_action()
+ # returns a two-part tuple with a tuple of column names and a tuple of
+ # data to be shown.
+ columns, data = self.cmd.take_action(parsed_args)
+
+ # ServerManager.backup(server, backup_name, backup_type, rotation)
+ self.servers_mock.backup.assert_called_with(
+ servers[0].id,
+ 'image',
+ 'daily',
+ 2,
+ )
+
+ self.assertEqual(self.image_columns(images[0]), columns)
+ self.assertEqual(self.image_data(images[0]), data)
+
+ @mock.patch.object(common_utils, 'wait_for_status', return_value=False)
+ def test_server_backup_wait_fail(self, mock_wait_for_status):
+ servers = self.setup_servers_mock(count=1)
+ images = image_fakes.FakeImage.create_images(
+ attrs={
+ 'name': servers[0].name,
+ 'status': 'active',
+ },
+ count=5,
+ )
+
+ self.images_mock.get = mock.MagicMock(
+ side_effect=images,
+ )
+
+ arglist = [
+ '--name', 'image',
+ '--type', 'daily',
+ '--wait',
+ servers[0].id,
+ ]
+ verifylist = [
+ ('name', 'image'),
+ ('type', 'daily'),
+ ('wait', True),
+ ('server', servers[0].id),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ self.assertRaises(
+ exceptions.CommandError,
+ self.cmd.take_action,
+ parsed_args,
+ )
+
+ # ServerManager.backup(server, backup_name, backup_type, rotation)
+ self.servers_mock.backup.assert_called_with(
+ servers[0].id,
+ 'image',
+ 'daily',
+ 1,
+ )
+
+ mock_wait_for_status.assert_called_once_with(
+ self.images_mock.get,
+ images[0].id,
+ callback=mock.ANY
+ )
+
+ @mock.patch.object(common_utils, 'wait_for_status', return_value=True)
+ def test_server_backup_wait_ok(self, mock_wait_for_status):
+ servers = self.setup_servers_mock(count=1)
+ images = image_fakes.FakeImage.create_images(
+ attrs={
+ 'name': servers[0].name,
+ 'status': 'active',
+ },
+ count=5,
+ )
+
+ self.images_mock.get = mock.MagicMock(
+ side_effect=images,
+ )
+
+ arglist = [
+ '--name', 'image',
+ '--type', 'daily',
+ '--wait',
+ servers[0].id,
+ ]
+ verifylist = [
+ ('name', 'image'),
+ ('type', 'daily'),
+ ('wait', True),
+ ('server', servers[0].id),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ # In base command class ShowOne in cliff, abstract method take_action()
+ # returns a two-part tuple with a tuple of column names and a tuple of
+ # data to be shown.
+ columns, data = self.cmd.take_action(parsed_args)
+
+ # ServerManager.backup(server, backup_name, backup_type, rotation)
+ self.servers_mock.backup.assert_called_with(
+ servers[0].id,
+ 'image',
+ 'daily',
+ 1,
+ )
+
+ mock_wait_for_status.assert_called_once_with(
+ self.images_mock.get,
+ images[0].id,
+ callback=mock.ANY
+ )
+
+ self.assertEqual(self.image_columns(images[0]), columns)
+ self.assertEqual(self.image_data(images[0]), data)
diff --git a/openstackclient/tests/unit/compute/v2/test_server_group.py b/openstackclient/tests/unit/compute/v2/test_server_group.py
new file mode 100644
index 00000000..d474f41d
--- /dev/null
+++ b/openstackclient/tests/unit/compute/v2/test_server_group.py
@@ -0,0 +1,284 @@
+# Copyright 2016 Huawei, Inc. 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 mock
+
+from osc_lib import exceptions
+from osc_lib import utils
+
+from openstackclient.compute.v2 import server_group
+from openstackclient.tests.unit.compute.v2 import fakes as compute_fakes
+from openstackclient.tests.unit import utils as tests_utils
+
+
+class TestServerGroup(compute_fakes.TestComputev2):
+
+ fake_server_group = compute_fakes.FakeServerGroup.create_one_server_group()
+
+ columns = (
+ 'id',
+ 'members',
+ 'name',
+ 'policies',
+ 'project_id',
+ 'user_id',
+ )
+
+ data = (
+ fake_server_group.id,
+ utils.format_list(fake_server_group.members),
+ fake_server_group.name,
+ utils.format_list(fake_server_group.policies),
+ fake_server_group.project_id,
+ fake_server_group.user_id,
+ )
+
+ def setUp(self):
+ super(TestServerGroup, self).setUp()
+
+ # Get a shortcut to the ServerGroupsManager Mock
+ self.server_groups_mock = self.app.client_manager.compute.server_groups
+ self.server_groups_mock.reset_mock()
+
+
+class TestServerGroupCreate(TestServerGroup):
+
+ def setUp(self):
+ super(TestServerGroupCreate, self).setUp()
+
+ self.server_groups_mock.create.return_value = self.fake_server_group
+ self.cmd = server_group.CreateServerGroup(self.app, None)
+
+ def test_server_group_create(self):
+ arglist = [
+ '--policy', 'affinity',
+ 'affinity_group',
+ ]
+ verifylist = [
+ ('policy', ['affinity']),
+ ('name', 'affinity_group'),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ columns, data = self.cmd.take_action(parsed_args)
+ self.server_groups_mock.create.assert_called_once_with(
+ name=parsed_args.name,
+ policies=parsed_args.policy,
+ )
+
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(self.data, data)
+
+ def test_server_group_create_with_multiple_policies(self):
+ arglist = [
+ '--policy', 'affinity',
+ '--policy', 'soft-affinity',
+ 'affinity_group',
+ ]
+ verifylist = [
+ ('policy', ['affinity', 'soft-affinity']),
+ ('name', 'affinity_group'),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ columns, data = self.cmd.take_action(parsed_args)
+ self.server_groups_mock.create.assert_called_once_with(
+ name=parsed_args.name,
+ policies=parsed_args.policy,
+ )
+
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(self.data, data)
+
+ def test_server_group_create_no_policy(self):
+ arglist = [
+ 'affinity_group',
+ ]
+ verifylist = None
+ self.assertRaises(tests_utils.ParserException,
+ self.check_parser,
+ self.cmd,
+ arglist,
+ verifylist)
+
+
+class TestServerGroupDelete(TestServerGroup):
+
+ def setUp(self):
+ super(TestServerGroupDelete, self).setUp()
+
+ self.server_groups_mock.get.return_value = self.fake_server_group
+ self.cmd = server_group.DeleteServerGroup(self.app, None)
+
+ def test_server_group_delete(self):
+ arglist = [
+ 'affinity_group',
+ ]
+ verifylist = [
+ ('server_group', ['affinity_group']),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ result = self.cmd.take_action(parsed_args)
+ self.server_groups_mock.get.assert_called_once_with('affinity_group')
+ self.server_groups_mock.delete.assert_called_once_with(
+ self.fake_server_group.id
+ )
+ self.assertIsNone(result)
+
+ def test_server_group_multiple_delete(self):
+ arglist = [
+ 'affinity_group',
+ 'anti_affinity_group'
+ ]
+ verifylist = [
+ ('server_group', ['affinity_group', 'anti_affinity_group']),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ result = self.cmd.take_action(parsed_args)
+ self.server_groups_mock.get.assert_any_call('affinity_group')
+ self.server_groups_mock.get.assert_any_call('anti_affinity_group')
+ self.server_groups_mock.delete.assert_called_with(
+ self.fake_server_group.id
+ )
+ self.assertEqual(2, self.server_groups_mock.get.call_count)
+ self.assertEqual(2, self.server_groups_mock.delete.call_count)
+ self.assertIsNone(result)
+
+ def test_server_group_delete_no_input(self):
+ arglist = []
+ verifylist = None
+ self.assertRaises(tests_utils.ParserException,
+ self.check_parser,
+ self.cmd,
+ arglist,
+ verifylist)
+
+ def test_server_group_multiple_delete_with_exception(self):
+ arglist = [
+ 'affinity_group',
+ 'anti_affinity_group'
+ ]
+ verifylist = [
+ ('server_group', ['affinity_group', 'anti_affinity_group']),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ find_mock_result = [self.fake_server_group, exceptions.CommandError]
+ with mock.patch.object(utils, 'find_resource',
+ side_effect=find_mock_result) as find_mock:
+ try:
+ self.cmd.take_action(parsed_args)
+ self.fail('CommandError should be raised.')
+ except exceptions.CommandError as e:
+ self.assertEqual('1 of 2 server groups failed to delete.',
+ str(e))
+
+ find_mock.assert_any_call(self.server_groups_mock,
+ 'affinity_group')
+ find_mock.assert_any_call(self.server_groups_mock,
+ 'anti_affinity_group')
+
+ self.assertEqual(2, find_mock.call_count)
+ self.server_groups_mock.delete.assert_called_once_with(
+ self.fake_server_group.id
+ )
+
+
+class TestServerGroupList(TestServerGroup):
+
+ list_columns = (
+ 'ID',
+ 'Name',
+ 'Policies',
+ )
+
+ list_columns_long = (
+ 'ID',
+ 'Name',
+ 'Policies',
+ 'Members',
+ 'Project Id',
+ 'User Id',
+ )
+
+ list_data = ((
+ TestServerGroup.fake_server_group.id,
+ TestServerGroup.fake_server_group.name,
+ utils.format_list(TestServerGroup.fake_server_group.policies),
+ ),)
+
+ list_data_long = ((
+ TestServerGroup.fake_server_group.id,
+ TestServerGroup.fake_server_group.name,
+ utils.format_list(TestServerGroup.fake_server_group.policies),
+ utils.format_list(TestServerGroup.fake_server_group.members),
+ TestServerGroup.fake_server_group.project_id,
+ TestServerGroup.fake_server_group.user_id,
+ ),)
+
+ def setUp(self):
+ super(TestServerGroupList, self).setUp()
+
+ self.server_groups_mock.list.return_value = [self.fake_server_group]
+ self.cmd = server_group.ListServerGroup(self.app, None)
+
+ def test_server_group_list(self):
+ arglist = []
+ verifylist = [
+ ('all_projects', False),
+ ('long', False),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ columns, data = self.cmd.take_action(parsed_args)
+ self.server_groups_mock.list.assert_called_once_with(False)
+
+ self.assertEqual(self.list_columns, columns)
+ self.assertEqual(self.list_data, tuple(data))
+
+ def test_server_group_list_with_all_projects_and_long(self):
+ arglist = [
+ '--all-projects',
+ '--long',
+ ]
+ verifylist = [
+ ('all_projects', True),
+ ('long', True),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ columns, data = self.cmd.take_action(parsed_args)
+ self.server_groups_mock.list.assert_called_once_with(True)
+
+ self.assertEqual(self.list_columns_long, columns)
+ self.assertEqual(self.list_data_long, tuple(data))
+
+
+class TestServerGroupShow(TestServerGroup):
+
+ def setUp(self):
+ super(TestServerGroupShow, self).setUp()
+
+ self.server_groups_mock.get.return_value = self.fake_server_group
+ self.cmd = server_group.ShowServerGroup(self.app, None)
+
+ def test_server_group_show(self):
+ arglist = [
+ 'affinity_group',
+ ]
+ verifylist = [
+ ('server_group', 'affinity_group'),
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ columns, data = self.cmd.take_action(parsed_args)
+
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(self.data, data)
diff --git a/openstackclient/tests/unit/compute/v2/test_server_image.py b/openstackclient/tests/unit/compute/v2/test_server_image.py
new file mode 100644
index 00000000..f53f08e6
--- /dev/null
+++ b/openstackclient/tests/unit/compute/v2/test_server_image.py
@@ -0,0 +1,228 @@
+# 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 osc_lib import exceptions
+from osc_lib import utils as common_utils
+
+from openstackclient.compute.v2 import server_image
+from openstackclient.tests.unit.compute.v2 import fakes as compute_fakes
+from openstackclient.tests.unit.image.v2 import fakes as image_fakes
+
+
+class TestServerImage(compute_fakes.TestComputev2):
+
+ def setUp(self):
+ super(TestServerImage, self).setUp()
+
+ # Get a shortcut to the compute client ServerManager Mock
+ self.servers_mock = self.app.client_manager.compute.servers
+ self.servers_mock.reset_mock()
+
+ # Get a shortcut to the image client ImageManager Mock
+ self.images_mock = self.app.client_manager.image.images
+ self.images_mock.reset_mock()
+
+ # Set object attributes to be tested. Could be overwritten in subclass.
+ self.attrs = {}
+
+ # Set object methods to be tested. Could be overwritten in subclass.
+ self.methods = {}
+
+ def setup_servers_mock(self, count):
+ servers = compute_fakes.FakeServer.create_servers(
+ attrs=self.attrs,
+ methods=self.methods,
+ count=count,
+ )
+
+ # This is the return value for utils.find_resource()
+ self.servers_mock.get = compute_fakes.FakeServer.get_servers(
+ servers,
+ 0,
+ )
+ return servers
+
+
+class TestServerImageCreate(TestServerImage):
+
+ def image_columns(self, image):
+ columnlist = tuple(sorted(image.keys()))
+ return columnlist
+
+ def image_data(self, image):
+ datalist = (
+ image['id'],
+ image['name'],
+ image['owner'],
+ image['protected'],
+ 'active',
+ common_utils.format_list(image.get('tags')),
+ image['visibility'],
+ )
+ return datalist
+
+ def setUp(self):
+ super(TestServerImageCreate, self).setUp()
+
+ # Get the command object to test
+ self.cmd = server_image.CreateServerImage(self.app, None)
+
+ self.methods = {
+ 'create_image': None,
+ }
+
+ def setup_images_mock(self, count, servers=None):
+ if servers:
+ images = image_fakes.FakeImage.create_images(
+ attrs={
+ 'name': servers[0].name,
+ 'status': 'active',
+ },
+ count=count,
+ )
+ else:
+ images = image_fakes.FakeImage.create_images(
+ attrs={
+ 'status': 'active',
+ },
+ count=count,
+ )
+
+ self.images_mock.get = mock.MagicMock(side_effect=images)
+ self.servers_mock.create_image = mock.MagicMock(
+ return_value=images[0].id,
+ )
+ return images
+
+ def test_server_image_create_defaults(self):
+ servers = self.setup_servers_mock(count=1)
+ images = self.setup_images_mock(count=1, servers=servers)
+
+ arglist = [
+ servers[0].id,
+ ]
+ verifylist = [
+ ('server', servers[0].id),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ # In base command class ShowOne in cliff, abstract method take_action()
+ # returns a two-part tuple with a tuple of column names and a tuple of
+ # data to be shown.
+ columns, data = self.cmd.take_action(parsed_args)
+
+ # ServerManager.create_image(server, image_name, metadata=)
+ self.servers_mock.create_image.assert_called_with(
+ servers[0].id,
+ servers[0].name,
+ )
+
+ self.assertEqual(self.image_columns(images[0]), columns)
+ self.assertEqual(self.image_data(images[0]), data)
+
+ def test_server_image_create_options(self):
+ servers = self.setup_servers_mock(count=1)
+ images = self.setup_images_mock(count=1, servers=servers)
+
+ arglist = [
+ '--name', 'img-nam',
+ servers[0].id,
+ ]
+ verifylist = [
+ ('name', 'img-nam'),
+ ('server', servers[0].id),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ # In base command class ShowOne in cliff, abstract method take_action()
+ # returns a two-part tuple with a tuple of column names and a tuple of
+ # data to be shown.
+ columns, data = self.cmd.take_action(parsed_args)
+
+ # ServerManager.create_image(server, image_name, metadata=)
+ self.servers_mock.create_image.assert_called_with(
+ servers[0].id,
+ 'img-nam',
+ )
+
+ self.assertEqual(self.image_columns(images[0]), columns)
+ self.assertEqual(self.image_data(images[0]), data)
+
+ @mock.patch.object(common_utils, 'wait_for_status', return_value=False)
+ def test_server_create_image_wait_fail(self, mock_wait_for_status):
+ servers = self.setup_servers_mock(count=1)
+ images = self.setup_images_mock(count=1, servers=servers)
+
+ arglist = [
+ '--wait',
+ servers[0].id,
+ ]
+ verifylist = [
+ ('wait', True),
+ ('server', servers[0].id),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ self.assertRaises(
+ exceptions.CommandError,
+ self.cmd.take_action,
+ parsed_args,
+ )
+
+ # ServerManager.create_image(server, image_name, metadata=)
+ self.servers_mock.create_image.assert_called_with(
+ servers[0].id,
+ servers[0].name,
+ )
+
+ mock_wait_for_status.assert_called_once_with(
+ self.images_mock.get,
+ images[0].id,
+ callback=mock.ANY
+ )
+
+ @mock.patch.object(common_utils, 'wait_for_status', return_value=True)
+ def test_server_create_image_wait_ok(self, mock_wait_for_status):
+ servers = self.setup_servers_mock(count=1)
+ images = self.setup_images_mock(count=1, servers=servers)
+
+ arglist = [
+ '--wait',
+ servers[0].id,
+ ]
+ verifylist = [
+ ('wait', True),
+ ('server', servers[0].id),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ # In base command class ShowOne in cliff, abstract method take_action()
+ # returns a two-part tuple with a tuple of column names and a tuple of
+ # data to be shown.
+ columns, data = self.cmd.take_action(parsed_args)
+
+ # ServerManager.create_image(server, image_name, metadata=)
+ self.servers_mock.create_image.assert_called_with(
+ servers[0].id,
+ servers[0].name,
+ )
+
+ mock_wait_for_status.assert_called_once_with(
+ self.images_mock.get,
+ images[0].id,
+ callback=mock.ANY
+ )
+
+ self.assertEqual(self.image_columns(images[0]), columns)
+ self.assertEqual(self.image_data(images[0]), data)
diff --git a/openstackclient/tests/unit/compute/v2/test_service.py b/openstackclient/tests/unit/compute/v2/test_service.py
new file mode 100644
index 00000000..1fd3b7d5
--- /dev/null
+++ b/openstackclient/tests/unit/compute/v2/test_service.py
@@ -0,0 +1,410 @@
+# Copyright 2015 Mirantis, Inc.
+#
+# 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 mock import call
+
+from osc_lib import exceptions
+
+from openstackclient.compute.v2 import service
+from openstackclient.tests.unit.compute.v2 import fakes as compute_fakes
+
+
+class TestService(compute_fakes.TestComputev2):
+
+ def setUp(self):
+ super(TestService, self).setUp()
+
+ # Get a shortcut to the ServiceManager Mock
+ self.service_mock = self.app.client_manager.compute.services
+ self.service_mock.reset_mock()
+
+
+class TestServiceDelete(TestService):
+
+ services = compute_fakes.FakeService.create_services(count=2)
+
+ def setUp(self):
+ super(TestServiceDelete, self).setUp()
+
+ self.service_mock.delete.return_value = None
+
+ # Get the command object to test
+ self.cmd = service.DeleteService(self.app, None)
+
+ def test_service_delete(self):
+ arglist = [
+ self.services[0].binary,
+ ]
+ verifylist = [
+ ('service', [self.services[0].binary]),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ result = self.cmd.take_action(parsed_args)
+
+ self.service_mock.delete.assert_called_with(
+ self.services[0].binary,
+ )
+ self.assertIsNone(result)
+
+ def test_multi_services_delete(self):
+ arglist = []
+ for s in self.services:
+ arglist.append(s.binary)
+ verifylist = [
+ ('service', arglist),
+ ]
+
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ result = self.cmd.take_action(parsed_args)
+
+ calls = []
+ for s in self.services:
+ calls.append(call(s.binary))
+ self.service_mock.delete.assert_has_calls(calls)
+ self.assertIsNone(result)
+
+ def test_multi_services_delete_with_exception(self):
+ arglist = [
+ self.services[0].binary,
+ 'unexist_service',
+ ]
+ verifylist = [
+ ('service', arglist)
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ delete_mock_result = [None, exceptions.CommandError]
+ self.service_mock.delete = (
+ mock.MagicMock(side_effect=delete_mock_result)
+ )
+
+ try:
+ self.cmd.take_action(parsed_args)
+ self.fail('CommandError should be raised.')
+ except exceptions.CommandError as e:
+ self.assertEqual(
+ '1 of 2 compute services failed to delete.', str(e))
+
+ self.service_mock.delete.assert_any_call(self.services[0].binary)
+ self.service_mock.delete.assert_any_call('unexist_service')
+
+
+class TestServiceList(TestService):
+
+ service = compute_fakes.FakeService.create_one_service()
+
+ columns = (
+ 'ID',
+ 'Binary',
+ 'Host',
+ 'Zone',
+ 'Status',
+ 'State',
+ 'Updated At',
+ )
+ columns_long = columns + (
+ 'Disabled Reason',
+ )
+
+ data = [(
+ service.id,
+ service.binary,
+ service.host,
+ service.zone,
+ service.status,
+ service.state,
+ service.updated_at,
+ )]
+ data_long = [data[0] + (service.disabled_reason, )]
+
+ def setUp(self):
+ super(TestServiceList, self).setUp()
+
+ self.service_mock.list.return_value = [self.service]
+
+ # Get the command object to test
+ self.cmd = service.ListService(self.app, None)
+
+ def test_service_list(self):
+ arglist = [
+ '--host', self.service.host,
+ '--service', self.service.binary,
+ ]
+ verifylist = [
+ ('host', self.service.host),
+ ('service', self.service.binary),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ # In base command class Lister in cliff, abstract method take_action()
+ # returns a tuple containing the column names and an iterable
+ # containing the data to be listed.
+ columns, data = self.cmd.take_action(parsed_args)
+
+ self.service_mock.list.assert_called_with(
+ self.service.host,
+ self.service.binary,
+ )
+
+ self.assertEqual(self.columns, columns)
+ self.assertEqual(self.data, list(data))
+
+ def test_service_list_with_long_option(self):
+ arglist = [
+ '--host', self.service.host,
+ '--service', self.service.binary,
+ '--long'
+ ]
+ verifylist = [
+ ('host', self.service.host),
+ ('service', self.service.binary),
+ ('long', True)
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ # In base command class Lister in cliff, abstract method take_action()
+ # returns a tuple containing the column names and an iterable
+ # containing the data to be listed.
+ columns, data = self.cmd.take_action(parsed_args)
+
+ self.service_mock.list.assert_called_with(
+ self.service.host,
+ self.service.binary,
+ )
+
+ self.assertEqual(self.columns_long, columns)
+ self.assertEqual(self.data_long, list(data))
+
+
+class TestServiceSet(TestService):
+
+ def setUp(self):
+ super(TestServiceSet, self).setUp()
+
+ self.service = compute_fakes.FakeService.create_one_service()
+
+ self.service_mock.enable.return_value = self.service
+ self.service_mock.disable.return_value = self.service
+
+ self.cmd = service.SetService(self.app, None)
+
+ def test_set_nothing(self):
+ arglist = [
+ self.service.host,
+ self.service.binary,
+ ]
+ verifylist = [
+ ('host', self.service.host),
+ ('service', self.service.binary),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ result = self.cmd.take_action(parsed_args)
+
+ self.service_mock.enable.assert_not_called()
+ self.service_mock.disable.assert_not_called()
+ self.service_mock.disable_log_reason.assert_not_called()
+ self.assertIsNone(result)
+
+ def test_service_set_enable(self):
+ arglist = [
+ '--enable',
+ self.service.host,
+ self.service.binary,
+ ]
+ verifylist = [
+ ('enable', True),
+ ('host', self.service.host),
+ ('service', self.service.binary),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ result = self.cmd.take_action(parsed_args)
+
+ self.service_mock.enable.assert_called_with(
+ self.service.host,
+ self.service.binary
+ )
+ self.assertIsNone(result)
+
+ def test_service_set_disable(self):
+ arglist = [
+ '--disable',
+ self.service.host,
+ self.service.binary,
+ ]
+ verifylist = [
+ ('disable', True),
+ ('host', self.service.host),
+ ('service', self.service.binary),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ result = self.cmd.take_action(parsed_args)
+
+ self.service_mock.disable.assert_called_with(
+ self.service.host,
+ self.service.binary
+ )
+ self.assertIsNone(result)
+
+ def test_service_set_disable_with_reason(self):
+ reason = 'earthquake'
+ arglist = [
+ '--disable',
+ '--disable-reason', reason,
+ self.service.host,
+ self.service.binary,
+ ]
+ verifylist = [
+ ('disable', True),
+ ('disable_reason', reason),
+ ('host', self.service.host),
+ ('service', self.service.binary),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ result = self.cmd.take_action(parsed_args)
+
+ self.service_mock.disable_log_reason.assert_called_with(
+ self.service.host,
+ self.service.binary,
+ reason
+ )
+ self.assertIsNone(result)
+
+ def test_service_set_only_with_disable_reason(self):
+ reason = 'earthquake'
+ arglist = [
+ '--disable-reason', reason,
+ self.service.host,
+ self.service.binary,
+ ]
+ verifylist = [
+ ('disable_reason', reason),
+ ('host', self.service.host),
+ ('service', self.service.binary),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ try:
+ self.cmd.take_action(parsed_args)
+ self.fail("CommandError should be raised.")
+ except exceptions.CommandError as e:
+ self.assertEqual("Cannot specify option --disable-reason without "
+ "--disable specified.", str(e))
+
+ def test_service_set_enable_with_disable_reason(self):
+ reason = 'earthquake'
+ arglist = [
+ '--enable',
+ '--disable-reason', reason,
+ self.service.host,
+ self.service.binary,
+ ]
+ verifylist = [
+ ('enable', True),
+ ('disable_reason', reason),
+ ('host', self.service.host),
+ ('service', self.service.binary),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ try:
+ self.cmd.take_action(parsed_args)
+ self.fail("CommandError should be raised.")
+ except exceptions.CommandError as e:
+ self.assertEqual("Cannot specify option --disable-reason without "
+ "--disable specified.", str(e))
+
+ def test_service_set_state_up(self):
+ arglist = [
+ '--up',
+ self.service.host,
+ self.service.binary,
+ ]
+ verifylist = [
+ ('up', True),
+ ('host', self.service.host),
+ ('service', self.service.binary),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ result = self.cmd.take_action(parsed_args)
+ self.service_mock.force_down.assert_called_once_with(
+ self.service.host, self.service.binary, force_down=False)
+ self.assertNotCalled(self.service_mock.enable)
+ self.assertNotCalled(self.service_mock.disable)
+ self.assertIsNone(result)
+
+ def test_service_set_state_down(self):
+ arglist = [
+ '--down',
+ self.service.host,
+ self.service.binary,
+ ]
+ verifylist = [
+ ('down', True),
+ ('host', self.service.host),
+ ('service', self.service.binary),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ result = self.cmd.take_action(parsed_args)
+ self.service_mock.force_down.assert_called_once_with(
+ self.service.host, self.service.binary, force_down=True)
+ self.assertNotCalled(self.service_mock.enable)
+ self.assertNotCalled(self.service_mock.disable)
+ self.assertIsNone(result)
+
+ def test_service_set_enable_and_state_down(self):
+ arglist = [
+ '--enable',
+ '--down',
+ self.service.host,
+ self.service.binary,
+ ]
+ verifylist = [
+ ('enable', True),
+ ('down', True),
+ ('host', self.service.host),
+ ('service', self.service.binary),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+ result = self.cmd.take_action(parsed_args)
+ self.service_mock.enable.assert_called_once_with(
+ self.service.host, self.service.binary)
+ self.service_mock.force_down.assert_called_once_with(
+ self.service.host, self.service.binary, force_down=True)
+ self.assertIsNone(result)
+
+ def test_service_set_enable_and_state_down_with_exception(self):
+ arglist = [
+ '--enable',
+ '--down',
+ self.service.host,
+ self.service.binary,
+ ]
+ verifylist = [
+ ('enable', True),
+ ('down', True),
+ ('host', self.service.host),
+ ('service', self.service.binary),
+ ]
+ parsed_args = self.check_parser(self.cmd, arglist, verifylist)
+
+ with mock.patch.object(self.service_mock, 'enable',
+ side_effect=Exception()):
+ self.assertRaises(exceptions.CommandError,
+ self.cmd.take_action, parsed_args)
+ self.service_mock.force_down.assert_called_once_with(
+ self.service.host, self.service.binary, force_down=True)