From d67006ea0f59ee6a7ab9bc5fd7c8d615167198c8 Mon Sep 17 00:00:00 2001 From: Stephen Finucane Date: Thu, 25 May 2023 12:00:29 +0100 Subject: [PATCH] tests: Move cloud-layer service tests to their own file This mirrors how we do unit testing elsewhere, with the test module's path mirroring the module-under-test's path. Change-Id: I0382c5bb0d86431f6f7520b16881360477eb2658 Signed-off-by: Stephen Finucane --- openstack/tests/unit/cloud/test_cloud.py | 615 --------------------- openstack/tests/unit/cloud/test_compute.py | 500 +++++++++++++++++ openstack/tests/unit/cloud/test_image.py | 16 + openstack/tests/unit/cloud/test_network.py | 124 ++++- 4 files changed, 639 insertions(+), 616 deletions(-) create mode 100644 openstack/tests/unit/cloud/test_compute.py diff --git a/openstack/tests/unit/cloud/test_cloud.py b/openstack/tests/unit/cloud/test_cloud.py index 147c8656d..cbed0c86a 100644 --- a/openstack/tests/unit/cloud/test_cloud.py +++ b/openstack/tests/unit/cloud/test_cloud.py @@ -17,8 +17,6 @@ import testtools from openstack.cloud import exc from openstack import connection -from openstack import exceptions -from openstack.tests import fakes from openstack.tests.unit import base from openstack import utils @@ -34,23 +32,6 @@ RANGE_DATA = [ class TestCloud(base.TestCase): - def setUp(self): - # This set of tests are not testing neutron, they're testing - # rebuilding servers, but we do several network calls in service - # of a NORMAL rebuild to find the default_network. Putting - # in all of the neutron mocks for that will make the tests harder - # to read. SO - we're going mock neutron into the off position - # and then turn it back on in the few tests that specifically do. - # Maybe we should reorg these into two classes - one with neutron - # mocked out - and one with it not mocked out - super().setUp() - self.has_neutron = False - - def fake_has_service(*args, **kwargs): - return self.has_neutron - - self.cloud.has_service = fake_has_service - def test_openstack_cloud(self): self.assertIsInstance(self.cloud, connection.Connection) @@ -116,20 +97,6 @@ class TestCloud(base.TestCase): self.assertEqual(c2.list_servers(), []) self.assert_calls() - @mock.patch.object(connection.Connection, 'search_images') - def test_get_images(self, mock_search): - image1 = dict(id='123', name='mickey') - mock_search.return_value = [image1] - r = self.cloud.get_image('mickey') - self.assertIsNotNone(r) - self.assertDictEqual(image1, r) - - @mock.patch.object(connection.Connection, 'search_images') - def test_get_image_not_found(self, mock_search): - mock_search.return_value = [] - r = self.cloud.get_image('doesNotExist') - self.assertIsNone(r) - def test_global_request_id(self): request_id = uuid.uuid4().hex self.register_uris( @@ -176,355 +143,6 @@ class TestCloud(base.TestCase): self.assert_calls() - def test_get_server(self): - server1 = fakes.make_fake_server('123', 'mickey') - server2 = fakes.make_fake_server('345', 'mouse') - - self.register_uris( - [ - self.get_nova_discovery_mock_dict(), - dict( - method='GET', - uri=self.get_mock_url( - 'compute', 'public', append=['servers', 'detail'] - ), - json={'servers': [server1, server2]}, - ), - ] - ) - - r = self.cloud.get_server('mickey') - self.assertIsNotNone(r) - self.assertEqual(server1['name'], r['name']) - - self.assert_calls() - - def test_get_server_not_found(self): - self.register_uris( - [ - self.get_nova_discovery_mock_dict(), - dict( - method='GET', - uri=self.get_mock_url( - 'compute', 'public', append=['servers', 'detail'] - ), - json={'servers': []}, - ), - ] - ) - - r = self.cloud.get_server('doesNotExist') - self.assertIsNone(r) - - self.assert_calls() - - def test_list_servers_exception(self): - self.register_uris( - [ - self.get_nova_discovery_mock_dict(), - dict( - method='GET', - uri=self.get_mock_url( - 'compute', 'public', append=['servers', 'detail'] - ), - status_code=400, - ), - ] - ) - - self.assertRaises(exc.OpenStackCloudException, self.cloud.list_servers) - - self.assert_calls() - - def test_neutron_not_found(self): - self.use_nothing() - self.cloud.has_service = mock.Mock(return_value=False) - self.assertEqual([], self.cloud.list_networks()) - self.assert_calls() - - def test_list_servers(self): - server_id = str(uuid.uuid4()) - server_name = self.getUniqueString('name') - fake_server = fakes.make_fake_server(server_id, server_name) - self.register_uris( - [ - self.get_nova_discovery_mock_dict(), - dict( - method='GET', - uri=self.get_mock_url( - 'compute', 'public', append=['servers', 'detail'] - ), - json={'servers': [fake_server]}, - ), - ] - ) - - r = self.cloud.list_servers() - - self.assertEqual(1, len(r)) - self.assertEqual(server_name, r[0]['name']) - - self.assert_calls() - - def test_list_server_private_ip(self): - self.has_neutron = True - fake_server = { - "OS-EXT-STS:task_state": None, - "addresses": { - "private": [ - { - "OS-EXT-IPS-MAC:mac_addr": "fa:16:3e:b4:a3:07", - "version": 4, - "addr": "10.4.0.13", - "OS-EXT-IPS:type": "fixed", - }, - { - "OS-EXT-IPS-MAC:mac_addr": "fa:16:3e:b4:a3:07", - "version": 4, - "addr": "89.40.216.229", - "OS-EXT-IPS:type": "floating", - }, - ] - }, - "links": [ - {"href": "http://example.com/images/95e4c4", "rel": "self"}, - { - "href": "http://example.com/images/95e4c4", - "rel": "bookmark", - }, - ], - "image": { - "id": "95e4c449-8abf-486e-97d9-dc3f82417d2d", - "links": [ - { - "href": "http://example.com/images/95e4c4", - "rel": "bookmark", - } - ], - }, - "OS-EXT-STS:vm_state": "active", - "OS-SRV-USG:launched_at": "2018-03-01T02:44:50.000000", - "flavor": { - "id": "3bd99062-2fe8-4eac-93f0-9200cc0f97ae", - "links": [ - { - "href": "http://example.com/flavors/95e4c4", - "rel": "bookmark", - } - ], - }, - "id": "97fe35e9-756a-41a2-960a-1d057d2c9ee4", - "security_groups": [{"name": "default"}], - "user_id": "c17534835f8f42bf98fc367e0bf35e09", - "OS-DCF:diskConfig": "MANUAL", - "accessIPv4": "", - "accessIPv6": "", - "progress": 0, - "OS-EXT-STS:power_state": 1, - "OS-EXT-AZ:availability_zone": "nova", - "metadata": {}, - "status": "ACTIVE", - "updated": "2018-03-01T02:44:51Z", - "hostId": "", - "OS-SRV-USG:terminated_at": None, - "key_name": None, - "name": "mttest", - "created": "2018-03-01T02:44:46Z", - "tenant_id": "65222a4d09ea4c68934fa1028c77f394", - "os-extended-volumes:volumes_attached": [], - "config_drive": "", - } - fake_networks = { - "networks": [ - { - "status": "ACTIVE", - "router:external": True, - "availability_zone_hints": [], - "availability_zones": ["nova"], - "description": None, - "subnets": [ - "df3e17fa-a4b2-47ae-9015-bc93eb076ba2", - "6b0c3dc9-b0b8-4d87-976a-7f2ebf13e7ec", - "fc541f48-fc7f-48c0-a063-18de6ee7bdd7", - ], - "shared": False, - "tenant_id": "a564613210ee43708b8a7fc6274ebd63", - "tags": [], - "ipv6_address_scope": "9f03124f-89af-483a-b6fd-10f08079db4d", # noqa: E501 - "mtu": 1550, - "is_default": False, - "admin_state_up": True, - "revision_number": 0, - "ipv4_address_scope": None, - "port_security_enabled": True, - "project_id": "a564613210ee43708b8a7fc6274ebd63", - "id": "0232c17f-2096-49bc-b205-d3dcd9a30ebf", - "name": "ext-net", - }, - { - "status": "ACTIVE", - "router:external": False, - "availability_zone_hints": [], - "availability_zones": ["nova"], - "description": "", - "subnets": ["f0ad1df5-53ee-473f-b86b-3604ea5591e9"], - "shared": False, - "tenant_id": "65222a4d09ea4c68934fa1028c77f394", - "created_at": "2016-10-22T13:46:26Z", - "tags": [], - "ipv6_address_scope": None, - "updated_at": "2016-10-22T13:46:26Z", - "admin_state_up": True, - "mtu": 1500, - "revision_number": 0, - "ipv4_address_scope": None, - "port_security_enabled": True, - "project_id": "65222a4d09ea4c68934fa1028c77f394", - "id": "2c9adcb5-c123-4c5a-a2ba-1ad4c4e1481f", - "name": "private", - }, - ] - } - fake_subnets = { - "subnets": [ - { - "service_types": [], - "description": "", - "enable_dhcp": True, - "tags": [], - "network_id": "827c6bb6-492f-4168-9577-f3a131eb29e8", - "tenant_id": "65222a4d09ea4c68934fa1028c77f394", - "created_at": "2017-06-12T13:23:57Z", - "dns_nameservers": [], - "updated_at": "2017-06-12T13:23:57Z", - "gateway_ip": "10.24.4.1", - "ipv6_ra_mode": None, - "allocation_pools": [ - {"start": "10.24.4.2", "end": "10.24.4.254"} - ], - "host_routes": [], - "revision_number": 0, - "ip_version": 4, - "ipv6_address_mode": None, - "cidr": "10.24.4.0/24", - "project_id": "65222a4d09ea4c68934fa1028c77f394", - "id": "3f0642d9-4644-4dff-af25-bcf64f739698", - "subnetpool_id": None, - "name": "foo_subnet", - }, - { - "service_types": [], - "description": "", - "enable_dhcp": True, - "tags": [], - "network_id": "2c9adcb5-c123-4c5a-a2ba-1ad4c4e1481f", - "tenant_id": "65222a4d09ea4c68934fa1028c77f394", - "created_at": "2016-10-22T13:46:26Z", - "dns_nameservers": ["89.36.90.101", "89.36.90.102"], - "updated_at": "2016-10-22T13:46:26Z", - "gateway_ip": "10.4.0.1", - "ipv6_ra_mode": None, - "allocation_pools": [ - {"start": "10.4.0.2", "end": "10.4.0.200"} - ], - "host_routes": [], - "revision_number": 0, - "ip_version": 4, - "ipv6_address_mode": None, - "cidr": "10.4.0.0/24", - "project_id": "65222a4d09ea4c68934fa1028c77f394", - "id": "f0ad1df5-53ee-473f-b86b-3604ea5591e9", - "subnetpool_id": None, - "name": "private-subnet-ipv4", - }, - ] - } - self.register_uris( - [ - self.get_nova_discovery_mock_dict(), - dict( - method='GET', - uri=self.get_mock_url( - 'compute', 'public', append=['servers', 'detail'] - ), - json={'servers': [fake_server]}, - ), - dict( - method='GET', - uri=self.get_mock_url( - 'network', 'public', append=['v2.0', 'networks'] - ), - json=fake_networks, - ), - dict( - method='GET', - uri=self.get_mock_url( - 'network', 'public', append=['v2.0', 'subnets'] - ), - json=fake_subnets, - ), - ] - ) - - r = self.cloud.get_server('97fe35e9-756a-41a2-960a-1d057d2c9ee4') - - self.assertEqual('10.4.0.13', r['private_v4']) - - self.assert_calls() - - def test_list_servers_all_projects(self): - '''This test verifies that when list_servers is called with - `all_projects=True` that it passes `all_tenants=True` to nova.''' - self.register_uris( - [ - self.get_nova_discovery_mock_dict(), - dict( - method='GET', - uri=self.get_mock_url( - 'compute', - 'public', - append=['servers', 'detail'], - qs_elements=['all_tenants=True'], - ), - complete_qs=True, - json={'servers': []}, - ), - ] - ) - - self.cloud.list_servers(all_projects=True) - - self.assert_calls() - - def test_list_servers_filters(self): - '''This test verifies that when list_servers is called with - `filters` dict that it passes it to nova.''' - self.register_uris( - [ - self.get_nova_discovery_mock_dict(), - dict( - method='GET', - uri=self.get_mock_url( - 'compute', - 'public', - append=['servers', 'detail'], - qs_elements=[ - 'deleted=True', - 'changes-since=2014-12-03T00:00:00Z', - ], - ), - complete_qs=True, - json={'servers': []}, - ), - ] - ) - - self.cloud.list_servers( - filters={'deleted': True, 'changes-since': '2014-12-03T00:00:00Z'} - ) - - self.assert_calls() - def test_iterate_timeout_bad_wait(self): with testtools.ExpectedException( exc.OpenStackCloudException, @@ -561,239 +179,6 @@ class TestCloud(base.TestCase): pass mock_sleep.assert_called_with(1.0) - def test__nova_extensions(self): - body = [ - { - "updated": "2014-12-03T00:00:00Z", - "name": "Multinic", - "links": [], - "namespace": "http://openstack.org/compute/ext/fake_xml", - "alias": "NMN", - "description": "Multiple network support.", - }, - { - "updated": "2014-12-03T00:00:00Z", - "name": "DiskConfig", - "links": [], - "namespace": "http://openstack.org/compute/ext/fake_xml", - "alias": "OS-DCF", - "description": "Disk Management Extension.", - }, - ] - self.register_uris( - [ - dict( - method='GET', - uri='{endpoint}/extensions'.format( - endpoint=fakes.COMPUTE_ENDPOINT - ), - json=dict(extensions=body), - ) - ] - ) - extensions = self.cloud._nova_extensions() - self.assertEqual(set(['NMN', 'OS-DCF']), extensions) - - self.assert_calls() - - def test__nova_extensions_fails(self): - self.register_uris( - [ - dict( - method='GET', - uri='{endpoint}/extensions'.format( - endpoint=fakes.COMPUTE_ENDPOINT - ), - status_code=404, - ), - ] - ) - self.assertRaises( - exceptions.ResourceNotFound, self.cloud._nova_extensions - ) - - self.assert_calls() - - def test__has_nova_extension(self): - body = [ - { - "updated": "2014-12-03T00:00:00Z", - "name": "Multinic", - "links": [], - "namespace": "http://openstack.org/compute/ext/fake_xml", - "alias": "NMN", - "description": "Multiple network support.", - }, - { - "updated": "2014-12-03T00:00:00Z", - "name": "DiskConfig", - "links": [], - "namespace": "http://openstack.org/compute/ext/fake_xml", - "alias": "OS-DCF", - "description": "Disk Management Extension.", - }, - ] - self.register_uris( - [ - dict( - method='GET', - uri='{endpoint}/extensions'.format( - endpoint=fakes.COMPUTE_ENDPOINT - ), - json=dict(extensions=body), - ) - ] - ) - self.assertTrue(self.cloud._has_nova_extension('NMN')) - - self.assert_calls() - - def test__has_nova_extension_missing(self): - body = [ - { - "updated": "2014-12-03T00:00:00Z", - "name": "Multinic", - "links": [], - "namespace": "http://openstack.org/compute/ext/fake_xml", - "alias": "NMN", - "description": "Multiple network support.", - }, - { - "updated": "2014-12-03T00:00:00Z", - "name": "DiskConfig", - "links": [], - "namespace": "http://openstack.org/compute/ext/fake_xml", - "alias": "OS-DCF", - "description": "Disk Management Extension.", - }, - ] - self.register_uris( - [ - dict( - method='GET', - uri='{endpoint}/extensions'.format( - endpoint=fakes.COMPUTE_ENDPOINT - ), - json=dict(extensions=body), - ) - ] - ) - self.assertFalse(self.cloud._has_nova_extension('invalid')) - - self.assert_calls() - - def test__neutron_extensions(self): - body = [ - { - "updated": "2014-06-1T10:00:00-00:00", - "name": "Distributed Virtual Router", - "links": [], - "alias": "dvr", - "description": "Enables configuration of Distributed Virtual Routers.", # noqa: E501 - }, - { - "updated": "2013-07-23T10:00:00-00:00", - "name": "Allowed Address Pairs", - "links": [], - "alias": "allowed-address-pairs", - "description": "Provides allowed address pairs", - }, - ] - self.register_uris( - [ - dict( - method='GET', - uri=self.get_mock_url( - 'network', 'public', append=['v2.0', 'extensions'] - ), - json=dict(extensions=body), - ) - ] - ) - extensions = self.cloud._neutron_extensions() - self.assertEqual(set(['dvr', 'allowed-address-pairs']), extensions) - - self.assert_calls() - - def test__neutron_extensions_fails(self): - self.register_uris( - [ - dict( - method='GET', - uri=self.get_mock_url( - 'network', 'public', append=['v2.0', 'extensions'] - ), - status_code=404, - ) - ] - ) - with testtools.ExpectedException(exceptions.ResourceNotFound): - self.cloud._neutron_extensions() - - self.assert_calls() - - def test__has_neutron_extension(self): - body = [ - { - "updated": "2014-06-1T10:00:00-00:00", - "name": "Distributed Virtual Router", - "links": [], - "alias": "dvr", - "description": "Enables configuration of Distributed Virtual Routers.", # noqa: E501 - }, - { - "updated": "2013-07-23T10:00:00-00:00", - "name": "Allowed Address Pairs", - "links": [], - "alias": "allowed-address-pairs", - "description": "Provides allowed address pairs", - }, - ] - self.register_uris( - [ - dict( - method='GET', - uri=self.get_mock_url( - 'network', 'public', append=['v2.0', 'extensions'] - ), - json=dict(extensions=body), - ) - ] - ) - self.assertTrue(self.cloud._has_neutron_extension('dvr')) - self.assert_calls() - - def test__has_neutron_extension_missing(self): - body = [ - { - "updated": "2014-06-1T10:00:00-00:00", - "name": "Distributed Virtual Router", - "links": [], - "alias": "dvr", - "description": "Enables configuration of Distributed Virtual Routers.", # noqa: E501 - }, - { - "updated": "2013-07-23T10:00:00-00:00", - "name": "Allowed Address Pairs", - "links": [], - "alias": "allowed-address-pairs", - "description": "Provides allowed address pairs", - }, - ] - self.register_uris( - [ - dict( - method='GET', - uri=self.get_mock_url( - 'network', 'public', append=['v2.0', 'extensions'] - ), - json=dict(extensions=body), - ) - ] - ) - self.assertFalse(self.cloud._has_neutron_extension('invalid')) - self.assert_calls() - def test_range_search(self): filters = {"key1": "min", "key2": "20"} retval = self.cloud.range_search(RANGE_DATA, filters) diff --git a/openstack/tests/unit/cloud/test_compute.py b/openstack/tests/unit/cloud/test_compute.py new file mode 100644 index 000000000..04622ee63 --- /dev/null +++ b/openstack/tests/unit/cloud/test_compute.py @@ -0,0 +1,500 @@ +# 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 uuid + +from openstack.cloud import exc +from openstack import exceptions +from openstack.tests import fakes +from openstack.tests.unit import base + + +class TestNovaExtensions(base.TestCase): + def test__nova_extensions(self): + body = [ + { + "updated": "2014-12-03T00:00:00Z", + "name": "Multinic", + "links": [], + "namespace": "http://openstack.org/compute/ext/fake_xml", + "alias": "NMN", + "description": "Multiple network support.", + }, + { + "updated": "2014-12-03T00:00:00Z", + "name": "DiskConfig", + "links": [], + "namespace": "http://openstack.org/compute/ext/fake_xml", + "alias": "OS-DCF", + "description": "Disk Management Extension.", + }, + ] + self.register_uris( + [ + dict( + method='GET', + uri='{endpoint}/extensions'.format( + endpoint=fakes.COMPUTE_ENDPOINT + ), + json=dict(extensions=body), + ) + ] + ) + extensions = self.cloud._nova_extensions() + self.assertEqual(set(['NMN', 'OS-DCF']), extensions) + + self.assert_calls() + + def test__nova_extensions_fails(self): + self.register_uris( + [ + dict( + method='GET', + uri='{endpoint}/extensions'.format( + endpoint=fakes.COMPUTE_ENDPOINT + ), + status_code=404, + ), + ] + ) + self.assertRaises( + exceptions.ResourceNotFound, self.cloud._nova_extensions + ) + + self.assert_calls() + + def test__has_nova_extension(self): + body = [ + { + "updated": "2014-12-03T00:00:00Z", + "name": "Multinic", + "links": [], + "namespace": "http://openstack.org/compute/ext/fake_xml", + "alias": "NMN", + "description": "Multiple network support.", + }, + { + "updated": "2014-12-03T00:00:00Z", + "name": "DiskConfig", + "links": [], + "namespace": "http://openstack.org/compute/ext/fake_xml", + "alias": "OS-DCF", + "description": "Disk Management Extension.", + }, + ] + self.register_uris( + [ + dict( + method='GET', + uri='{endpoint}/extensions'.format( + endpoint=fakes.COMPUTE_ENDPOINT + ), + json=dict(extensions=body), + ) + ] + ) + self.assertTrue(self.cloud._has_nova_extension('NMN')) + + self.assert_calls() + + def test__has_nova_extension_missing(self): + body = [ + { + "updated": "2014-12-03T00:00:00Z", + "name": "Multinic", + "links": [], + "namespace": "http://openstack.org/compute/ext/fake_xml", + "alias": "NMN", + "description": "Multiple network support.", + }, + { + "updated": "2014-12-03T00:00:00Z", + "name": "DiskConfig", + "links": [], + "namespace": "http://openstack.org/compute/ext/fake_xml", + "alias": "OS-DCF", + "description": "Disk Management Extension.", + }, + ] + self.register_uris( + [ + dict( + method='GET', + uri='{endpoint}/extensions'.format( + endpoint=fakes.COMPUTE_ENDPOINT + ), + json=dict(extensions=body), + ) + ] + ) + self.assertFalse(self.cloud._has_nova_extension('invalid')) + + self.assert_calls() + + +class TestServers(base.TestCase): + def test_get_server(self): + server1 = fakes.make_fake_server('123', 'mickey') + server2 = fakes.make_fake_server('345', 'mouse') + + self.register_uris( + [ + self.get_nova_discovery_mock_dict(), + dict( + method='GET', + uri=self.get_mock_url( + 'compute', 'public', append=['servers', 'detail'] + ), + json={'servers': [server1, server2]}, + ), + dict( + method='GET', + uri=self.get_mock_url( + 'network', 'public', append=['v2.0', 'networks'] + ), + json={"networks": []}, + ), + ] + ) + + r = self.cloud.get_server('mickey') + self.assertIsNotNone(r) + self.assertEqual(server1['name'], r['name']) + + self.assert_calls() + + def test_get_server_not_found(self): + self.register_uris( + [ + self.get_nova_discovery_mock_dict(), + dict( + method='GET', + uri=self.get_mock_url( + 'compute', 'public', append=['servers', 'detail'] + ), + json={'servers': []}, + ), + ] + ) + + r = self.cloud.get_server('doesNotExist') + self.assertIsNone(r) + + self.assert_calls() + + def test_list_servers(self): + server_id = str(uuid.uuid4()) + server_name = self.getUniqueString('name') + fake_server = fakes.make_fake_server(server_id, server_name) + self.register_uris( + [ + self.get_nova_discovery_mock_dict(), + dict( + method='GET', + uri=self.get_mock_url( + 'compute', 'public', append=['servers', 'detail'] + ), + json={'servers': [fake_server]}, + ), + dict( + method='GET', + uri=self.get_mock_url( + 'network', 'public', append=['v2.0', 'networks'] + ), + json={"networks": []}, + ), + ] + ) + + r = self.cloud.list_servers() + + self.assertEqual(1, len(r)) + self.assertEqual(server_name, r[0]['name']) + + self.assert_calls() + + def test_list_server_private_ip(self): + self.has_neutron = True + fake_server = { + "OS-EXT-STS:task_state": None, + "addresses": { + "private": [ + { + "OS-EXT-IPS-MAC:mac_addr": "fa:16:3e:b4:a3:07", + "version": 4, + "addr": "10.4.0.13", + "OS-EXT-IPS:type": "fixed", + }, + { + "OS-EXT-IPS-MAC:mac_addr": "fa:16:3e:b4:a3:07", + "version": 4, + "addr": "89.40.216.229", + "OS-EXT-IPS:type": "floating", + }, + ] + }, + "links": [ + {"href": "http://example.com/images/95e4c4", "rel": "self"}, + { + "href": "http://example.com/images/95e4c4", + "rel": "bookmark", + }, + ], + "image": { + "id": "95e4c449-8abf-486e-97d9-dc3f82417d2d", + "links": [ + { + "href": "http://example.com/images/95e4c4", + "rel": "bookmark", + } + ], + }, + "OS-EXT-STS:vm_state": "active", + "OS-SRV-USG:launched_at": "2018-03-01T02:44:50.000000", + "flavor": { + "id": "3bd99062-2fe8-4eac-93f0-9200cc0f97ae", + "links": [ + { + "href": "http://example.com/flavors/95e4c4", + "rel": "bookmark", + } + ], + }, + "id": "97fe35e9-756a-41a2-960a-1d057d2c9ee4", + "security_groups": [{"name": "default"}], + "user_id": "c17534835f8f42bf98fc367e0bf35e09", + "OS-DCF:diskConfig": "MANUAL", + "accessIPv4": "", + "accessIPv6": "", + "progress": 0, + "OS-EXT-STS:power_state": 1, + "OS-EXT-AZ:availability_zone": "nova", + "metadata": {}, + "status": "ACTIVE", + "updated": "2018-03-01T02:44:51Z", + "hostId": "", + "OS-SRV-USG:terminated_at": None, + "key_name": None, + "name": "mttest", + "created": "2018-03-01T02:44:46Z", + "tenant_id": "65222a4d09ea4c68934fa1028c77f394", + "os-extended-volumes:volumes_attached": [], + "config_drive": "", + } + fake_networks = { + "networks": [ + { + "status": "ACTIVE", + "router:external": True, + "availability_zone_hints": [], + "availability_zones": ["nova"], + "description": None, + "subnets": [ + "df3e17fa-a4b2-47ae-9015-bc93eb076ba2", + "6b0c3dc9-b0b8-4d87-976a-7f2ebf13e7ec", + "fc541f48-fc7f-48c0-a063-18de6ee7bdd7", + ], + "shared": False, + "tenant_id": "a564613210ee43708b8a7fc6274ebd63", + "tags": [], + "ipv6_address_scope": "9f03124f-89af-483a-b6fd-10f08079db4d", # noqa: E501 + "mtu": 1550, + "is_default": False, + "admin_state_up": True, + "revision_number": 0, + "ipv4_address_scope": None, + "port_security_enabled": True, + "project_id": "a564613210ee43708b8a7fc6274ebd63", + "id": "0232c17f-2096-49bc-b205-d3dcd9a30ebf", + "name": "ext-net", + }, + { + "status": "ACTIVE", + "router:external": False, + "availability_zone_hints": [], + "availability_zones": ["nova"], + "description": "", + "subnets": ["f0ad1df5-53ee-473f-b86b-3604ea5591e9"], + "shared": False, + "tenant_id": "65222a4d09ea4c68934fa1028c77f394", + "created_at": "2016-10-22T13:46:26Z", + "tags": [], + "ipv6_address_scope": None, + "updated_at": "2016-10-22T13:46:26Z", + "admin_state_up": True, + "mtu": 1500, + "revision_number": 0, + "ipv4_address_scope": None, + "port_security_enabled": True, + "project_id": "65222a4d09ea4c68934fa1028c77f394", + "id": "2c9adcb5-c123-4c5a-a2ba-1ad4c4e1481f", + "name": "private", + }, + ] + } + fake_subnets = { + "subnets": [ + { + "service_types": [], + "description": "", + "enable_dhcp": True, + "tags": [], + "network_id": "827c6bb6-492f-4168-9577-f3a131eb29e8", + "tenant_id": "65222a4d09ea4c68934fa1028c77f394", + "created_at": "2017-06-12T13:23:57Z", + "dns_nameservers": [], + "updated_at": "2017-06-12T13:23:57Z", + "gateway_ip": "10.24.4.1", + "ipv6_ra_mode": None, + "allocation_pools": [ + {"start": "10.24.4.2", "end": "10.24.4.254"} + ], + "host_routes": [], + "revision_number": 0, + "ip_version": 4, + "ipv6_address_mode": None, + "cidr": "10.24.4.0/24", + "project_id": "65222a4d09ea4c68934fa1028c77f394", + "id": "3f0642d9-4644-4dff-af25-bcf64f739698", + "subnetpool_id": None, + "name": "foo_subnet", + }, + { + "service_types": [], + "description": "", + "enable_dhcp": True, + "tags": [], + "network_id": "2c9adcb5-c123-4c5a-a2ba-1ad4c4e1481f", + "tenant_id": "65222a4d09ea4c68934fa1028c77f394", + "created_at": "2016-10-22T13:46:26Z", + "dns_nameservers": ["89.36.90.101", "89.36.90.102"], + "updated_at": "2016-10-22T13:46:26Z", + "gateway_ip": "10.4.0.1", + "ipv6_ra_mode": None, + "allocation_pools": [ + {"start": "10.4.0.2", "end": "10.4.0.200"} + ], + "host_routes": [], + "revision_number": 0, + "ip_version": 4, + "ipv6_address_mode": None, + "cidr": "10.4.0.0/24", + "project_id": "65222a4d09ea4c68934fa1028c77f394", + "id": "f0ad1df5-53ee-473f-b86b-3604ea5591e9", + "subnetpool_id": None, + "name": "private-subnet-ipv4", + }, + ] + } + self.register_uris( + [ + self.get_nova_discovery_mock_dict(), + dict( + method='GET', + uri=self.get_mock_url( + 'compute', 'public', append=['servers', 'detail'] + ), + json={'servers': [fake_server]}, + ), + dict( + method='GET', + uri=self.get_mock_url( + 'network', 'public', append=['v2.0', 'networks'] + ), + json=fake_networks, + ), + dict( + method='GET', + uri=self.get_mock_url( + 'network', 'public', append=['v2.0', 'subnets'] + ), + json=fake_subnets, + ), + ] + ) + + r = self.cloud.get_server('97fe35e9-756a-41a2-960a-1d057d2c9ee4') + + self.assertEqual('10.4.0.13', r['private_v4']) + + self.assert_calls() + + def test_list_servers_all_projects(self): + """This test verifies that when list_servers is called with + `all_projects=True` that it passes `all_tenants=True` to nova.""" + self.register_uris( + [ + self.get_nova_discovery_mock_dict(), + dict( + method='GET', + uri=self.get_mock_url( + 'compute', + 'public', + append=['servers', 'detail'], + qs_elements=['all_tenants=True'], + ), + complete_qs=True, + json={'servers': []}, + ), + ] + ) + + self.cloud.list_servers(all_projects=True) + + self.assert_calls() + + def test_list_servers_filters(self): + """This test verifies that when list_servers is called with + `filters` dict that it passes it to nova.""" + self.register_uris( + [ + self.get_nova_discovery_mock_dict(), + dict( + method='GET', + uri=self.get_mock_url( + 'compute', + 'public', + append=['servers', 'detail'], + qs_elements=[ + 'deleted=True', + 'changes-since=2014-12-03T00:00:00Z', + ], + ), + complete_qs=True, + json={'servers': []}, + ), + ] + ) + + self.cloud.list_servers( + filters={'deleted': True, 'changes-since': '2014-12-03T00:00:00Z'} + ) + + self.assert_calls() + + def test_list_servers_exception(self): + self.register_uris( + [ + self.get_nova_discovery_mock_dict(), + dict( + method='GET', + uri=self.get_mock_url( + 'compute', 'public', append=['servers', 'detail'] + ), + status_code=400, + ), + ] + ) + + self.assertRaises(exc.OpenStackCloudException, self.cloud.list_servers) + + self.assert_calls() diff --git a/openstack/tests/unit/cloud/test_image.py b/openstack/tests/unit/cloud/test_image.py index 889a28d32..731ab7ee3 100644 --- a/openstack/tests/unit/cloud/test_image.py +++ b/openstack/tests/unit/cloud/test_image.py @@ -15,10 +15,12 @@ import io import operator import tempfile +from unittest import mock import uuid from openstack.cloud import exc from openstack.cloud import meta +from openstack import connection from openstack import exceptions from openstack.image.v1 import image as image_v1 from openstack.image.v2 import image @@ -164,6 +166,20 @@ class TestImage(BaseTestImage): self.assertEqual(output_file.read(), self.output) self.assert_calls() + @mock.patch.object(connection.Connection, 'search_images') + def test_get_images(self, mock_search): + image1 = dict(id='123', name='mickey') + mock_search.return_value = [image1] + r = self.cloud.get_image('mickey') + self.assertIsNotNone(r) + self.assertDictEqual(image1, r) + + @mock.patch.object(connection.Connection, 'search_images') + def test_get_image_not_found(self, mock_search): + mock_search.return_value = [] + r = self.cloud.get_image('doesNotExist') + self.assertIsNone(r) + def test_get_image_name(self, cloud=None): cloud = cloud or self.cloud self.register_uris( diff --git a/openstack/tests/unit/cloud/test_network.py b/openstack/tests/unit/cloud/test_network.py index 717697f09..04880b9e1 100644 --- a/openstack/tests/unit/cloud/test_network.py +++ b/openstack/tests/unit/cloud/test_network.py @@ -11,16 +11,132 @@ # limitations under the License. import copy +from unittest import mock import testtools import openstack import openstack.cloud +from openstack import exceptions from openstack.network.v2 import network as _network from openstack.tests.unit import base -class TestNetwork(base.TestCase): +class TestNeutronExtensions(base.TestCase): + def test__neutron_extensions(self): + body = [ + { + "updated": "2014-06-1T10:00:00-00:00", + "name": "Distributed Virtual Router", + "links": [], + "alias": "dvr", + "description": "Enables configuration of Distributed Virtual Routers.", # noqa: E501 + }, + { + "updated": "2013-07-23T10:00:00-00:00", + "name": "Allowed Address Pairs", + "links": [], + "alias": "allowed-address-pairs", + "description": "Provides allowed address pairs", + }, + ] + self.register_uris( + [ + dict( + method='GET', + uri=self.get_mock_url( + 'network', 'public', append=['v2.0', 'extensions'] + ), + json=dict(extensions=body), + ) + ] + ) + extensions = self.cloud._neutron_extensions() + self.assertEqual(set(['dvr', 'allowed-address-pairs']), extensions) + + self.assert_calls() + + def test__neutron_extensions_fails(self): + self.register_uris( + [ + dict( + method='GET', + uri=self.get_mock_url( + 'network', 'public', append=['v2.0', 'extensions'] + ), + status_code=404, + ) + ] + ) + with testtools.ExpectedException(exceptions.ResourceNotFound): + self.cloud._neutron_extensions() + + self.assert_calls() + + def test__has_neutron_extension(self): + body = [ + { + "updated": "2014-06-1T10:00:00-00:00", + "name": "Distributed Virtual Router", + "links": [], + "alias": "dvr", + "description": "Enables configuration of Distributed Virtual Routers.", # noqa: E501 + }, + { + "updated": "2013-07-23T10:00:00-00:00", + "name": "Allowed Address Pairs", + "links": [], + "alias": "allowed-address-pairs", + "description": "Provides allowed address pairs", + }, + ] + self.register_uris( + [ + dict( + method='GET', + uri=self.get_mock_url( + 'network', 'public', append=['v2.0', 'extensions'] + ), + json=dict(extensions=body), + ) + ] + ) + self.assertTrue(self.cloud._has_neutron_extension('dvr')) + self.assert_calls() + + def test__has_neutron_extension_missing(self): + body = [ + { + "updated": "2014-06-1T10:00:00-00:00", + "name": "Distributed Virtual Router", + "links": [], + "alias": "dvr", + "description": "Enables configuration of Distributed Virtual Routers.", # noqa: E501 + }, + { + "updated": "2013-07-23T10:00:00-00:00", + "name": "Allowed Address Pairs", + "links": [], + "alias": "allowed-address-pairs", + "description": "Provides allowed address pairs", + }, + ] + self.register_uris( + [ + dict( + method='GET', + uri=self.get_mock_url( + 'network', 'public', append=['v2.0', 'extensions'] + ), + json=dict(extensions=body), + ) + ] + ) + self.assertFalse(self.cloud._has_neutron_extension('invalid')) + self.assert_calls() + + +class TestNetworks(base.TestCase): mock_new_network_rep = { 'provider:physical_network': None, 'ipv6_address_scope': None, @@ -109,6 +225,12 @@ class TestNetwork(base.TestCase): self.cloud.list_networks(filters={'name': 'test'}) self.assert_calls() + def test_list_networks_neutron_not_found(self): + self.use_nothing() + self.cloud.has_service = mock.Mock(return_value=False) + self.assertEqual([], self.cloud.list_networks()) + self.assert_calls() + def test_create_network(self): self.register_uris( [