diff options
Diffstat (limited to 'openstackclient/tests/unit/compute')
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) |