openstack/openstacksdk
Code Issues Proposed changes

tests: Move cloud-layer service tests to their own file

Browse Source 
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 <stephenfin@redhat.com>
This commit is contained in:
Stephen Finucane
2023-05-25 12:00:29 +01:00
parent 97d0349686
commit d67006ea0f
4 changed files with 639 additions and 616 deletions
Download Patch File Download Diff File Expand all files Collapse all files

615
openstack/tests/unit/cloud/test_cloud.py
View File

@@ -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)

500
openstack/tests/unit/cloud/test_compute.py Normal file
View File

@@ -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()

16
openstack/tests/unit/cloud/test_image.py
View File

@@ -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(

124
openstack/tests/unit/cloud/test_network.py
View File

@@ -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(
[