KaneAI - World’s First E2E Software Testing Agent.Book a Demo 
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.
LambdaTest Learning Hubs:
- JUnit Tutorial
- TestNG Tutorial
- Webdriver Tutorial
- WebDriverIO Tutorial
- Protractor Tutorial
- Selenium 4 Tutorial
- Jenkins Tutorial
- NUnit Tutorial
- Jest Tutorial
- Playwright Tutorial
- Cypress Tutorial
- PyTest Tutorial
YouTube
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Run tempest automation tests on LambdaTest cloud grid
Perform automation testing on 3000+ real desktop and mobile devices online.
Try LambdaTest Now !!
Get 100 minutes of automation test minutes FREE!!
Was this article helpful?
