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!!
