FM / cloud

/opt/imh-python/lib/python3.9/site-packages/openstack/tests/unit/cloud/

/opt/imh-python/lib/python3.9/site-packages/openstack/tests/unit/cloud

Name	Size	Mode	Actions
__pycache__/	-	0755	rm
test_accelerator.py	11346	0644	edit dl rm
test_aggregate.py	7153	0644	edit dl rm
test_availability_zones.py	2308	0644	edit dl rm
test_baremetal_node.py	68958	0644	edit dl rm
test_baremetal_ports.py	4356	0644	edit dl rm
test_caching.py	25060	0644	edit dl rm
test_clustering.py	25641	0644	edit dl rm
test_cluster_templates.py	9946	0644	edit dl rm
test_coe_clusters.py	6118	0644	edit dl rm
test_coe_clusters_certificate.py	2565	0644	edit dl rm
test_create_server.py	39514	0644	edit dl rm
test_create_volume_snapshot.py	5571	0644	edit dl rm
test_delete_server.py	10622	0644	edit dl rm
test_delete_volume_snapshot.py	4040	0644	edit dl rm
test_domains.py	9858	0644	edit dl rm
test_domain_params.py	2903	0644	edit dl rm
test_endpoints.py	16035	0644	edit dl rm
test_flavors.py	11861	0644	edit dl rm
test_floating_ip_common.py	8270	0644	edit dl rm
test_floating_ip_neutron.py	48972	0644	edit dl rm
test_floating_ip_nova.py	11122	0644	edit dl rm
test_floating_ip_pool.py	2886	0644	edit dl rm
test_fwaas.py	52343	0644	edit dl rm
test_groups.py	3886	0644	edit dl rm
test_identity_roles.py	12236	0644	edit dl rm
test_image.py	60601	0644	edit dl rm
test_image_snapshot.py	4171	0644	edit dl rm
test_inventory.py	5826	0644	edit dl rm
test_keypair.py	4910	0644	edit dl rm
test_limits.py	3929	0644	edit dl rm
test_magnum_services.py	1378	0644	edit dl rm
test_meta.py	43155	0644	edit dl rm
test_network.py	14095	0644	edit dl rm
test_normalize.py	58339	0644	edit dl rm
test_object.py	51327	0644	edit dl rm
test_operator.py	6488	0644	edit dl rm
test_operator_noauth.py	8188	0644	edit dl rm
test_port.py	14285	0644	edit dl rm
test_project.py	11156	0644	edit dl rm
test_qos_bandwidth_limit_rule.py	17327	0644	edit dl rm
test_qos_dscp_marking_rule.py	12224	0644	edit dl rm
test_qos_minimum_bandwidth_rule.py	12399	0644	edit dl rm
test_qos_policy.py	13244	0644	edit dl rm
test_qos_rule_type.py	5648	0644	edit dl rm
test_quotas.py	9388	0644	edit dl rm
test_rebuild_server.py	9479	0644	edit dl rm
test_recordset.py	14767	0644	edit dl rm
test_role_assignment.py	150282	0644	edit dl rm
test_router.py	17107	0644	edit dl rm
test_security_groups.py	32570	0644	edit dl rm
test_server_console.py	2846	0644	edit dl rm
test_server_delete_metadata.py	2691	0644	edit dl rm
test_server_group.py	2366	0644	edit dl rm
test_server_set_metadata.py	2784	0644	edit dl rm
test_services.py	11700	0644	edit dl rm
test_shade.py	27060	0644	edit dl rm
test_shade_operator.py	743	0644	edit dl rm
test_stack.py	26453	0644	edit dl rm
test_subnet.py	18581	0644	edit dl rm
test_update_server.py	3339	0644	edit dl rm
test_usage.py	2595	0644	edit dl rm
test_users.py	9044	0644	edit dl rm
test_volume.py	23505	0644	edit dl rm
test_volume_access.py	8556	0644	edit dl rm
test_volume_backups.py	8947	0644	edit dl rm
test_zone.py	7635	0644	edit dl rm
test__utils.py	13857	0644	edit dl rm
__init__.py	0	0644	edit dl rm

Edit: /opt/imh-python/lib/python3.9/site-packages/openstack/tests/unit/cloud/test_shade.py (27060B)

# 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 unittest import mock
import uuid

import testtools

from openstack.cloud import exc
from openstack import connection
from openstack.tests import fakes
from openstack.tests.unit import base
from openstack import utils

RANGE_DATA = [
    dict(id=1, key1=1, key2=5),
    dict(id=2, key1=1, key2=20),
    dict(id=3, key1=2, key2=10),
    dict(id=4, key1=2, key2=30),
    dict(id=5, key1=3, key2=40),
    dict(id=6, key1=3, key2=40),
]

class TestShade(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(TestShade, self).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)

def test_endpoint_for(self):
        dns_override = 'https://override.dns.example.com'
        self.cloud.config.config['dns_endpoint_override'] = dns_override
        self.assertEqual(
            'https://compute.example.com/v2.1/',
            self.cloud.endpoint_for('compute'))
        self.assertEqual(
            'https://internal.compute.example.com/v2.1/',
            self.cloud.endpoint_for('compute', interface='internal'))
        self.assertIsNone(
            self.cloud.endpoint_for('compute', region_name='unknown-region'))
        self.assertEqual(
            dns_override,
            self.cloud.endpoint_for('dns'))

def test_connect_as(self):
        # Do initial auth/catalog steps
        # This should authenticate a second time, but should not
        # need a second identity discovery
        project_name = 'test_project'
        self.register_uris([
            self.get_keystone_v3_token(project_name=project_name),
            self.get_nova_discovery_mock_dict(),
            dict(
                method='GET',
                uri=self.get_mock_url(
                    'compute', 'public', append=['servers', 'detail']),
                json={'servers': []},
            ),
        ])

c2 = self.cloud.connect_as(project_name=project_name)
        self.assertEqual(c2.list_servers(), [])
        self.assert_calls()

def test_connect_as_context(self):
        # Do initial auth/catalog steps
        # This should authenticate a second time, but should not
        # need a second identity discovery
        project_name = 'test_project'
        self.register_uris([
            self.get_keystone_v3_token(project_name=project_name),
            self.get_nova_discovery_mock_dict(),
            dict(
                method='GET',
                uri=self.get_mock_url(
                    'compute', 'public', append=['servers', 'detail']),
                json={'servers': []},
            ),
        ])

with self.cloud.connect_as(project_name=project_name) as c2:
            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([
            self.get_nova_discovery_mock_dict(),
            dict(
                method='GET',
                uri=self.get_mock_url(
                    'compute', 'public', append=['servers', 'detail']),
                json={'servers': []},
                validate=dict(
                    headers={'X-Openstack-Request-Id': request_id}),
            ),
        ])

cloud2 = self.cloud.global_request(request_id)
        self.assertEqual([], cloud2.list_servers())

self.assert_calls()

def test_global_request_id_context(self):
        request_id = uuid.uuid4().hex
        self.register_uris([
            self.get_nova_discovery_mock_dict(),
            dict(
                method='GET',
                uri=self.get_mock_url(
                    'compute', 'public', append=['servers', 'detail']),
                json={'servers': []},
                validate=dict(
                    headers={'X-Openstack-Request-Id': request_id}),
            ),
        ])

with self.cloud.global_request(request_id) as c2:
            self.assertEqual([], c2.list_servers())

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",
                "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']),
                 json=fake_networks),
            dict(method='GET',
                 uri=self.get_mock_url(
                     'network', 'public', append=['v2.0', 'subnets.json']),
                 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,
                "Wait value must be an int or float value."):
            for count in utils.iterate_timeout(
                    1, "test_iterate_timeout_bad_wait", wait="timeishard"):
                pass

@mock.patch('time.sleep')
    def test_iterate_timeout_str_wait(self, mock_sleep):
        iter = utils.iterate_timeout(
            10, "test_iterate_timeout_str_wait", wait="1.6")
        next(iter)
        next(iter)
        mock_sleep.assert_called_with(1.6)

@mock.patch('time.sleep')
    def test_iterate_timeout_int_wait(self, mock_sleep):
        iter = utils.iterate_timeout(
            10, "test_iterate_timeout_int_wait", wait=1)
        next(iter)
        next(iter)
        mock_sleep.assert_called_with(1.0)

@mock.patch('time.sleep')
    def test_iterate_timeout_timeout(self, mock_sleep):
        message = "timeout test"
        with testtools.ExpectedException(
                exc.OpenStackCloudTimeout,
                message):
            for count in utils.iterate_timeout(0.1, message, wait=1):
                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),
        ])
        with testtools.ExpectedException(
            exc.OpenStackCloudURINotFound,
            "Error fetching extension list for nova"
        ):
            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."
            },
            {
                "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']),
                 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.json']),
                 status_code=404)
        ])
        with testtools.ExpectedException(
            exc.OpenStackCloudURINotFound,
            "Error fetching extension list for neutron"
        ):
            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."
            },
            {
                "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']),
                 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."
            },
            {
                "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']),
                 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)
        self.assertIsInstance(retval, list)
        self.assertEqual(1, len(retval))
        self.assertEqual([RANGE_DATA[1]], retval)

def test_range_search_2(self):
        filters = {"key1": "<=2", "key2": ">10"}
        retval = self.cloud.range_search(RANGE_DATA, filters)
        self.assertIsInstance(retval, list)
        self.assertEqual(2, len(retval))
        self.assertEqual([RANGE_DATA[1], RANGE_DATA[3]], retval)

def test_range_search_3(self):
        filters = {"key1": "2", "key2": "min"}
        retval = self.cloud.range_search(RANGE_DATA, filters)
        self.assertIsInstance(retval, list)
        self.assertEqual(0, len(retval))

def test_range_search_4(self):
        filters = {"key1": "max", "key2": "min"}
        retval = self.cloud.range_search(RANGE_DATA, filters)
        self.assertIsInstance(retval, list)
        self.assertEqual(0, len(retval))

def test_range_search_5(self):
        filters = {"key1": "min", "key2": "min"}
        retval = self.cloud.range_search(RANGE_DATA, filters)
        self.assertIsInstance(retval, list)
        self.assertEqual(1, len(retval))
        self.assertEqual([RANGE_DATA[0]], retval)

cmd: