How to use neutron method in tempest

Best Python code snippet using tempest_python

neutron.py

Source:neutron.py

1import logging2import speedling3from speedling import conf4from speedling import facility5from speedling import gitutils6from speedling import localsh7from speedling import tasks8from speedling import usrgrp9from speedling import util10LOG = logging.getLogger(__name__)11sp = 'sl-'12def do_ovs(cname):13    localsh.run('systemctl start openvswitch.service')14def task_ovs(self):15    facility.task_wants(speedling.tasks.task_cfg_etccfg_steps)16    # TODO add concept for implied service17    ovss = self.hosts_with_any_service({'neutron-openvswitch-agent', 'ovs'})18    self.call_do(ovss, do_ovs)19def ovs_pkgs(self):20    return {'srv-ovs\\switch'}21def task_net_config(self):22    # This is temporary here, normally it should do interface persistent config23    self.call_do(self.hosts_with_service('neutron-l3-agent'),24                 self.do_dummy_netconfig)25def task_neutron_steps(self):26    self.wait_for_components(self.sql)27    schema_node_candidate = self.hosts_with_service('neutron-server')28    schema_node = util.rand_pick(schema_node_candidate)29    sync_cmd = 'su -s /bin/sh -c "neutron-db-manage --config-file /etc/neutron/neutron.conf --config-file /etc/neutron/plugins/ml2/ml2_conf.ini upgrade head" neutron'30    self.call_do(schema_node, facility.do_retrycmd_after_content, c_args=(sync_cmd, ))31    facility.task_will_need(self.task_net_config)32    self.wait_for_components(self.messaging)33    facility.task_wants(self.task_net_config)34    q_srv = set(self.services.keys())35    self.call_do(self.hosts_with_any_service(q_srv), self.do_local_neutron_service_start)36    self.wait_for_components(self.messaging)37    self.wait_for_components(self.osclient)38    self.call_do(util.rand_pick(self.hosts_with_service('neutron-server')), self.do_dummy_public_net)39q_srv = {'neutron-server', 'neutron-openvswitch-agent', 'neutron-vpn-agent',40         'neutron-dhcp-agent', 'neutron-metadata-agent', 'neutron-l3-agent',41         'neutron-metering-agent', 'neutron-lbaasv2-agent'}42class NeutronML2OVS(facility.OpenStack):43    pass44class Neutron(facility.OpenStack):45    origin_repo = 'https://github.com/openstack/neutron.git'46    deploy_source = 'src'47    deploy_source_options = {'src', 'pkg'}48    component = 'neutron'49    services = {'neutron-server': {'deploy_mode': 'standalone',50                                   'unit_name': {'src': sp + 'q-svc',51                                                 'pkg': 'neutron-server'}},52                'neutron-metadata-agent': {'deploy_mode': 'standalone',53                                           'unit_name': {'src': sp + 'q-meta',54                                                         'pkg': 'neutron-metadata-agent'}},55                'neutron-l3-agent': {'deploy_mode': 'standalone',56                                     'unit_name': {'src': sp + 'q-l3',57                                                   'pkg': 'neutron-l3-agent'}},58                'neutron-metering-agent': {'deploy_mode': 'standalone',59                                           'unit_name': {'src': sp + 'q-metering',60                                                         'pkg': 'neutron-metering-agent'}},61                'neutron-vpn-agent': {'deploy_mode': 'standalone',62                                      'unit_name': {'src': sp + 'q-vpn',63                                                    'pkg': 'neutron-vpn-agent'}},64                'neutron-dhcp-agent': {'deploy_mode': 'standalone',65                                       'unit_name': {'src': sp + 'q-dhcp',66                                                     'pkg': 'neutron-dhcp-agent'}},67                'neutron-lbaasv2-agent': {'deploy_mode': 'standalone',68                                          'unit_name': {'src': sp + 'q-lbaas',69                                                        'pkg': 'neutron-vpn-agent'}},70                'neutron-openvswitch-agent': {'deploy_mode': 'standalone',71                                              'unit_name': {'src': sp + 'q-ovs',  # q-agt72                                                            'pkg': 'neutron-openvswitch-agent'}}}73    def __init__(self, *args, **kwargs):74        super(Neutron, self).__init__(*args, **kwargs)75        self.final_task = self.bound_to_instance(task_neutron_steps)76        self.bound_to_instance(task_net_config)77        self.peer_info = {}78        self.sql = self.dependencies["sql"]79        self.haproxy = self.dependencies["loadbalancer"]80        self.keystone = self.dependencies["keystone"]81        self.messaging = self.dependencies["messaging"]82        self.osclient = self.dependencies["osclient"]83    def find_nova_comp_shared(self):84        return [c for c in self.consumers.keys() if c.short_name == 'nova'] + [self]85    def etc_neutron_conf_d_common_agent_conf(self): return {'agent': {'root_helper': "sudo /usr/local/bin/neutron-rootwrap /etc/neutron/rootwrap.conf",86                                                                      'root_helper_daemon': "sudo /usr/local/bin/neutron-rootwrap-daemon /etc/neutron/rootwrap.conf"}}87    def etc_neutron_metadata_agent_ini(self):88        ivip = conf.get_vip('internal')['domain_name']89        return {'DEFAULT': {'nova_metadata_ip':  ivip,90                            'metadata_proxy_shared_secret':91                            util.get_keymgr()(self.find_nova_comp_shared(),92                                              'neutron_nova_metadata')}}93    def etc_neutron_neutron_conf(self):94        gconf = conf.get_global_config()95        service_flags = gconf['global_service_flags']96        service_pulugins = ['neutron.services.l3_router.l3_router_plugin.L3RouterPlugin']97        if 'neutron-fwaas' in service_flags:98            service_pulugins.append('neutron_fwaas.services.firewall.fwaas_plugin.FirewallPlugin')99        if 'neutron-vpn-agent' in service_flags:100            service_pulugins.append('neutron_vpnaas.services.vpn.plugin.VPNDriverPlugin')101        if 'neutron-lbaasv2-agent' in service_flags:102            service_pulugins.append('neutron_lbaas.services.loadbalancer.plugin.LoadBalancerPluginv2')103        if 'neutron-metering-agent' in service_flags:104            service_pulugins.append('neutron.services.metering.metering_plugin.MeteringPlugin')105        return {'DEFAULT': {'debug':  'True',106                            'core_plugin': 'ml2',107                            'service_plugins': ','.join(service_pulugins),108                            'state_path': '/var/lib/neutron',109                            'allow_overlapping_ips': 'True',110                            'transport_url': self.messaging.transport_url()},111                'database': {'connection': self.sql.db_url('neutron')},112                'keystone_authtoken': self.keystone.authtoken_section('neutron'),113                'nova': self.keystone.authtoken_section('nova_for_neutron'),114                'oslo_concurrency': {'lock_path': '$state_path/lock'}}115    def etc_neutron_plugins_ml2_ml2_conf_ini(self): return {116        'ml2': {'tenant_network_types': 'vxlan',  # TODO: switch to geneve}117                'mechanism_drivers': 'openvswitch,linuxbridge',118                'extension_drivers': 'port_security'},119        'ml2_type_vxlan': {'vni_ranges': '1001:4001'}120    }121    # vpnaas, consider other drivers122    def etc_neutron_vpn_agent_ini(self): return {123        'vpnagent': {'vpn_device_driver': 'neutron_vpnaas.services.vpn.' +124                                          'device_drivers.fedora_strongswan_ipsec.FedoraStrongSwanDriver'}}125    def etc_neutron_neutron_vpnaas_conf(self): return {126        'service_providers': {'service_provider': 'VPN:openswan:neutron_vpnaas.services.vpn.service_drivers.ipsec.IPsecVPNDriver:default'}}127    # fwaas added, why not.. (TEST ONLY)128    def etc_neutron_fwaas_driver_ini(self): return {129        'fwaas': {'driver': 'neutron_fwaas.services.firewall.drivers.linux' +130                            '.iptables_fwaas.IptablesFwaasDriver'}}131    def etccfg_content(self):132        super(Neutron, self).etccfg_content()133        gconf = conf.get_global_config()134        global_service_union = gconf['global_service_flags']135        usrgrp.group('neutron', 996)136        usrgrp.user('neutron', 'neutron')137        util.base_service_dirs('neutron')138        self.file_path('/etc/neutron/conf.d',139                       owner='neutron', group='neutron')140        self.file_path('/etc/neutron/conf.d/common',141                       owner='neutron', group='neutron')142        self.file_ini('/etc/neutron/conf.d/common/agent.conf',143                      self.etc_neutron_conf_d_common_agent_conf(),144                      owner='neutron', group='neutron')145        neutron_git_dir = gitutils.component_git_dir(self)146        # consider alternate data paths147        # var/lib/neutron/dhcp needs to be reachable by the dnsmasq user148        self.file_path('/var/lib/neutron',149                       owner='neutron', group='neutron',150                       mode=0o755)151        self.file_path('/var/lib/neutron/lock',152                       owner='neutron', group='neutron')153        self.file_path('/etc/neutron/plugins',154                       owner='neutron', group='neutron')155        self.file_path('/etc/neutron/plugins/ml2',156                       owner='neutron', group='neutron')157        self.file_ini('/etc/neutron/neutron.conf', self.etc_neutron_neutron_conf(),158                      owner='neutron', group='neutron')159        self.file_sym_link('/etc/neutron/plugin.ini',160                           '/etc/neutron/plugins/ml2/ml2_conf.ini')161        # move to common ?162        self.file_ini('/etc/neutron/plugins/ml2/ml2_conf.ini',163                      self.etc_neutron_plugins_ml2_ml2_conf_ini(),164                      owner='neutron', group='neutron')165        services = self.filter_node_enabled_services(self.services.keys())166        if self.deploy_source == 'src':167            if services.intersection(q_srv - {'neutron-server'}):168                self.file_plain('/etc/sudoers.d/neutron', """Defaults:neutron !requiretty169neutron ALL = (root) NOPASSWD: /usr/bin/neutron-rootwrap /etc/neutron/rootwrap.conf *170neutron ALL = (root) NOPASSWD: /usr/bin/neutron-rootwrap-daemon /etc/neutron/rootwrap.conf171neutron ALL = (root) NOPASSWD: /usr/local/bin/neutron-rootwrap /etc/neutron/rootwrap.conf *172neutron ALL = (root) NOPASSWD: /usr/local/bin/neutron-rootwrap-daemon /etc/neutron/rootwrap.conf173""")174                self.file_path('/etc/neutron/rootwrap.d',175                               owner='root')176                # TODO: exclude stuff based on config177                for filter_file in ['debug.filters', 'dibbler.filters', 'ipset-firewall.filters',178                                    'l3.filters', 'netns-cleanup.filters', 'privsep.filters',179                                    'dhcp.filters', 'ebtables.filters', 'iptables-firewall.filters',180                                    'linuxbridge-plugin.filters', 'openvswitch-plugin.filters']:181                    self.file_install('/etc/neutron/rootwrap.d/' + filter_file,182                                      '/'.join((neutron_git_dir,183                                                'etc/neutron/rootwrap.d', filter_file)),184                                      mode=0o444)185            self.file_install('/etc/neutron/rootwrap.conf',186                              '/'.join((neutron_git_dir,187                                        'etc/rootwrap.conf')),188                              mode=0o444)189            self.file_install('/etc/neutron/api-paste.ini',190                              '/'.join((neutron_git_dir,191                                        'etc/api-paste.ini')),192                              mode=0o644,193                              owner='neutron', group='neutron')194            c_srv = self.services195            util.unit_file(c_srv['neutron-server']['unit_name']['src'],196                           '/usr/local/bin/neutron-server --config-file /etc/neutron/neutron.conf --config-dir /etc/neutron/conf.d/common --config-file /etc/neutron/plugin.ini',197                           'neutron')198            util.unit_file(c_srv['neutron-metadata-agent']['unit_name']['src'],199                           '/usr/local/bin/neutron-metadata-agent --config-file /etc/neutron/neutron.conf --config-dir /etc/neutron/conf.d/common --config-file /etc/neutron/metadata_agent.ini',200                           'neutron')201            util.unit_file(c_srv['neutron-l3-agent']['unit_name']['src'],202                           '/usr/local/bin/neutron-l3-agent --config-file /etc/neutron/neutron.conf --config-dir /etc/neutron/conf.d/common --config-file /etc/neutron/l3_agent.ini',203                           'neutron')204            util.unit_file(c_srv['neutron-metering-agent']['unit_name']['src'],205                           '/usr/local/bin/neutron-metering-agent --config-file /etc/neutron/neutron.conf --config-dir /etc/neutron/conf.d/common --config-file /etc/neutron/metering_agent.ini',206                           'neutron')207            util.unit_file(c_srv['neutron-vpn-agent']['unit_name']['src'],208                           '/usr/local/bin/neutron-vpn-agent --config-file /etc/neutron/neutron.conf --config-dir /etc/neutron/conf.d/common --config-file /etc/neutron/l3_agent.ini --config-file /etc/neutron/vpn_agent.ini',209                           'neutron')210            util.unit_file(c_srv['neutron-dhcp-agent']['unit_name']['src'],211                           '/usr/local/bin/neutron-dhcp-agent --config-file /etc/neutron/neutron.conf --config-dir /etc/neutron/conf.d/common --config-file /etc/neutron/dhcp_agent.ini',212                           'neutron')213            util.unit_file(c_srv['neutron-lbaasv2-agent']['unit_name']['src'],214                           '/usr/local/bin/neutron-lbaasv2-agent --config-file /etc/neutron/neutron.conf --config-dir /etc/neutron/conf.d/common --config-file /etc/neutron/lbaas_agent.ini',215                           'neutron')216            if util.get_distro()['family'] != 'debian':217                osrv = 'openvswitch.service'218            else:219                osrv = 'openvswitch-switch.service'220            util.unit_file(c_srv['neutron-openvswitch-agent']['unit_name']['src'],221                           '/usr/local/bin/neutron-openvswitch-agent --config-file /etc/neutron/neutron.conf --config-dir /etc/neutron/conf.d/common --config-file /etc/neutron/plugins/ml2/openvswitch_agent.ini',222                           'neutron', requires=osrv, restart='on-failure')223        if 'neutron-metadata-agent' in services:224            self.file_ini('/etc/neutron/metadata_agent.ini',225                          self.etc_neutron_metadata_agent_ini(),226                          owner='neutron', group='neutron')227        if 'neutron-vpn-agent' in services or 'neutron-l3-agent' in services:228            self.file_ini('/etc/neutron/l3_agent.ini', {229                'DEFAULT': {'interface_driver': 'openvswitch',230                            'debug': True}231            }, owner='neutron', group='neutron')232        if 'neutron-metering-agent' in services:233            self.file_ini('/etc/neutron/metering_agent.ini', {234                'DEFAULT': {'interface_driver': 'openvswitch',235                            'debug': True}236            }, owner='neutron', group='neutron')237        if 'neutron-vpn-agent' in services:238            self.file_ini('/etc/neutron/vpn_agent.ini',239                          self.etc_neutron_vpn_agent_ini(),240                          owner='neutron', group='neutron')241        if 'neutron-dhcp-agent' in services:242            self.file_ini('/etc/neutron/dhcp_agent.ini', {243                'DEFAULT': {'interface_driver': 'openvswitch',244                            'dnsmasq_local_resolv': True,245                            'debug': True}246            }, owner='neutron', group='neutron')247        if 'neutron-lbaasv2-agent' in services:248            self.file_ini('/etc/neutron/lbaas_agent.ini', {249                'DEFAULT': {'interface_driver': 'openvswitch',250                            'debug': True}},251                          owner='neutron', group='neutron')252        if 'neutron-openvswitch-agent' in services:253            tunnel_ip = self.get_addr_for(self.get_this_inv(), 'tunneling',254                                          service=self.services['neutron-openvswitch-agent'],255                                          net_attr='tunneling_network')256            ovs = {'local_ip': tunnel_ip}257            if 'neutron-l3-agent' in services:258                ovs['bridge_mappings'] = 'extnet:br-ex'259            self.file_ini('/etc/neutron/plugins/ml2/openvswitch_agent.ini', {260                'securitygroup': {'firewall_driver': 'iptables_hybrid'},261                'ovs': ovs,262                'agent': {'tunnel_types': 'vxlan'}},263                owner='neutron', group='neutron')264        # the inv version is not transfered, let it be part of the global config265        #    global_service_union = self.get_enabled_services()266        # NOTE: check these fwass,lbaas, vpaans conditions,267        # we might want to update them even if they not present268        if ('neutron-lbaasv2-agent' in services or ('neutron-lbaasv2-agent' in global_service_union and269                                                    'neutron-server' in services)):270            self.file_ini('/etc/neutron/neutron_lbaas.conf', {271                'service_providers': {'service_provider':272                                      'LOADBALANCERV2:Haproxy:' +273                                      'neutron_lbaas.drivers.haproxy.plugin_driver.HaproxyOnHostPluginDriver' +274                                      ':default'}}, owner='neutron', group='neutron')275        if ('neutron-vpn-agent' in services or ('neutron-vpn-agent' in global_service_union and276                                                'neutron-server' in services)):277            self.file_ini('/etc/neutron/neutron_vpnaas.conf',278                          self.etc_neutron_neutron_vpnaas_conf(),279                          owner='neutron', group='neutron')280        if 'neutron-fwaas' in global_service_union:281            self.file_ini('/etc/neutron/fwaas_driver.ini',282                          self.etc_neutron_fwaas_driver_ini(),283                          owner='neutron', group='neutron')284    def do_dummy_public_net(cname):285        # guest net hack286        # 192.0.2.1 expected to be configured on an interface287        localsh.run(util.userrc_script('admin') + """288        (289        retry=30290        while ! neutron net-create public --router:external=True --is-default=True --provider:network_type flat --provider:physical_network extnet ; do291           ((retry--))292           if [[ retry == 0 ]]; then293              break;294           fi295        done296        FLOATING_IP_CIDR=${FLOATING_IP_CIDR:-"192.0.2.0/24"}297        FLOATING_IP_START=${FLOATING_IP_START:-"192.0.2.32"}298        FLOATING_IP_END=${FLOATING_IP_END:-"192.0.2.196"}299        EXTERNAL_NETWORK_GATEWAY=${EXTERNAL_NETWORK_GATEWAY:-"192.0.2.1"}300        neutron subnet-create --name ext-subnet --allocation-pool start=$FLOATING_IP_START,end=$FLOATING_IP_END --disable-dhcp --gateway $EXTERNAL_NETWORK_GATEWAY public $FLOATING_IP_CIDR301        # for auto allocation test302        openstack subnet pool create --share --default --pool-prefix 192.0.3.0/24 --default-prefix-length 26  shared-default303        openstack subnet pool create --share --default --pool-prefix 2001:db8:8000::/48 --default-prefix-length 64 default-v6304        )""")305    def do_dummy_netconfig(cname):306        if util.get_distro()['family'] != 'debian':307            osrv = 'openvswitch.service'308        else:309            osrv = 'openvswitch-switch.service'310        localsh.run('systemctl start ' + osrv)311        # TODO switch to os-net-config312        # wait (no --no-wait)313        localsh.run('ovs-vsctl --may-exist add-br br-ex')314        # add ip to external bridge instead of adding a phyisical if315        localsh.run("""316       ifconfig br-ex 192.0.2.1317       ip link set br-ex up318       ROUTE_TO_INTERNET=$(ip route get 8.8.8.8)319       OBOUND_DEV=$(echo ${ROUTE_TO_INTERNET#*dev} | awk '{print $1}')320       iptables -t nat -A POSTROUTING -o $OBOUND_DEV -j MASQUERADE321       tee /proc/sys/net/ipv4/ip_forward <<<1 >/dev/null322       """)323    def do_local_neutron_service_start(cname):324        self = facility.get_component(cname)325        tasks.local_os_service_start_by_component(self)326    def get_node_packages(self):327        pkgs = super(Neutron, self).get_node_packages()328        pkgs.update({'acl', 'dnsmasq', 'dnsmasq-utils', 'ebtables', 'haproxy',329                     'iptables', 'util-cli\\iputils', 'lib-dev\\mariadb', 'radvd', 'sqlite',330                     'sudo', 'keepalived',331                     'ipset', 'srv-ovs\\switch'})332        # move ovs to ovs333        if self.deploy_source == 'pkg':334            pkgs.update({'openstack-neutron'})335        return pkgs336    def compose(self):337        super(Neutron, self).compose()338        # it can consider the full inventory and config to influnce facility registered339        # resources340        url_base = "http://" + conf.get_vip('public')['domain_name']341        dr = conf.get_default_region()342        self.keystone.register_endpoint_tri(region=dr,343                                            name='neutron',344                                            etype='network',345                                            description='OpenStack Network Service',346                                            url_base=url_base + ':9696/')347        self.keystone.register_service_admin_user('neutron')348        self.keystone.register_service_admin_user('nova_for_neutron')349        neutrons = self.hosts_with_any_service(set(self.services.keys()))350        self.messaging.populate_peer(neutrons)351        self.sql.register_user_with_schemas('neutron', ['neutron'])352        self.sql.populate_peer(neutrons, ['client'])  # TODO: maybe not all node needs it353        secret_service = self.find_nova_comp_shared()354        util.bless_with_principal(neutrons,355                                  [(self.keystone, 'neutron@default'),356                                   (self.keystone, 'nova_for_neutron@default'),357                                   (self.sql, 'neutron'),358                                   (secret_service, 'neutron_nova_metadata'),359                                   (self.messaging.name, 'openstack')])360def register(self, ):361    sp = conf.get_service_prefix()362    ovs = {'component': 'openvswitch',363           'deploy_source': 'pkg',364           'deploy_source_options': {'pkg'},365           'services': {'openvswitch': {'deploy_mode': 'standalone',366                                        'unit_name': {'src': sp + 'ovs',367                                                      'pkg': 'openvswitch'}}},368           'pkg_deps': ovs_pkgs,369           'goal': task_ovs}...

neutron_rest_tests.py

Source:neutron_rest_tests.py

1#2#    (c) Copyright 2015 Hewlett-Packard Development Company, L.P.3#4# Licensed under the Apache License, Version 2.0 (the "License");5# you may not use this file except in compliance with the License.6# You may obtain a copy of the License at7#8#    http://www.apache.org/licenses/LICENSE-2.09#10# Unless required by applicable law or agreed to in writing, software11# distributed under the License is distributed on an "AS IS" BASIS,12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.13# See the License for the specific language governing permissions and14# limitations under the License.15from django.http import request as django_request16import mock17from openstack_dashboard import api18from openstack_dashboard.api import base19from openstack_dashboard.api.rest import neutron20from openstack_dashboard.test import helpers as test21from openstack_dashboard.test.test_data import neutron_data22from openstack_dashboard.test.test_data.utils import TestData23TEST = TestData(neutron_data.data)24class NeutronNetworksTestCase(test.TestCase):25    def setUp(self):26        super(NeutronNetworksTestCase, self).setUp()27        self._networks = [test.mock_factory(n)28                          for n in TEST.api_networks.list()]29    @mock.patch.object(neutron.api, 'neutron')30    def test_get_list_for_tenant(self, client):31        request = self.mock_rest_request()32        networks = self._networks33        client.network_list_for_tenant.return_value = networks34        response = neutron.Networks().get(request)35        self.assertStatusCode(response, 200)36        self.assertItemsCollectionEqual(response, TEST.api_networks.list())37        client.network_list_for_tenant.assert_called_once_with(38            request, request.user.tenant_id)39    @mock.patch.object(neutron.api, 'neutron')40    def test_create(self, client):41        self._test_create(42            '{"name": "mynetwork"}',43            {'name': 'mynetwork'}44        )45    @mock.patch.object(neutron.api, 'neutron')46    def test_create_with_bogus_param(self, client):47        self._test_create(48            '{"name": "mynetwork","bilbo":"baggins"}',49            {'name': 'mynetwork'}50        )51    @mock.patch.object(neutron.api, 'neutron')52    def _test_create(self, supplied_body, expected_call, client):53        request = self.mock_rest_request(body=supplied_body)54        client.network_create.return_value = self._networks[0]55        response = neutron.Networks().post(request)56        self.assertStatusCode(response, 201)57        self.assertEqual(response['location'],58                         '/api/neutron/networks/'59                         + str(TEST.api_networks.first().get("id")))60        self.assertEqual(response.json, TEST.api_networks.first())61    #62    # Services63    #64    @test.create_stubs({api.base: ('is_service_enabled',)})65    @test.create_stubs({api.neutron: ('is_extension_supported',)})66    @mock.patch.object(neutron.api, 'neutron')67    def test_services_get(self, client):68        params = django_request.QueryDict('network_id=the_network')69        request = self.mock_rest_request(GET=params)70        api.base.is_service_enabled(request, 'network').AndReturn(True)71        api.neutron.is_extension_supported(request, 'agent').AndReturn(True)72        client.agent_list.return_value = [73            mock.Mock(**{'to_dict.return_value': {'id': '1'}}),74            mock.Mock(**{'to_dict.return_value': {'id': '2'}})75        ]76        self.mox.ReplayAll()77        response = neutron.Services().get(request)78        self.assertStatusCode(response, 200)79        client.agent_list.assert_called_once_with(80            request, network_id='the_network')81        self.assertEqual(response.content.decode('utf-8'),82                         '{"items": [{"id": "1"}, {"id": "2"}]}')83    @test.create_stubs({api.base: ('is_service_enabled',)})84    def test_services_get_disabled(self):85        request = self.mock_rest_request(86            GET={"network_id": self._networks[0].id})87        api.base.is_service_enabled(request, 'network').AndReturn(False)88        self.mox.ReplayAll()89        response = neutron.Services().get(request)90        self.assertStatusCode(response, 501)91class NeutronSubnetsTestCase(test.TestCase):92    def setUp(self):93        super(NeutronSubnetsTestCase, self).setUp()94        self._networks = [test.mock_factory(n)95                          for n in TEST.api_networks.list()]96        self._subnets = [test.mock_factory(n)97                         for n in TEST.api_subnets.list()]98    @mock.patch.object(neutron.api, 'neutron')99    def test_get(self, client):100        params = django_request.QueryDict('network_id=%s' %101                                          self._networks[0].id)102        request = self.mock_rest_request(GET=params)103        client.subnet_list.return_value = [self._subnets[0]]104        response = neutron.Subnets().get(request)105        self.assertStatusCode(response, 200)106        client.subnet_list.assert_called_once_with(107            request, network_id=TEST.api_networks.first().get("id"))108    @mock.patch.object(neutron.api, 'neutron')109    def test_create(self, client):110        request = self.mock_rest_request(111            body='{"network_id": "%s",'112                 ' "ip_version": "4",'113                 ' "cidr": "192.168.199.0/24"}' % self._networks[0].id)114        client.subnet_create.return_value = self._subnets[0]115        response = neutron.Subnets().post(request)116        self.assertStatusCode(response, 201)117        self.assertEqual(response['location'],118                         '/api/neutron/subnets/' +119                         str(TEST.api_subnets.first().get("id")))120        self.assertEqual(response.json, TEST.api_subnets.first())121class NeutronPortsTestCase(test.TestCase):122    def setUp(self):123        super(NeutronPortsTestCase, self).setUp()124        self._networks = [test.mock_factory(n)125                          for n in TEST.api_networks.list()]126        self._ports = [test.mock_factory(n)127                       for n in TEST.api_ports.list()]128    @mock.patch.object(neutron.api, 'neutron')129    def test_get(self, client):130        params = django_request.QueryDict('network_id=%s' %131                                          self._networks[0].id)132        request = self.mock_rest_request(GET=params)133        client.port_list.return_value = [self._ports[0]]134        response = neutron.Ports().get(request)135        self.assertStatusCode(response, 200)136        client.port_list.assert_called_once_with(137            request, network_id=TEST.api_networks.first().get("id"))138class NeutronTrunkTestCase(test.TestCase):139    @mock.patch.object(neutron.api, 'neutron')140    def test_trunk_delete(self, client):141        request = self.mock_rest_request()142        neutron.Trunk().delete(request, 1)143        client.trunk_delete.assert_called_once_with(request, 1)144    @mock.patch.object(neutron.api, 'neutron')145    def test_trunk_get(self, client):146        trunk_id = TEST.api_trunks.first().get("id")147        request = self.mock_rest_request(GET={"trunk_id": trunk_id})148        client.trunk_show.return_value = self.trunks.first()149        response = neutron.Trunk().get(request, trunk_id=trunk_id)150        self.assertStatusCode(response, 200)151        client.trunk_show.assert_called_once_with(152            request, trunk_id)153class NeutronTrunksTestCase(test.TestCase):154    @mock.patch.object(neutron.api, 'neutron')155    def test_trunks_get(self, client):156        request = self.mock_rest_request(GET=django_request.QueryDict())157        client.trunk_list.return_value = self.trunks.list()158        response = neutron.Trunks().get(request)159        self.assertStatusCode(response, 200)160        self.assertItemsCollectionEqual(161            response,162            [t.to_dict() for t in self.trunks.list()])163class NeutronExtensionsTestCase(test.TestCase):164    def setUp(self):165        super(NeutronExtensionsTestCase, self).setUp()166        self._extensions = [n for n in TEST.api_extensions.list()]167    @mock.patch.object(neutron.api, 'neutron')168    def test_list_extensions(self, nc):169        request = self.mock_rest_request(**{'GET': {}})170        nc.list_extensions.return_value = self._extensions171        response = neutron.Extensions().get(request)172        self.assertStatusCode(response, 200)173        self.assertItemsCollectionEqual(response, TEST.api_extensions.list())174        nc.list_extensions.assert_called_once_with(request)175class NeutronDefaultQuotasTestCase(test.TestCase):176    @test.create_stubs({base: ('is_service_enabled',)})177    @mock.patch.object(neutron.api, 'neutron')178    def test_quotas_sets_defaults_get_when_service_is_enabled(self, client):179        filters = {'user': {'tenant_id': 'tenant'}}180        request = self.mock_rest_request(**{'GET': dict(filters)})181        base.is_service_enabled(request, 'network').AndReturn(True)182        client.tenant_quota_get.return_value = [183            base.Quota("network", 100),184            base.Quota("q2", 101)]185        self.mox.ReplayAll()186        response = neutron.DefaultQuotaSets().get(request)187        self.assertStatusCode(response, 200)188        self.assertItemsCollectionEqual(response, [189            {'limit': 100, 'display_name': 'Networks', 'name': 'network'},190            {'limit': 101, 'display_name': 'Q2', 'name': 'q2'}])191        client.tenant_quota_get.assert_called_once_with(192            request,193            request.user.tenant_id)194    @test.create_stubs({neutron.api.base: ('is_service_enabled',)})195    @mock.patch.object(neutron.api, 'neutron')196    def test_quota_sets_defaults_get_when_service_is_disabled(self, client):197        filters = {'user': {'tenant_id': 'tenant'}}198        request = self.mock_rest_request(**{'GET': dict(filters)})199        base.is_service_enabled(request, 'network').AndReturn(False)200        self.mox.ReplayAll()201        response = neutron.DefaultQuotaSets().get(request)202        self.assertStatusCode(response, 501)203        self.assertEqual(response.content.decode('utf-8'),204                         '"Service Neutron is disabled."')205        client.tenant_quota_get.assert_not_called()206class NeutronQuotaSetsTestCase(test.TestCase):207    def setUp(self):208        super(NeutronQuotaSetsTestCase, self).setUp()209        quota_set = self.neutron_quotas.list()[0]210        self._quota_data = {}211        for quota in quota_set:212            self._quota_data[quota.name] = quota.limit213    @mock.patch.object(neutron, 'quotas')214    @mock.patch.object(neutron.api, 'neutron')215    @mock.patch.object(neutron.api, 'base')216    def test_quotas_sets_patch(self, bc, nc, qc):217        request = self.mock_rest_request(body='''218            {"network": "5", "subnet": "5", "port": "50",219             "router": "5", "floatingip": "50",220             "security_group": "5", "security_group_rule": "50",221             "volumes": "5", "cores": "50"}222        ''')223        qc.get_disabled_quotas.return_value = []224        qc.NEUTRON_QUOTA_FIELDS = {n for n in self._quota_data}225        bc.is_service_enabled.return_value = True226        nc.is_extension_supported.return_value = True227        response = neutron.QuotasSets().patch(request, 'spam123')228        self.assertStatusCode(response, 204)229        self.assertEqual(response.content.decode('utf-8'), '')230        nc.tenant_quota_update.assert_called_once_with(231            request, 'spam123', network='5',232            subnet='5', port='50', router='5',233            floatingip='50', security_group='5',234            security_group_rule='50')235    @mock.patch.object(neutron, 'quotas')236    @mock.patch.object(neutron.api, 'neutron')237    @mock.patch.object(neutron.api, 'base')238    def test_quotas_sets_patch_when_service_is_disabled(self, bc, nc, qc):239        request = self.mock_rest_request(body='''240            {"network": "5", "subnet": "5", "port": "50",241             "router": "5", "floatingip": "50",242             "security_group": "5", "security_group_rule": "50",243             "volumes": "5", "cores": "50"}244        ''')245        qc.get_disabled_quotas.return_value = []246        qc.NEUTRON_QUOTA_FIELDS = {n for n in self._quota_data}247        bc.is_service_enabled.return_value = False248        response = neutron.QuotasSets().patch(request, 'spam123')249        message = \250            '"Service Neutron is disabled or quotas extension not available."'251        self.assertStatusCode(response, 501)252        self.assertEqual(response.content.decode('utf-8'), message)253        nc.tenant_quota_update.assert_not_called()254def mock_obj_to_dict(r):255    return mock.Mock(**{'to_dict.return_value': r})256def mock_factory(r):257    """mocks all the attributes as well as the to_dict """258    mocked = mock_obj_to_dict(r)259    mocked.configure_mock(**r)...

test_neutron.py

Source:test_neutron.py

...37        self.assertRaises(subprocess.CalledProcessError,38                          self.neutron,39                          'this-does-not-exist')40    def test_neutron_net_list(self):41        self.neutron('net-list')42    def test_neutron_ext_list(self):43        ext = self.parser.listing(self.neutron('ext-list'))44        self.assertTableStruct(ext, ['alias', 'name'])45    def test_neutron_dhcp_agent_list_hosting_net(self):46        self.neutron('dhcp-agent-list-hosting-net',47                     params=CONF.compute.fixed_network_name)48    def test_neutron_agent_list(self):49        agents = self.parser.listing(self.neutron('agent-list'))50        field_names = ['id', 'agent_type', 'host', 'alive', 'admin_state_up']51        self.assertTableStruct(agents, field_names)52    def test_neutron_floatingip_list(self):53        self.neutron('floatingip-list')54    @test.skip_because(bug="1240694")55    def test_neutron_meter_label_list(self):56        self.neutron('meter-label-list')57    @test.skip_because(bug="1240694")58    def test_neutron_meter_label_rule_list(self):59        self.neutron('meter-label-rule-list')60    def _test_neutron_lbaas_command(self, command):61        try:62            self.neutron(command)63        except tempest.cli.CommandFailed as e:64            if '404 Not Found' not in e.stderr:65                self.fail('%s: Unexpected failure.' % command)66    def test_neutron_lb_healthmonitor_list(self):67        self._test_neutron_lbaas_command('lb-healthmonitor-list')68    def test_neutron_lb_member_list(self):69        self._test_neutron_lbaas_command('lb-member-list')70    def test_neutron_lb_pool_list(self):71        self._test_neutron_lbaas_command('lb-pool-list')72    def test_neutron_lb_vip_list(self):73        self._test_neutron_lbaas_command('lb-vip-list')74    def test_neutron_net_external_list(self):75        self.neutron('net-external-list')76    def test_neutron_port_list(self):77        self.neutron('port-list')78    def test_neutron_quota_list(self):79        self.neutron('quota-list')80    def test_neutron_router_list(self):81        self.neutron('router-list')82    def test_neutron_security_group_list(self):83        security_grp = self.parser.listing(self.neutron('security-group-list'))84        self.assertTableStruct(security_grp, ['id', 'name', 'description'])85    def test_neutron_security_group_rule_list(self):86        self.neutron('security-group-rule-list')87    def test_neutron_subnet_list(self):88        self.neutron('subnet-list')89    def test_neutron_help(self):90        help_text = self.neutron('help')91        lines = help_text.split('\n')92        self.assertFirstLineStartsWith(lines, 'usage: neutron')93        commands = []94        cmds_start = lines.index('Commands for API v2.0:')95        command_pattern = re.compile('^ {2}([a-z0-9\-\_]+)')96        for line in lines[cmds_start:]:97            match = command_pattern.match(line)98            if match:99                commands.append(match.group(1))100        commands = set(commands)101        wanted_commands = set(('net-create', 'subnet-list', 'port-delete',102                               'router-show', 'agent-update', 'help'))103        self.assertFalse(wanted_commands - commands)104     # Optional arguments:105    def test_neutron_version(self):106        self.neutron('', flags='--version')107    def test_neutron_debug_net_list(self):108        self.neutron('net-list', flags='--debug')109    def test_neutron_quiet_net_list(self):...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.