diff options
Diffstat (limited to 'openstackclient/tests/compute')
| -rw-r--r-- | openstackclient/tests/compute/__init__.py | 0 | ||||
| -rw-r--r-- | openstackclient/tests/compute/v2/__init__.py | 0 | ||||
| -rw-r--r-- | openstackclient/tests/compute/v2/fakes.py | 1249 | ||||
| -rw-r--r-- | openstackclient/tests/compute/v2/test_agent.py | 320 | ||||
| -rw-r--r-- | openstackclient/tests/compute/v2/test_aggregate.py | 447 | ||||
| -rw-r--r-- | openstackclient/tests/compute/v2/test_console.py | 196 | ||||
| -rw-r--r-- | openstackclient/tests/compute/v2/test_flavor.py | 825 | ||||
| -rw-r--r-- | openstackclient/tests/compute/v2/test_host.py | 179 | ||||
| -rw-r--r-- | openstackclient/tests/compute/v2/test_hypervisor.py | 221 | ||||
| -rw-r--r-- | openstackclient/tests/compute/v2/test_hypervisor_stats.py | 79 | ||||
| -rw-r--r-- | openstackclient/tests/compute/v2/test_keypair.py | 312 | ||||
| -rw-r--r-- | openstackclient/tests/compute/v2/test_server.py | 1844 | ||||
| -rw-r--r-- | openstackclient/tests/compute/v2/test_server_backup.py | 271 | ||||
| -rw-r--r-- | openstackclient/tests/compute/v2/test_server_group.py | 284 | ||||
| -rw-r--r-- | openstackclient/tests/compute/v2/test_server_image.py | 228 | ||||
| -rw-r--r-- | openstackclient/tests/compute/v2/test_service.py | 410 |
16 files changed, 0 insertions, 6865 deletions
diff --git a/openstackclient/tests/compute/__init__.py b/openstackclient/tests/compute/__init__.py deleted file mode 100644 index e69de29b..00000000 --- a/openstackclient/tests/compute/__init__.py +++ /dev/null diff --git a/openstackclient/tests/compute/v2/__init__.py b/openstackclient/tests/compute/v2/__init__.py deleted file mode 100644 index e69de29b..00000000 --- a/openstackclient/tests/compute/v2/__init__.py +++ /dev/null diff --git a/openstackclient/tests/compute/v2/fakes.py b/openstackclient/tests/compute/v2/fakes.py deleted file mode 100644 index 85c11c94..00000000 --- a/openstackclient/tests/compute/v2/fakes.py +++ /dev/null @@ -1,1249 +0,0 @@ -# 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 import fakes -from openstackclient.tests.identity.v2_0 import fakes as identity_fakes -from openstackclient.tests.image.v2 import fakes as image_fakes -from openstackclient.tests.network.v2 import fakes as network_fakes -from openstackclient.tests import utils -from openstackclient.tests.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/compute/v2/test_agent.py b/openstackclient/tests/compute/v2/test_agent.py deleted file mode 100644 index 07265bb0..00000000 --- a/openstackclient/tests/compute/v2/test_agent.py +++ /dev/null @@ -1,320 +0,0 @@ -# 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.compute.v2 import fakes as compute_fakes -from openstackclient.tests 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/compute/v2/test_aggregate.py b/openstackclient/tests/compute/v2/test_aggregate.py deleted file mode 100644 index 3ebca35f..00000000 --- a/openstackclient/tests/compute/v2/test_aggregate.py +++ /dev/null @@ -1,447 +0,0 @@ -# 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.compute.v2 import fakes as compute_fakes -from openstackclient.tests 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/compute/v2/test_console.py b/openstackclient/tests/compute/v2/test_console.py deleted file mode 100644 index fe4f26ae..00000000 --- a/openstackclient/tests/compute/v2/test_console.py +++ /dev/null @@ -1,196 +0,0 @@ -# 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.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/compute/v2/test_flavor.py b/openstackclient/tests/compute/v2/test_flavor.py deleted file mode 100644 index d326ea8a..00000000 --- a/openstackclient/tests/compute/v2/test_flavor.py +++ /dev/null @@ -1,825 +0,0 @@ -# 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.compute.v2 import fakes as compute_fakes -from openstackclient.tests.identity.v3 import fakes as identity_fakes -from openstackclient.tests 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/compute/v2/test_host.py b/openstackclient/tests/compute/v2/test_host.py deleted file mode 100644 index 9ebed68d..00000000 --- a/openstackclient/tests/compute/v2/test_host.py +++ /dev/null @@ -1,179 +0,0 @@ -# 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.compute.v2 import fakes as compute_fakes -from openstackclient.tests 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/compute/v2/test_hypervisor.py b/openstackclient/tests/compute/v2/test_hypervisor.py deleted file mode 100644 index ee0f40ed..00000000 --- a/openstackclient/tests/compute/v2/test_hypervisor.py +++ /dev/null @@ -1,221 +0,0 @@ -# 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.compute.v2 import fakes as compute_fakes -from openstackclient.tests 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/compute/v2/test_hypervisor_stats.py b/openstackclient/tests/compute/v2/test_hypervisor_stats.py deleted file mode 100644 index ca5ce29b..00000000 --- a/openstackclient/tests/compute/v2/test_hypervisor_stats.py +++ /dev/null @@ -1,79 +0,0 @@ -# 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.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/compute/v2/test_keypair.py b/openstackclient/tests/compute/v2/test_keypair.py deleted file mode 100644 index 25949e31..00000000 --- a/openstackclient/tests/compute/v2/test_keypair.py +++ /dev/null @@ -1,312 +0,0 @@ -# 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.compute.v2 import fakes as compute_fakes -from openstackclient.tests 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/compute/v2/test_server.py b/openstackclient/tests/compute/v2/test_server.py deleted file mode 100644 index a98398ee..00000000 --- a/openstackclient/tests/compute/v2/test_server.py +++ /dev/null @@ -1,1844 +0,0 @@ -# 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.compute.v2 import fakes as compute_fakes -from openstackclient.tests.image.v2 import fakes as image_fakes -from openstackclient.tests import utils -from openstackclient.tests.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/compute/v2/test_server_backup.py b/openstackclient/tests/compute/v2/test_server_backup.py deleted file mode 100644 index 8eeb0dca..00000000 --- a/openstackclient/tests/compute/v2/test_server_backup.py +++ /dev/null @@ -1,271 +0,0 @@ -# Licensed under the Apache License, Version 2.0 (the "License"); you may -# not use this file except in compliance with the License. You may obtain -# a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT -# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the -# License for the specific language governing permissions and limitations -# under the License. -# - -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.compute.v2 import fakes as compute_fakes -from openstackclient.tests.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/compute/v2/test_server_group.py b/openstackclient/tests/compute/v2/test_server_group.py deleted file mode 100644 index bd5f8471..00000000 --- a/openstackclient/tests/compute/v2/test_server_group.py +++ /dev/null @@ -1,284 +0,0 @@ -# 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.compute.v2 import fakes as compute_fakes -from openstackclient.tests 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/compute/v2/test_server_image.py b/openstackclient/tests/compute/v2/test_server_image.py deleted file mode 100644 index c3c52da0..00000000 --- a/openstackclient/tests/compute/v2/test_server_image.py +++ /dev/null @@ -1,228 +0,0 @@ -# Licensed under the Apache License, Version 2.0 (the "License"); you may -# not use this file except in compliance with the License. You may obtain -# a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT -# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the -# License for the specific language governing permissions and limitations -# under the License. -# -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.compute.v2 import fakes as compute_fakes -from openstackclient.tests.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/compute/v2/test_service.py b/openstackclient/tests/compute/v2/test_service.py deleted file mode 100644 index 1599f466..00000000 --- a/openstackclient/tests/compute/v2/test_service.py +++ /dev/null @@ -1,410 +0,0 @@ -# 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.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) |