Best Python code snippet using tempest_python
test_plugin.py
Source:test_plugin.py
...74 l3_sync_worker.sync_router_interfaces(routers)75 lapse = time.time() - start_time76 if lapse >= argvalue:77 break78def _call_add_router_interface(l3_plugin, ctx, router_id, interface_info, with_hold = False):79 #db_base_plugin_v2.NeutronDbPluginV2's method without wrapper "@db_api.context_manager.reader", to avoid sqlite80 def get_port(self, context, id, fields=None):81 port = self._get_port(context, id)82 return self._make_port_dict(port, fields)83 def delayed__validate_interface_info(*args, **kwargs):84 import time85 time.sleep(6)86 interface_info = args[0]87 return original__validate_interface_info(interface_info)88 method = getattr(l3_plugin, 'add_router_interface')89 with patch.object(db_base_plugin_common.DbBasePluginCommon, '_store_ip_allocation' , test_helper._store_ip_allocation):90 with patch.object(db_base_plugin_v2.NeutronDbPluginV2, 'get_port' , get_port):91 if with_hold:92 #add delay inside transaction block of add_router_interface93 original__validate_interface_info = l3_plugin._validate_interface_info94 l3_plugin._validate_interface_info = Mock(side_effect = delayed__validate_interface_info)95 # call l3_plugin.add_router_interface96 method(ctx, router_id, interface_info)97 #revert back delay98 if with_hold:99 l3_plugin._validate_interface_info = original__validate_interface_info100def _call_update_router(l3_plugin, ctx, router_id, router, with_hold = False):101 #db_base_plugin_v2.db_base_plugin_common.DbBasePluginCommon's method without wrapper "@db_api.context_manager.reader", to avoid sqlite102 def _get_subnets_by_network(self, context, network_id):103 from neutron.objects import subnet as subnet_obj104 return subnet_obj.Subnet.get_objects(context, network_id=network_id)105 def _kaloom_nw_name(prefix, network_id):106 return prefix + network_id107 def delayed__kaloom_nw_name(prefix, network_id):108 import time109 time.sleep(6)110 return prefix + network_id111 if with_hold:112 patch_as = delayed__kaloom_nw_name113 else:114 patch_as = _kaloom_nw_name115 db_base_plugin_v2.db_base_plugin_common.DbBasePluginCommon._get_subnets_by_network = _get_subnets_by_network116 db_base_plugin_v2.db_base_plugin_common.DbBasePluginCommon._store_ip_allocation = test_helper._store_ip_allocation117 with patch.object(plugin.utils, '_kaloom_nw_name' , patch_as):118 method = getattr(l3_plugin, 'update_router')119 method(ctx, router_id, router)120def _call_remove_router_interface(l3_plugin, ctx, router_id, interface_info, with_hold = False):121 def delayed__validate_interface_info(*args, **kwargs):122 import time123 time.sleep(6)124 interface_info = args[0]125 return original__validate_interface_info(interface_info)126 method = getattr(l3_plugin, 'remove_router_interface')127 if with_hold:128 #add delay inside transaction block of delete_router_interface129 original__validate_interface_info = l3_plugin._validate_interface_info130 l3_plugin._validate_interface_info = Mock(side_effect = delayed__validate_interface_info)131 #call l3_plugin.remove_router_interface132 method(ctx, router_id, interface_info)133 #revert back delay134 if with_hold:135 l3_plugin._validate_interface_info = original__validate_interface_info136class KaloomL3ServicePluginTestCase(base.BaseTestCase):137 def setUp(self):138 super(KaloomL3ServicePluginTestCase, self).setUp()139 self.mysqld = testing.mysqld.Mysqld(my_cnf={'skip-networking': None})140 URL = create_engine(self.mysqld.url()).url141 oslo_cfg.CONF.set_override('connection', URL, group='database')142 #"neutron-db-manage upgrade head" equivalent143 alembic_cfg = migration_cli.get_neutron_config()144 alembic_cfg.neutron_config = oslo_cfg.CONF145 test_helper.upgrade(alembic_cfg)146 #"neutron-kaloom-db-manage upgrade head" equivalent147 script_location = 'networking_kaloom.ml2.drivers.kaloom.db.migration:alembic_migrations'148 alembic_cfg_kaloom = alembic_config.Config()149 alembic_cfg_kaloom.set_main_option("script_location", script_location)150 alembic_cfg_kaloom.neutron_config = oslo_cfg.CONF151 test_helper.upgrade(alembic_cfg_kaloom)152 with patch.object(neutron_context.db_api,'_CTX_MANAGER', test_helper.get_context_manager(URL)):153 self.ctx = neutron_context.get_admin_context()154 self.ctx.session155 #print self.ctx.session.connection().engine156 self.setup_coreplugin(load_plugins=False)157 oslo_cfg.CONF.set_override("core_plugin", test_db_base_plugin_v2.DB_PLUGIN_KLASS)158 oslo_cfg.CONF.set_override("service_plugins", ['segments'])159 manager.init()160 self.l2_plugin = directory.get_plugin()161 self.segments_plugin = importutils.import_object('neutron.services.segments.plugin.Plugin')162 #patch163 plugin.db_api._CTX_MANAGER = test_helper.get_context_manager(URL)164 plugin.kaloom_db.db_api.context_manager = test_helper.get_context_manager(URL)165 db_api.context_manager = test_helper.get_context_manager(URL)166 #fix l3_port_check issue167 def delete_port(context, id, l3_port_check=True):168 return original_delete_port(context, id)169 original_delete_port = self.l2_plugin.delete_port170 self.l2_plugin.delete_port = Mock(side_effect = delete_port)171 def tearDown(self):172 super(KaloomL3ServicePluginTestCase, self).tearDown()173 self.mysqld.stop()174 def create_network_subnet_router(self, l3_plugin, mock_KaloomL3Driver_instance, ext_net = False):175 #create network, subnet in db176 if ext_net:177 net_id = test_helper._create_network_ext(self)178 else:179 net_id = test_helper._create_network(self)180 seg_id = test_helper._create_segment(self, net_id)181 subnet_id = test_helper._create_subnet(self, seg_id, net_id)182 router = test_helper.get_mock_router_kwargs()183 # creates a router184 method = getattr(l3_plugin, 'create_router')185 method(self.ctx, router)186 mock_KaloomL3Driver_instance.create_router.assert_called_once()187 mock_KaloomL3Driver_instance.reset_mock()188 return net_id, subnet_id, router189 @patch('networking_kaloom.services.l3.driver.KaloomNetconf',autospec=True)190 def test_router_create_delete_sync_interface(self, mock_KaloomNetconf):191 #patch super class, which has mocked certain methods192 patcher = patch.object(plugin.KaloomL3ServicePlugin, '__bases__', (test_helper.MockParent,))193 with patcher:194 patcher.is_local = True #avoids delattr error195 l3_plugin = plugin.KaloomL3ServicePlugin()196 mock_KaloomNetconf_instance = mock_KaloomNetconf.return_value197 router = test_helper.get_mock_router_kwargs()198 # creates a router199 method = getattr(l3_plugin, 'create_router')200 method(self.ctx, router)201 mock_KaloomNetconf_instance.create_router.assert_called_once()202 # l3_sync atomic read between neutron_db and vfabric203 # starts a thread that will take the write lock and hold it for a while204 read_time = 4205 routers = [router['router']]206 vfabric_routers = []207 concurrent_test_fake_synchronize_read = threading.Thread(target=_test_fake_synchronize_read,208 args=(read_time, routers, vfabric_routers))209 concurrent_test_fake_synchronize_read.start()210 # meanwhile try to delete the router211 method = getattr(l3_plugin, 'delete_router')212 method(self.ctx, router['router']['id'])213 mock_KaloomNetconf_instance.delete_router.assert_called_once()214 # once the router is deleted, invokes sync_router_interfaces which should simply 215 # complain that the router does not exist but still return gracefully216 _test_sync_router_interfaces(routers)217 LOG.warning.assert_called_with(test_helper.SubstringMatcher(containing=LOG_MESSAGE_SYNC_INTERFACES_LOCK_WARNING), ANY)218 #wait for threads219 concurrent_test_fake_synchronize_read.join()220 @patch('networking_kaloom.services.l3.plugin.kaloom_l3_driver.KaloomL3Driver',autospec=True)221 def test_concurrent_add_interfaces(self, mock_KaloomL3Driver):222 #patch super class, which has mocked certain methods223 patcher = patch.object(plugin.KaloomL3ServicePlugin, '__bases__', (test_helper.MockParent,))224 with patcher:225 patcher.is_local = True #avoids delattr error226 l3_plugin = plugin.KaloomL3ServicePlugin()227 mock_KaloomL3Driver_instance = mock_KaloomL3Driver.return_value228 net_id, subnet_id, router = self.create_network_subnet_router(l3_plugin, mock_KaloomL3Driver_instance)229 #concurrent "sync router interfaces": loops until loop_time seconds to make sure "sync router interfaces"230 # runs before, in-between and after "add_router_interface"231 loop_time = 4232 routers = [router['router']]233 concurrent_sync_router_interfaces = threading.Thread(target=_test_sync_router_interfaces, args=(routers, 'loop_time', loop_time))234 concurrent_sync_router_interfaces.start()235 #concurrent "add_router_interface"236 time.sleep(2)237 interface_info = {'subnet_id': subnet_id}238 _call_add_router_interface(l3_plugin, self.ctx, router['router']['id'], interface_info, with_hold = False)239 #wait for thread240 concurrent_sync_router_interfaces.join()241 # atomicity rule: Once L3Driver.add_router_interface get called (by plugin.add_router_interface), then only242 # L3Driver.router_l2node_link_exists should be called (by plugin.sync_router_interfaces).243 assert mock_KaloomL3Driver_instance.method_calls[0] == call.add_router_interface(ANY, ANY)244 @patch('networking_kaloom.services.l3.plugin.kaloom_l3_driver.KaloomL3Driver',autospec=True)245 def test_atomicity_of_add_router_interface(self, mock_KaloomL3Driver):246 #patch super class, which has mocked certain methods247 patcher = patch.object(plugin.KaloomL3ServicePlugin, '__bases__', (test_helper.MockParent,))248 with patcher:249 patcher.is_local = True #avoids delattr error250 l3_plugin = plugin.KaloomL3ServicePlugin()251 mock_KaloomL3Driver_instance = mock_KaloomL3Driver.return_value252 net_id, subnet_id, router = self.create_network_subnet_router(l3_plugin, mock_KaloomL3Driver_instance)253 #concurrent "sync router interfaces", runs when below "add_router_interface" is in critical section.254 routers = [router['router']]255 delay_start = 3256 concurrent_sync_router_interfaces = threading.Thread(target=_test_sync_router_interfaces, args=(routers,'delay_start', delay_start))257 concurrent_sync_router_interfaces.start()258 #concurrent "add_router_interface": that holds for certain time259 interface_info = {'subnet_id': subnet_id}260 _call_add_router_interface(l3_plugin, self.ctx, router['router']['id'], interface_info, with_hold = True)261 #wait for thread262 concurrent_sync_router_interfaces.join()263 #sync_router_interfaces should not see router interfaces (before add_router_interface completes)264 # atomicity rule: Once L3Driver.add_router_interface get called (by plugin.add_router_interface), then only265 # L3Driver.router_l2node_link_exists should be called (by plugin.sync_router_interfaces).266 assert mock_KaloomL3Driver_instance.method_calls[0] == call.add_router_interface(ANY, ANY)267 assert mock_KaloomL3Driver_instance.method_calls[1] == call.router_l2node_link_exists(ANY, ANY, ANY)268 @patch('networking_kaloom.services.l3.plugin.kaloom_l3_driver.KaloomL3Driver',autospec=True)269 def test_atomicity_of_sync_router_interfaces(self, mock_KaloomL3Driver):270 #patch super class, which has mocked certain methods271 patcher = patch.object(plugin.KaloomL3ServicePlugin, '__bases__', (test_helper.MockParent,))272 with patcher:273 patcher.is_local = True #avoids delattr error274 l3_plugin = plugin.KaloomL3ServicePlugin()275 mock_KaloomL3Driver_instance = mock_KaloomL3Driver.return_value276 net_id, subnet_id, router = self.create_network_subnet_router(l3_plugin, mock_KaloomL3Driver_instance)277 #concurrent "sync router interfaces", that holds for certain time278 routers = [router['router']]279 concurrent_sync_router_interfaces = threading.Thread(target=_test_sync_router_interfaces, args=(routers,'hold', ))280 concurrent_sync_router_interfaces.start()281 #concurrent "add_router_interface": runs when above "sync router interfaces" is in critical section.282 interface_info = {'subnet_id': subnet_id}283 _call_add_router_interface(l3_plugin, self.ctx, router['router']['id'], interface_info, with_hold = False)284 #wait for thread285 concurrent_sync_router_interfaces.join()286 #sync_router_interfaces should not see router interfaces (that came once it called)287 mock_KaloomL3Driver_instance.router_l2node_link_exists.assert_not_called()288 @patch('networking_kaloom.services.l3.plugin.kaloom_l3_driver.KaloomL3Driver',autospec=True)289 def test_atomicity_of_remove_router_interface(self, mock_KaloomL3Driver):290 #patch super class, which has mocked certain methods291 patcher = patch.object(plugin.KaloomL3ServicePlugin, '__bases__', (test_helper.MockParent,))292 with patcher:293 patcher.is_local = True #avoids delattr error294 l3_plugin = plugin.KaloomL3ServicePlugin()295 mock_KaloomL3Driver_instance = mock_KaloomL3Driver.return_value296 net_id, subnet_id, router = self.create_network_subnet_router(l3_plugin, mock_KaloomL3Driver_instance)297 #add router_interface298 interface_info = {'subnet_id': subnet_id}299 _call_add_router_interface(l3_plugin, self.ctx, router['router']['id'], interface_info, with_hold = False)300 mock_KaloomL3Driver_instance.reset_mock()301 #concurrent "remove_router_interface", that holds for certain time302 with_hold = True303 concurrent_remove_router_interface = threading.Thread(target=_call_remove_router_interface,304 args=(l3_plugin, self.ctx, router['router']['id'], interface_info, with_hold))305 concurrent_remove_router_interface.start()306 #add router_interface, runs when above "delete_router_interface" is in critical section.307 time.sleep(2)308 l3_plugin_1 = plugin.KaloomL3ServicePlugin()309 _call_add_router_interface(l3_plugin_1, self.ctx, router['router']['id'], interface_info, with_hold = False)310 #wait for thread311 concurrent_remove_router_interface.join()312 # atomicity rule: while remove_router_interface is running, add_router_interface should get blocked until313 # remove_router_interface completes.314 assert mock_KaloomL3Driver_instance.method_calls[0] == call.remove_router_interface(ANY, ANY)315 assert mock_KaloomL3Driver_instance.method_calls[1] == call.add_router_interface(ANY, ANY)316 @patch('networking_kaloom.services.l3.plugin.kaloom_l3_driver.KaloomL3Driver',autospec=True)317 def test_atomicity_of_unset_external_gateway(self, mock_KaloomL3Driver):318 #patch super class, which has mocked certain methods319 patcher = patch.object(plugin.KaloomL3ServicePlugin, '__bases__', (test_helper.MockParent,))320 with patcher:321 patcher.is_local = True #avoids delattr error322 l3_plugin = plugin.KaloomL3ServicePlugin()323 mock_KaloomL3Driver_instance = mock_KaloomL3Driver.return_value324 net_id, subnet_id, router = self.create_network_subnet_router(l3_plugin, mock_KaloomL3Driver_instance, ext_net = True)325 #set --external-gateway326 router_set_ext = {'router':{'external_gateway_info': {'network_id': net_id, 'enable_snat': False, 327 'external_fixed_ips': [{'ip_address':'192.168.10.3'}]}}}328 _call_update_router(l3_plugin, self.ctx, router['router']['id'], router_set_ext, with_hold = False)329 mock_KaloomL3Driver_instance.reset_mock()330 #concurrent "unset --external-gateway", that holds for certain time331 with_hold = True332 router_unset_ext = {'router':{'external_gateway_info': {}}}333 concurrent_unset_external_gateway = threading.Thread(target=_call_update_router,334 args=(l3_plugin, self.ctx, router['router']['id'], router_unset_ext, with_hold))335 concurrent_unset_external_gateway.start()336 #set --external-gateway, runs when above "unset --external-gateway" is in critical section.337 time.sleep(2)338 l3_plugin_1 = plugin.KaloomL3ServicePlugin()339 _call_update_router(l3_plugin_1, self.ctx, router['router']['id'], router_set_ext, with_hold = False)340 #wait for thread341 concurrent_unset_external_gateway.join()342 # atomicity rule: while unset_external_gateway is running, set_external_gateway should get blocked until343 # unset_external_gateway completes.344 assert mock_KaloomL3Driver_instance.method_calls[0] == call.remove_router_interface(ANY, ANY)...
test_routers_rbac.py
Source:test_routers_rbac.py
...213 network = self.create_network()214 subnet = self.create_subnet(network)215 router = self.create_router()216 self.rbac_utils.switch_role(self, toggle_rbac_role=True)217 self.routers_client.add_router_interface(218 router['id'], subnet_id=subnet['id'])219 self.addCleanup(220 test_utils.call_and_ignore_notfound_exc,221 self.routers_client.remove_router_interface,222 router['id'],223 subnet_id=subnet['id'])224 @rbac_rule_validation.action(service="neutron",225 rule="remove_router_interface",226 expected_error_code=404)227 @decorators.idempotent_id('ff2593a4-2bff-4c27-97d3-dd3702b27dfb')228 def test_remove_router_interfaces(self):229 """Remove Router Interface230 RBAC test for the neutron remove_router_interface policy231 """232 network = self.create_network()233 subnet = self.create_subnet(network)234 router = self.create_router()235 self.routers_client.add_router_interface(236 router['id'], subnet_id=subnet['id'])237 self.addCleanup(test_utils.call_and_ignore_notfound_exc,238 self.routers_client.remove_router_interface,239 router['id'],240 subnet_id=subnet['id'])241 self.rbac_utils.switch_role(self, toggle_rbac_role=True)242 self.routers_client.remove_router_interface(243 router['id'],...
k5-connect-router.py
Source:k5-connect-router.py
1#!/usr/bin/env python2# -*- coding: utf-8 -*-3"""4PUT /v2.0/routers/{router_id}/add_router_interface5Add interface to router6è«çã«ã¼ã¿ã¸å
é¨ã¤ã³ã¿ã¼ãã§ã¼ã¹ã追å ãã7"""8"""9å®è¡ä¾(æå)10bash-4.4$ ./bin/k5-connect-router.py \11 --router-id ffbd70be-24cf-4dff-a4f6-661bf892e313 \12 --port-id 689d24c7-02a2-4dfd-b809-9ad4060e079f13status_code: 20014PUT /v2.0/routers/{router_id}/add_router_interface15========= ====================================16id ffbd70be-24cf-4dff-a4f6-661bf892e31317port_id 689d24c7-02a2-4dfd-b809-9ad4060e079f18subnet_id abbbbcf4-ea8f-4218-bbe7-669231eeba3019az jp-east-1a20tenant_id a5001a8b9c4a4712985c11377bd6d4fe21========= ====================================22bash-4.4$23"""24import json25import logging26import os27import sys28def here(path=''):29 """ç¸å¯¾ãã¹ã絶対ãã¹ã«å¤æãã¦è¿å´ãã¾ã"""30 if getattr(sys, 'frozen', False):31 # cx_Freezeã§åºããå ´åã¯å®è¡ãã¡ã¤ã«ããã®ç¸å¯¾ãã¹32 return os.path.abspath(os.path.join(os.path.dirname(sys.executable), path))33 else:34 # é常ã¯ãã®ãã¡ã¤ã«ã®å ´æããã®ç¸å¯¾ãã¹35 return os.path.abspath(os.path.join(os.path.dirname(__file__), path))36# libãã©ã«ãã«ãããpythonã¹ã¯ãªãããèªã¿ãã¾ããããã®å¦ç37if not here("../lib") in sys.path:38 sys.path.append(here("../lib"))39if not here("../lib/site-packages") in sys.path:40 sys.path.append(here("../lib/site-packages"))41try:42 from k5c import k5c43except ImportError as e:44 logging.exception("k5cã¢ã¸ã¥ã¼ã«ã®ã¤ã³ãã¼ãã«å¤±æãã¾ãã: %s", e)45 sys.exit(1)46try:47 from tabulate import tabulate48except ImportError as e:49 logging.exception("tabulateã¢ã¸ã¥ã¼ã«ã®ã¤ã³ãã¼ãã«å¤±æãã¾ãã: %s", e)50 sys.exit(1)51#52# ãªã¯ã¨ã¹ããã¼ã¿ãä½æãã53#54def make_request_data(port_id=""):55 """ãªã¯ã¨ã¹ããã¼ã¿ãä½æãã¦è¿å´ãã¾ã"""56 return {'port_id': port_id}57#58# APIã«ã¢ã¯ã»ã¹ãã59#60def access_api(router_id="", data=None):61 """REST APIã«ã¢ã¯ã»ã¹ãã¾ã"""62 # æ¥ç¶å
63 url = k5c.EP_NETWORK + "/v2.0/routers/" + router_id + "/add_router_interface"64 # Clientã¯ã©ã¹ãã¤ã³ã¹ã¿ã³ã¹å65 c = k5c.Client()66 # PUTã¡ã½ãããçºè¡ãã¦ãçµæã®ãªãã¸ã§ã¯ããå¾ã67 r = c.put(url=url, data=data)68 return r69#70# çµæã表示ãã71#72def print_result(result):73 """çµæã表示ãã¾ã"""74 # ã¹ãã¼ã¿ã¹ã³ã¼ãã¯'status_code'ãã¼ã«æ ¼ç´75 status_code = result.get('status_code', -1)76 # ã¹ãã¼ã¿ã¹ã³ã¼ããç°å¸¸ãªå ´å77 if status_code < 0 or status_code >= 400:78 print(json.dumps(result, indent=2))79 return80 # ãã¼ã¿ã¯'data'ãã¼ã«æ ¼ç´81 data = result.get('data', None)82 if not data:83 logging.error("no data found")84 return85 # ãã¼ã¿ãªãã¸ã§ã¯ãç´ä¸ã«å
¥ã£ã¦ãã86 # "data": {87 # "tenant_id": "a5001a8b9c4a4712985c11377bd6d4fe",88 # "port_id": "b9fc91fd-6ae9-4c75-be47-94d818a6296f",89 # "id": "ffbd70be-24cf-4dff-a4f6-661bf892e313",90 # "availability_zone": "jp-east-1a",91 # "subnet_id": "abbbbcf4-ea8f-4218-bbe7-669231eeba30"92 # },93 # 表示ç¨ã«é
åã«ãã94 rtrs = []95 rtrs.append(['id', data.get('id', '')])96 rtrs.append(['port_id', data.get('port_id', '')])97 rtrs.append(['subnet_id', data.get('subnet_id', '')])98 rtrs.append(['az', data.get('availability_zone', '')])99 rtrs.append(['tenant_id', data.get('tenant_id', '')])100 # çµæ表示101 print("status_code: {0}".format(result.get('status_code', "")))102 print("PUT /v2.0/routers/{router_id}/add_router_interface")103 print(tabulate(rtrs, tablefmt='rst'))104if __name__ == '__main__':105 import argparse106 def main():107 """ã¡ã¤ã³é¢æ°"""108 parser = argparse.ArgumentParser(description='Adds an internal interface to a logical router.')109 parser.add_argument('--router-id', dest='router_id', metavar='id', required=True, help='The router id.')110 parser.add_argument('--port-id', dest='port_id', metavar='id', required=True, help='The port id.')111 parser.add_argument('--dump', action='store_true', default=False, help='Dump json result and exit.')112 args = parser.parse_args()113 router_id = args.router_id114 port_id = args.port_id115 dump = args.dump116 # ãªã¯ã¨ã¹ããã¼ã¿ãä½æ117 data = make_request_data(port_id=port_id)118 # å®è¡119 result = access_api(router_id=router_id, data=data)120 # ä¸èº«ã確èª121 if dump:122 print(json.dumps(result, indent=2))123 return 0124 # 表示125 print_result(result)126 return 0127 # å®è¡...
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.
You could also refer to video tutorials over LambdaTest YouTube channel to get step by step demonstration from industry experts.
Get 100 minutes of automation test minutes FREE!!