<commit_before># Copyright 2012 NEC Corporation
#
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.
import copy

from mox3.mox import IsA

from django import http
from django.test.utils import override_settings

from oslo_utils import uuidutils

from neutronclient.common import exceptions as neutron_exc

from openstack_dashboard import api
from openstack_dashboard import policy
from openstack_dashboard.test import helpers as test


class NeutronApiTests(test.APITestCase):
    def test_network_list(self):
        networks = {'networks': self.api_networks.list()}
        subnets = {'subnets': self.api_subnets.list()}

        neutronclient = self.stub_neutronclient()
        neutronclient.list_networks().AndReturn(networks)
        neutronclient.list_subnets().AndReturn(subnets)
        self.mox.ReplayAll()

        ret_val = api.neutron.network_list(self.request)
        for n in ret_val:
            self.assertIsInstance(n, api.neutron.Network)

    @test.create_stubs({api.neutron: ('network_list',
                                      'subnet_list')})
    def _test_network_list_for_tenant(
            self, include_external,
            filter_params, should_called):
        """Convenient method to test network_list_for_tenant.

        :param include_external: Passed to network_list_for_tenant.
        :param filter_params: Filters passed to network_list_for_tenant
        :param should_called: this argument specifies which methods
            should be called. Methods in this list should be called.
            Valid values are non_shared, shared, and external.
        """
        filter_params = filter_params or {}
        all_networks = self.networks.list()
        tenant_id = '1'
        if 'non_shared' in should_called:
            params = filter_params.copy()
            params['shared'] = False
            api.neutron.network_list(
                IsA(http.HttpRequest),
                tenant_id=tenant_id,
                **params).AndReturn([
                    network for network in all_networks
                    if network['tenant_id'] == tenant_id
                ])
        if 'shared' in should_called:
            params = filter_params.copy()
            params['shared'] = True
            api.neutron.network_list(
                IsA(http.HttpRequest),
                **params).AndReturn([
                    network for network in all_networks
                    if network.get('shared')
                ])
        if 'external' in should_called:
            params = filter_params.copy()
            params['router:external'] = True
            api.neutron.network_list(
                IsA(http.HttpRequest), **params).AndReturn([
                    network for network in all_networks
                    if network.get('router:external')
                ])
        self.mox.ReplayAll()

        ret_val = api.neutron.network_list_for_tenant(
            self.request, tenant_id,
            include_external=include_external,
            **filter_params)

        expected = [n for n in all_networks
                    if (('non_shared' in should_called and
                         n['tenant_id'] == tenant_id) or
                        ('shared' in should_called and n['shared']) or
                        ('external' in should_called and
                         include_external and n['router:external']))]
        self.assertEqual(set(n.id for n in expected),
                         set(n.id for n in ret_val))

    def test_network_list_for_tenant(self):
        self._test_network_list_for_tenant(
            include_external=False, filter_params=None,
            should_called=['non_shared', 'shared'])

    def test_network_list_for_tenant_with_external(self):
        self._test_network_list_for_tenant(
            include_external=True, filter_params=None,
            should_called=['non_shared', 'shared', 'external'])

    def test_network_list_for_tenant_with_filters_shared_false_wo_incext(self):
        self._test_network_list_for_tenant(
            include_external=False, filter_params={'shared': True},
            should_called=['shared'])

    def test_network_list_for_tenant_with_filters_shared_true_w_incext(self):
        self._test_network_list_for_tenant(
            include_external=True, filter_params={'shared': True},
            should_called=['shared', 'external'])

    def test_network_list_for_tenant_with_filters_ext_false_wo_incext(self):
        self._test_network_list_for_tenant(
            include_external=False, filter_params={'router:external': False},
            should_called=['non_shared', 'shared'])

    def test_network_list_for_tenant_with_filters_ext_true_wo_incext(self):
        self._test_network_list_for_tenant(
            include_external=False, filter_params={'router:external': True},
            should_called=['non_shared', 'shared'])

    def test_network_list_for_tenant_with_filters_ext_false_w_incext(self):
        self._test_network_list_for_tenant(
            include_external=True, filter_params={'router:external': False},
            should_called=['non_shared', 'shared'])

    def test_network_list_for_tenant_with_filters_ext_true_w_incext(self):
        self._test_network_list_for_tenant(
            include_external=True, filter_params={'router:external': True},
            should_called=['non_shared', 'shared', 'external'])

    def test_network_list_for_tenant_with_filters_both_shared_ext(self):
        # To check 'shared' filter is specified in network_list
        # to look up external networks.
        self._test_network_list_for_tenant(
            include_external=True,
            filter_params={'router:external': True, 'shared': True},
            should_called=['shared', 'external'])

    def test_network_list_for_tenant_with_other_filters(self):
        # To check filter parameters other than shared and
        # router:external are passed as expected.
        self._test_network_list_for_tenant(
            include_external=True,
            filter_params={'router:external': True, 'shared': False,
                           'foo': 'bar'},
            should_called=['non_shared', 'external'])

    def test_network_get(self):
        network = {'network': self.api_networks.first()}
        subnet = {'subnet': self.api_subnets.first()}
        network_id = self.api_networks.first()['id']
        subnet_id = self.api_networks.first()['subnets'][0]

        neutronclient = self.stub_neutronclient()
        neutronclient.show_network(network_id).AndReturn(network)
        neutronclient.show_subnet(subnet_id).AndReturn(subnet)
        self.mox.ReplayAll()

        ret_val = api.neutron.network_get(self.request, network_id)
        self.assertIsInstance(ret_val, api.neutron.Network)

    def test_network_create(self):
        network = {'network': self.api_networks.first()}
        form_data = {'network': {'name': 'net1',
                                 'tenant_id': self.request.user.project_id}}
        neutronclient = self.stub_neutronclient()
        neutronclient.create_network(body=form_data).AndReturn(network)
        self.mox.ReplayAll()
        ret_val = api.neutron.network_create(self.request, name='net1')
        self.assertIsInstance(ret_val, api.neutron.Network)

    def test_network_update(self):
        network = {'network': self.api_networks.first()}
        network_id = self.api_networks.first()['id']

        neutronclient = self.stub_neutronclient()
        form_data = {'network': {'name': 'net1'}}
        neutronclient.update_network(network_id, body=form_data)\
            .AndReturn(network)
        self.mox.ReplayAll()

        ret_val = api.neutron.network_update(self.request, network_id,
                                             name='net1')
        self.assertIsInstance(ret_val, api.neutron.Network)

    def test_network_delete(self):
        network_id = self.api_networks.first()['id']

        neutronclient = self.stub_neutronclient()
        neutronclient.delete_network(network_id)
        self.mox.ReplayAll()

        api.neutron.network_delete(self.request, network_id)

    def test_get_network_ip_availability(self):
        network = {'network': self.api_networks.first()}
        mock_ip_availability = self.ip_availability.get()
        neutronclient = self.stub_neutronclient()
        neutronclient.show_network_ip_availability(network).\
            AndReturn(mock_ip_availability)

        self.mox.ReplayAll()
        ret_val = api.neutron.show_network_ip_availability(self.request,
                                                           network)

        self.assertIsInstance(ret_val, dict)

    def test_subnet_network_ip_availability(self):
        network = {'network': self.api_networks.first()}
        mock_ip_availability = self.ip_availability.get()
        neutronclient = self.stub_neutronclient()
        neutronclient.show_network_ip_availability(network).\
            AndReturn(mock_ip_availability)

        self.mox.ReplayAll()
        ip_availability = api.neutron. \
            show_network_ip_availability(self.request, network)
        availabilities = ip_availability.get("network_ip_availability",
                                             {})
        ret_val = availabilities.get("subnet_ip_availability", [])

        self.assertIsInstance(ret_val, list)

    def test_subnet_list(self):
        subnets = {'subnets': self.api_subnets.list()}

        neutronclient = self.stub_neutronclient()
        neutronclient.list_subnets().AndReturn(subnets)
        self.mox.ReplayAll()

        ret_val = api.neutron.subnet_list(self.request)
        for n in ret_val:
            self.assertIsInstance(n, api.neutron.Subnet)

    def test_subnet_get(self):
        subnet = {'subnet': self.api_subnets.first()}
        subnet_id = self.api_subnets.first()['id']

        neutronclient = self.stub_neutronclient()
        neutronclient.show_subnet(subnet_id).AndReturn(subnet)
        self.mox.ReplayAll()

        ret_val = api.neutron.subnet_get(self.request, subnet_id)
        self.assertIsInstance(ret_val, api.neutron.Subnet)

    def test_subnet_create(self):
        subnet_data = self.api_subnets.first()
        params = {'network_id': subnet_data['network_id'],
                  'tenant_id': subnet_data['tenant_id'],
                  'name': subnet_data['name'],
                  'cidr': subnet_data['cidr'],
                  'ip_version': subnet_data['ip_version'],
                  'gateway_ip': subnet_data['gateway_ip']}

        neutronclient = self.stub_neutronclient()
        neutronclient.create_subnet(body={'subnet': params})\
            .AndReturn({'subnet': subnet_data})
        self.mox.ReplayAll()

        ret_val = api.neutron.subnet_create(self.request, **params)
        self.assertIsInstance(ret_val, api.neutron.Subnet)

    def test_subnet_update(self):
        subnet_data = self.api_subnets.first()
        subnet_id = subnet_data['id']
        params = {'name': subnet_data['name'],
                  'gateway_ip': subnet_data['gateway_ip']}

        neutronclient = self.stub_neutronclient()
        neutronclient.update_subnet(subnet_id, body={'subnet': params})\
            .AndReturn({'subnet': subnet_data})
        self.mox.ReplayAll()

        ret_val = api.neutron.subnet_update(self.request, subnet_id, **params)
        self.assertIsInstance(ret_val, api.neutron.Subnet)

    def test_subnet_delete(self):
        subnet_id = self.api_subnets.first()['id']

        neutronclient = self.stub_neutronclient()
        neutronclient.delete_subnet(subnet_id)
        self.mox.ReplayAll()

        api.neutron.subnet_delete(self.request, subnet_id)

    def test_subnetpool_list(self):
        subnetpools = {'subnetpools': self.api_subnetpools.list()}

        neutronclient = self.stub_neutronclient()
        neutronclient.list_subnetpools().AndReturn(subnetpools)
        self.mox.ReplayAll()

        ret_val = api.neutron.subnetpool_list(self.request)
        for n in ret_val:
            self.assertIsInstance(n, api.neutron.SubnetPool)

    def test_subnetpool_get(self):
        subnetpool = {'subnetpool': self.api_subnetpools.first()}
        subnetpool_id = self.api_subnetpools.first()['id']

        neutronclient = self.stub_neutronclient()
        neutronclient.show_subnetpool(subnetpool_id).AndReturn(subnetpool)
        self.mox.ReplayAll()

        ret_val = api.neutron.subnetpool_get(self.request, subnetpool_id)
        self.assertIsInstance(ret_val, api.neutron.SubnetPool)

    def test_subnetpool_create(self):
        subnetpool_data = self.api_subnetpools.first()
        params = {'name': subnetpool_data['name'],
                  'prefixes': subnetpool_data['prefixes'],
                  'tenant_id': subnetpool_data['tenant_id']}

        neutronclient = self.stub_neutronclient()
        neutronclient.create_subnetpool(body={'subnetpool': params})\
            .AndReturn({'subnetpool': subnetpool_data})
        self.mox.ReplayAll()

        ret_val = api.neutron.subnetpool_create(self.request, **params)
        self.assertIsInstance(ret_val, api.neutron.SubnetPool)

    def test_subnetpool_update(self):
        subnetpool_data = self.api_subnetpools.first()
        subnetpool_id = subnetpool_data['id']
        params = {'name': subnetpool_data['name'],
                  'prefixes': subnetpool_data['prefixes']}

        neutronclient = self.stub_neutronclient()
        neutronclient.update_subnetpool(subnetpool_id, body={'subnetpool': params})\
            .AndReturn({'subnetpool': subnetpool_data})
        self.mox.ReplayAll()

        ret_val = api.neutron.subnetpool_update(self.request, subnetpool_id,
                                                **params)
        self.assertIsInstance(ret_val, api.neutron.SubnetPool)

    def test_subnetpool_delete(self):
        subnetpool_id = self.api_subnetpools.first()['id']

        neutronclient = self.stub_neutronclient()
        neutronclient.delete_subnetpool(subnetpool_id)
        self.mox.ReplayAll()

        api.neutron.subnetpool_delete(self.request, subnetpool_id)

    def test_port_list(self):
        ports = {'ports': self.api_ports.list()}

        neutronclient = self.stub_neutronclient()
        neutronclient.list_ports().AndReturn(ports)
        self.mox.ReplayAll()

        ret_val = api.neutron.port_list(self.request)
        for p in ret_val:
            self.assertIsInstance(p, api.neutron.Port)

    def test_port_get(self):
        port = {'port': self.api_ports.first()}
        port_id = self.api_ports.first()['id']

        neutronclient = self.stub_neutronclient()
        neutronclient.show_port(port_id).AndReturn(port)
        self.mox.ReplayAll()

        ret_val = api.neutron.port_get(self.request, port_id)
        self.assertIsInstance(ret_val, api.neutron.Port)

    def test_port_create(self):
        port = {'port': self.api_ports.first()}
        params = {'network_id': port['port']['network_id'],
                  'tenant_id': port['port']['tenant_id'],
                  'name': port['port']['name'],
                  'device_id': port['port']['device_id']}

        neutronclient = self.stub_neutronclient()
        neutronclient.create_port(body={'port': params}).AndReturn(port)
        self.mox.ReplayAll()
        ret_val = api.neutron.port_create(self.request, **params)
        self.assertIsInstance(ret_val, api.neutron.Port)
        self.assertEqual(api.neutron.Port(port['port']).id, ret_val.id)

    def test_port_update(self):
        port_data = self.api_ports.first()
        port_id = port_data['id']
        params = {'name': port_data['name'],
                  'device_id': port_data['device_id']}

        neutronclient = self.stub_neutronclient()
        neutronclient.update_port(port_id, body={'port': params})\
            .AndReturn({'port': port_data})
        self.mox.ReplayAll()

        ret_val = api.neutron.port_update(self.request, port_id, **params)
        self.assertIsInstance(ret_val, api.neutron.Port)
        self.assertEqual(api.neutron.Port(port_data).id, ret_val.id)

    def test_port_delete(self):
        port_id = self.api_ports.first()['id']

        neutronclient = self.stub_neutronclient()
        neutronclient.delete_port(port_id)
        self.mox.ReplayAll()

        api.neutron.port_delete(self.request, port_id)

    def test_trunk_list(self):
        trunks = {'trunks': self.api_trunks.list()}
        neutron_client = self.stub_neutronclient()
        neutron_client.list_trunks().AndReturn(trunks)
        self.mox.ReplayAll()

        ret_val = api.neutron.trunk_list(self.request)
        for t in ret_val:
            self.assertIsInstance(t, api.neutron.Trunk)

    def test_trunk_object(self):
        trunk = self.api_trunks.first().copy()
        obj = api.neutron.Trunk(trunk)
        self.assertEqual(0, obj.subport_count)
        trunk_dict = obj.to_dict()
        self.assertIsInstance(trunk_dict, dict)
        self.assertEqual(trunk['name'], trunk_dict['name_or_id'])
        self.assertEqual(0, trunk_dict['subport_count'])

        trunk['name'] = ''  # to test name_or_id
        trunk['sub_ports'] = [uuidutils.generate_uuid() for i in range(2)]
        obj = api.neutron.Trunk(trunk)
        self.assertEqual(2, obj.subport_count)
        trunk_dict = obj.to_dict()
        self.assertEqual(obj.name_or_id, trunk_dict['name_or_id'])
        self.assertEqual(2, trunk_dict['subport_count'])

    def test_router_list(self):
        routers = {'routers': self.api_routers.list()}

        neutronclient = self.stub_neutronclient()
        neutronclient.list_routers().AndReturn(routers)
        self.mox.ReplayAll()

        ret_val = api.neutron.router_list(self.request)
        for n in ret_val:
            self.assertIsInstance(n, api.neutron.Router)

    def test_router_get(self):
        router = {'router': self.api_routers.first()}
        router_id = self.api_routers.first()['id']

        neutronclient = self.stub_neutronclient()
        neutronclient.show_router(router_id).AndReturn(router)
        self.mox.ReplayAll()

        ret_val = api.neutron.router_get(self.request, router_id)
        self.assertIsInstance(ret_val, api.neutron.Router)

    def test_router_create(self):
        router = {'router': self.api_routers.first()}

        neutronclient = self.stub_neutronclient()
        form_data = {'router': {'name': 'router1',
                                'tenant_id': self.request.user.project_id}}
        neutronclient.create_router(body=form_data).AndReturn(router)
        self.mox.ReplayAll()

        ret_val = api.neutron.router_create(self.request, name='router1')
        self.assertIsInstance(ret_val, api.neutron.Router)

    def test_router_delete(self):
        router_id = self.api_routers.first()['id']

        neutronclient = self.stub_neutronclient()
        neutronclient.delete_router(router_id)
        self.mox.ReplayAll()

        api.neutron.router_delete(self.request, router_id)

    def test_router_add_interface(self):
        subnet_id = self.api_subnets.first()['id']
        router_id = self.api_routers.first()['id']

        neutronclient = self.stub_neutronclient()
        form_data = {'subnet_id': subnet_id}
        neutronclient.add_interface_router(
            router_id, form_data).AndReturn(None)
        self.mox.ReplayAll()

        api.neutron.router_add_interface(
            self.request, router_id, subnet_id=subnet_id)

    def test_router_remove_interface(self):
        router_id = self.api_routers.first()['id']
        fake_port = self.api_ports.first()['id']

        neutronclient = self.stub_neutronclient()
        neutronclient.remove_interface_router(
            router_id, {'port_id': fake_port})
        self.mox.ReplayAll()

        api.neutron.router_remove_interface(
            self.request, router_id, port_id=fake_port)

    @test.create_stubs({api.neutron: ('is_extension_supported',)})
    def test_is_extension_supported(self):
        api.neutron.is_extension_supported(self.request, "quotas")\
            .AndReturn(True)
        api.neutron.is_extension_supported(self.request, "doesntexist") \
            .AndReturn(False)
        self.mox.ReplayAll()

        self.assertTrue(
            api.neutron.is_extension_supported(self.request, 'quotas'))
        self.assertFalse(
            api.neutron.is_extension_supported(self.request, 'doesntexist'))<commit_msg>Fix neutron API tests<commit_after># Copyright 2012 NEC Corporation
#
#    Licensed under the Apache License, Version 2.0 (the "License"); you may
#    not use this file except in compliance with the License. You may obtain
#    a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#    License for the specific language governing permissions and limitations
#    under the License.
import copy

from mox3.mox import IsA

from django import http
from django.test.utils import override_settings

from oslo_utils import uuidutils

from neutronclient.common import exceptions as neutron_exc

from openstack_dashboard import api
from openstack_dashboard import policy
from openstack_dashboard.test import helpers as test


class NeutronApiTests(test.APITestCase):
    def test_network_list(self):
        networks = {'networks': self.api_networks.list()}
        subnets = {'subnets': self.api_subnets.list()}

        neutronclient = self.stub_neutronclient()
        neutronclient.list_networks().AndReturn(networks)
        neutronclient.list_subnets().AndReturn(subnets)
        self.mox.ReplayAll()

        ret_val = api.neutron.network_list(self.request)
        for n in ret_val:
            self.assertIsInstance(n, api.neutron.Network)

    @test.create_stubs({api.neutron: ('network_list',
                                      'subnet_list')})
    def _test_network_list_for_tenant(
            self, include_external,
            filter_params, should_called):
        """Convenient method to test network_list_for_tenant.

        :param include_external: Passed to network_list_for_tenant.
        :param filter_params: Filters passed to network_list_for_tenant
        :param should_called: this argument specifies which methods
            should be called. Methods in this list should be called.
            Valid values are non_shared, shared, and external.
        """
        filter_params = filter_params or {}
        all_networks = self.networks.list()
        tenant_id = '1'
        if 'non_shared' in should_called:
            params = filter_params.copy()
            params['shared'] = False
            api.neutron.network_list(
                IsA(http.HttpRequest),
                tenant_id=tenant_id,
                **params).AndReturn([
                    network for network in all_networks
                    if network['tenant_id'] == tenant_id
                ])
        if 'shared' in should_called:
            params = filter_params.copy()
            params['shared'] = True
            api.neutron.network_list(
                IsA(http.HttpRequest),
                **params).AndReturn([
                    network for network in all_networks
                    if network.get('shared')
                ])
        if 'external' in should_called:
            params = filter_params.copy()
            params['router:external'] = True
            api.neutron.network_list(
                IsA(http.HttpRequest), **params).AndReturn([
                    network for network in all_networks
                    if network.get('router:external')
                ])
        self.mox.ReplayAll()

        ret_val = api.neutron.network_list_for_tenant(
            self.request, tenant_id,
            include_external=include_external,
            **filter_params)

        expected = [n for n in all_networks
                    if (('non_shared' in should_called and
                         n['tenant_id'] == tenant_id) or
                        ('shared' in should_called and n['shared']) or
                        ('external' in should_called and
                         include_external and n['router:external']))]
        self.assertEqual(set(n.id for n in expected),
                         set(n.id for n in ret_val))

    def test_network_list_for_tenant(self):
        self._test_network_list_for_tenant(
            include_external=False, filter_params=None,
            should_called=['non_shared', 'shared'])

    def test_network_list_for_tenant_with_external(self):
        self._test_network_list_for_tenant(
            include_external=True, filter_params=None,
            should_called=['non_shared', 'shared', 'external'])

    def test_network_list_for_tenant_with_filters_shared_false_wo_incext(self):
        self._test_network_list_for_tenant(
            include_external=False, filter_params={'shared': True},
            should_called=['shared'])

    def test_network_list_for_tenant_with_filters_shared_true_w_incext(self):
        self._test_network_list_for_tenant(
            include_external=True, filter_params={'shared': True},
            should_called=['shared', 'external'])

    def test_network_list_for_tenant_with_filters_ext_false_wo_incext(self):
        self._test_network_list_for_tenant(
            include_external=False, filter_params={'router:external': False},
            should_called=['non_shared', 'shared'])

    def test_network_list_for_tenant_with_filters_ext_true_wo_incext(self):
        self._test_network_list_for_tenant(
            include_external=False, filter_params={'router:external': True},
            should_called=['non_shared', 'shared'])

    def test_network_list_for_tenant_with_filters_ext_false_w_incext(self):
        self._test_network_list_for_tenant(
            include_external=True, filter_params={'router:external': False},
            should_called=['non_shared', 'shared'])

    def test_network_list_for_tenant_with_filters_ext_true_w_incext(self):
        self._test_network_list_for_tenant(
            include_external=True, filter_params={'router:external': True},
            should_called=['non_shared', 'shared', 'external'])

    def test_network_list_for_tenant_with_filters_both_shared_ext(self):
        # To check 'shared' filter is specified in network_list
        # to look up external networks.
        self._test_network_list_for_tenant(
            include_external=True,
            filter_params={'router:external': True, 'shared': True},
            should_called=['shared', 'external'])

    def test_network_list_for_tenant_with_other_filters(self):
        # To check filter parameters other than shared and
        # router:external are passed as expected.
        self._test_network_list_for_tenant(
            include_external=True,
            filter_params={'router:external': True, 'shared': False,
                           'foo': 'bar'},
            should_called=['non_shared', 'external'])

    def test_network_get(self):
        network = {'network': self.api_networks.first()}
        subnet = {'subnet': self.api_subnets.first()}
        network_id = self.api_networks.first()['id']
        subnet_id = self.api_networks.first()['subnets'][0]

        neutronclient = self.stub_neutronclient()
        neutronclient.show_network(network_id).AndReturn(network)
        neutronclient.show_subnet(subnet_id).AndReturn(subnet)
        self.mox.ReplayAll()

        ret_val = api.neutron.network_get(self.request, network_id)
        self.assertIsInstance(ret_val, api.neutron.Network)

    def test_network_create(self):
        network = {'network': self.api_networks.first()}
        form_data = {'network': {'name': 'net1',
                                 'tenant_id': self.request.user.project_id}}
        neutronclient = self.stub_neutronclient()
        neutronclient.create_network(body=form_data).AndReturn(network)
        self.mox.ReplayAll()
        ret_val = api.neutron.network_create(self.request, name='net1')
        self.assertIsInstance(ret_val, api.neutron.Network)

    def test_network_update(self):
        network = {'network': self.api_networks.first()}
        network_id = self.api_networks.first()['id']

        neutronclient = self.stub_neutronclient()
        form_data = {'network': {'name': 'net1'}}
        neutronclient.update_network(network_id, body=form_data)\
            .AndReturn(network)
        self.mox.ReplayAll()

        ret_val = api.neutron.network_update(self.request, network_id,
                                             name='net1')
        self.assertIsInstance(ret_val, api.neutron.Network)

    def test_network_delete(self):
        network_id = self.api_networks.first()['id']

        neutronclient = self.stub_neutronclient()
        neutronclient.delete_network(network_id)
        self.mox.ReplayAll()

        api.neutron.network_delete(self.request, network_id)

    def test_get_network_ip_availability(self):
        network = {'network': self.api_networks.first()}
        mock_ip_availability = self.ip_availability.get()
        neutronclient = self.stub_neutronclient()
        neutronclient.show_network_ip_availability(network).\
            AndReturn(mock_ip_availability)

        self.mox.ReplayAll()
        ret_val = api.neutron.show_network_ip_availability(self.request,
                                                           network)

        self.assertIsInstance(ret_val, dict)

    def test_subnet_network_ip_availability(self):
        network = {'network': self.api_networks.first()}
        mock_ip_availability = self.ip_availability.get()
        neutronclient = self.stub_neutronclient()
        neutronclient.show_network_ip_availability(network).\
            AndReturn(mock_ip_availability)

        self.mox.ReplayAll()
        ip_availability = api.neutron. \
            show_network_ip_availability(self.request, network)
        availabilities = ip_availability.get("network_ip_availability",
                                             {})
        ret_val = availabilities.get("subnet_ip_availability", [])

        self.assertIsInstance(ret_val, list)

    def test_subnet_list(self):
        subnets = {'subnets': self.api_subnets.list()}

        neutronclient = self.stub_neutronclient()
        neutronclient.list_subnets().AndReturn(subnets)
        self.mox.ReplayAll()

        ret_val = api.neutron.subnet_list(self.request)
        for n in ret_val:
            self.assertIsInstance(n, api.neutron.Subnet)

    def test_subnet_get(self):
        subnet = {'subnet': self.api_subnets.first()}
        subnet_id = self.api_subnets.first()['id']

        neutronclient = self.stub_neutronclient()
        neutronclient.show_subnet(subnet_id).AndReturn(subnet)
        self.mox.ReplayAll()

        ret_val = api.neutron.subnet_get(self.request, subnet_id)
        self.assertIsInstance(ret_val, api.neutron.Subnet)

    def test_subnet_create(self):
        subnet_data = self.api_subnets.first()
        params = {'network_id': subnet_data['network_id'],
                  'tenant_id': subnet_data['tenant_id'],
                  'name': subnet_data['name'],
                  'cidr': subnet_data['cidr'],
                  'ip_version': subnet_data['ip_version'],
                  'gateway_ip': subnet_data['gateway_ip']}

        neutronclient = self.stub_neutronclient()
        neutronclient.create_subnet(body={'subnet': params})\
            .AndReturn({'subnet': subnet_data})
        self.mox.ReplayAll()

        ret_val = api.neutron.subnet_create(self.request, **params)
        self.assertIsInstance(ret_val, api.neutron.Subnet)

    def test_subnet_update(self):
        subnet_data = self.api_subnets.first()
        subnet_id = subnet_data['id']
        params = {'name': subnet_data['name'],
                  'gateway_ip': subnet_data['gateway_ip']}

        neutronclient = self.stub_neutronclient()
        neutronclient.update_subnet(subnet_id, body={'subnet': params})\
            .AndReturn({'subnet': subnet_data})
        self.mox.ReplayAll()

        ret_val = api.neutron.subnet_update(self.request, subnet_id, **params)
        self.assertIsInstance(ret_val, api.neutron.Subnet)

    def test_subnet_delete(self):
        subnet_id = self.api_subnets.first()['id']

        neutronclient = self.stub_neutronclient()
        neutronclient.delete_subnet(subnet_id)
        self.mox.ReplayAll()

        api.neutron.subnet_delete(self.request, subnet_id)

    def test_subnetpool_list(self):
        subnetpools = {'subnetpools': self.api_subnetpools.list()}

        neutronclient = self.stub_neutronclient()
        neutronclient.list_subnetpools().AndReturn(subnetpools)
        self.mox.ReplayAll()

        ret_val = api.neutron.subnetpool_list(self.request)
        for n in ret_val:
            self.assertIsInstance(n, api.neutron.SubnetPool)

    def test_subnetpool_get(self):
        subnetpool = {'subnetpool': self.api_subnetpools.first()}
        subnetpool_id = self.api_subnetpools.first()['id']

        neutronclient = self.stub_neutronclient()
        neutronclient.show_subnetpool(subnetpool_id).AndReturn(subnetpool)
        self.mox.ReplayAll()

        ret_val = api.neutron.subnetpool_get(self.request, subnetpool_id)
        self.assertIsInstance(ret_val, api.neutron.SubnetPool)

    def test_subnetpool_create(self):
        subnetpool_data = self.api_subnetpools.first()
        params = {'name': subnetpool_data['name'],
                  'prefixes': subnetpool_data['prefixes'],
                  'tenant_id': subnetpool_data['tenant_id']}

        neutronclient = self.stub_neutronclient()
        neutronclient.create_subnetpool(body={'subnetpool': params})\
            .AndReturn({'subnetpool': subnetpool_data})
        self.mox.ReplayAll()

        ret_val = api.neutron.subnetpool_create(self.request, **params)
        self.assertIsInstance(ret_val, api.neutron.SubnetPool)

    def test_subnetpool_update(self):
        subnetpool_data = self.api_subnetpools.first()
        subnetpool_id = subnetpool_data['id']
        params = {'name': subnetpool_data['name'],
                  'prefixes': subnetpool_data['prefixes']}

        neutronclient = self.stub_neutronclient()
        neutronclient.update_subnetpool(subnetpool_id, body={'subnetpool': params})\
            .AndReturn({'subnetpool': subnetpool_data})
        self.mox.ReplayAll()

        ret_val = api.neutron.subnetpool_update(self.request, subnetpool_id,
                                                **params)
        self.assertIsInstance(ret_val, api.neutron.SubnetPool)

    def test_subnetpool_delete(self):
        subnetpool_id = self.api_subnetpools.first()['id']

        neutronclient = self.stub_neutronclient()
        neutronclient.delete_subnetpool(subnetpool_id)
        self.mox.ReplayAll()

        api.neutron.subnetpool_delete(self.request, subnetpool_id)

    def test_port_list(self):
        ports = {'ports': self.api_ports.list()}

        neutronclient = self.stub_neutronclient()
        neutronclient.list_ports().AndReturn(ports)
        self.mox.ReplayAll()

        ret_val = api.neutron.port_list(self.request)
        for p in ret_val:
            self.assertIsInstance(p, api.neutron.Port)

    def test_port_get(self):
        port = {'port': self.api_ports.first()}
        port_id = self.api_ports.first()['id']

        neutronclient = self.stub_neutronclient()
        neutronclient.show_port(port_id).AndReturn(port)
        self.mox.ReplayAll()

        ret_val = api.neutron.port_get(self.request, port_id)
        self.assertIsInstance(ret_val, api.neutron.Port)

    def test_port_create(self):
        port = {'port': self.api_ports.first()}
        params = {'network_id': port['port']['network_id'],
                  'tenant_id': port['port']['tenant_id'],
                  'name': port['port']['name'],
                  'device_id': port['port']['device_id']}

        neutronclient = self.stub_neutronclient()
        neutronclient.create_port(body={'port': params}).AndReturn(port)
        self.mox.ReplayAll()
        ret_val = api.neutron.port_create(self.request, **params)
        self.assertIsInstance(ret_val, api.neutron.Port)
        self.assertEqual(api.neutron.Port(port['port']).id, ret_val.id)

    def test_port_update(self):
        port_data = self.api_ports.first()
        port_id = port_data['id']
        params = {'name': port_data['name'],
                  'device_id': port_data['device_id']}

        neutronclient = self.stub_neutronclient()
        neutronclient.update_port(port_id, body={'port': params})\
            .AndReturn({'port': port_data})
        self.mox.ReplayAll()

        ret_val = api.neutron.port_update(self.request, port_id, **params)
        self.assertIsInstance(ret_val, api.neutron.Port)
        self.assertEqual(api.neutron.Port(port_data).id, ret_val.id)

    def test_port_delete(self):
        port_id = self.api_ports.first()['id']

        neutronclient = self.stub_neutronclient()
        neutronclient.delete_port(port_id)
        self.mox.ReplayAll()

        api.neutron.port_delete(self.request, port_id)

    def test_trunk_list(self):
        trunks = {'trunks': self.api_trunks.list()}
        neutron_client = self.stub_neutronclient()
        neutron_client.list_trunks().AndReturn(trunks)
        self.mox.ReplayAll()

        ret_val = api.neutron.trunk_list(self.request)
        for t in ret_val:
            self.assertIsInstance(t, api.neutron.Trunk)

    def test_trunk_object(self):
        trunk = self.api_trunks.first().copy()
        obj = api.neutron.Trunk(trunk)
        self.assertEqual(0, obj.subport_count)
        trunk_dict = obj.to_dict()
        self.assertIsInstance(trunk_dict, dict)
        self.assertEqual(trunk['name'], trunk_dict['name_or_id'])
        self.assertEqual(0, trunk_dict['subport_count'])

        trunk['name'] = ''  # to test name_or_id
        trunk['sub_ports'] = [uuidutils.generate_uuid() for i in range(2)]
        obj = api.neutron.Trunk(trunk)
        self.assertEqual(2, obj.subport_count)
        trunk_dict = obj.to_dict()
        self.assertEqual(obj.name_or_id, trunk_dict['name_or_id'])
        self.assertEqual(2, trunk_dict['subport_count'])

    def test_router_list(self):
        routers = {'routers': self.api_routers.list()}

        neutronclient = self.stub_neutronclient()
        neutronclient.list_routers().AndReturn(routers)
        self.mox.ReplayAll()

        ret_val = api.neutron.router_list(self.request)
        for n in ret_val:
            self.assertIsInstance(n, api.neutron.Router)

    def test_router_get(self):
        router = {'router': self.api_routers.first()}
        router_id = self.api_routers.first()['id']

        neutronclient = self.stub_neutronclient()
        neutronclient.show_router(router_id).AndReturn(router)
        self.mox.ReplayAll()

        ret_val = api.neutron.router_get(self.request, router_id)
        self.assertIsInstance(ret_val, api.neutron.Router)

    def test_router_create(self):
        router = {'router': self.api_routers.first()}

        neutronclient = self.stub_neutronclient()
        form_data = {'router': {'name': 'router1',
                                'tenant_id': self.request.user.project_id}}
        neutronclient.create_router(body=form_data).AndReturn(router)
        self.mox.ReplayAll()

        ret_val = api.neutron.router_create(self.request, name='router1')
        self.assertIsInstance(ret_val, api.neutron.Router)

    def test_router_delete(self):
        router_id = self.api_routers.first()['id']

        neutronclient = self.stub_neutronclient()
        neutronclient.delete_router(router_id)
        self.mox.ReplayAll()

        api.neutron.router_delete(self.request, router_id)

    def test_router_add_interface(self):
        subnet_id = self.api_subnets.first()['id']
        router_id = self.api_routers.first()['id']

        neutronclient = self.stub_neutronclient()
        form_data = {'subnet_id': subnet_id}
        neutronclient.add_interface_router(
            router_id, form_data).AndReturn(None)
        self.mox.ReplayAll()

        api.neutron.router_add_interface(
            self.request, router_id, subnet_id=subnet_id)

    def test_router_remove_interface(self):
        router_id = self.api_routers.first()['id']
        fake_port = self.api_ports.first()['id']

        neutronclient = self.stub_neutronclient()
        neutronclient.remove_interface_router(
            router_id, {'port_id': fake_port})
        self.mox.ReplayAll()

        api.neutron.router_remove_interface(
            self.request, router_id, port_id=fake_port)

    def test_is_extension_supported(self):
        neutronclient = self.stub_neutronclient()
        neutronclient.list_extensions() \
            .AndReturn({'extensions': self.api_extensions.list()})
        self.mox.ReplayAll()

        self.assertTrue(
            api.neutron.is_extension_supported(self.request, 'quotas'))
        self.assertFalse(
            api.neutron.is_extension_supported(self.request, 'doesntexist'))
{'fraction_non_alphanumeric': 0.10802476513757081, 'fraction_numerical': 0.001051801209571391, 'mean_word_length': 2.7199004090343233, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 20, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7552749', 'n_tokens_mistral': 12842, 'n_tokens_neox': 12695, 'n_words': 1955}