Test your AI Agents with the all-new Agent to Agent Testing Platform.Learn More

How to use _create_router method in tempest

Best Python code snippet using tempest_python

test_l3_ha_db.py

Source:test_l3_ha_db.py

...53        self.plugin.create_or_update_agent(self.admin_ctx, agent_status)54        agent_status['host'] = 'l3host_2'55        self.plugin.create_or_update_agent(self.admin_ctx, agent_status)56        self.agent1, self.agent2 = self.plugin.get_agents(self.admin_ctx)57    def _create_router(self, ha=True, tenant_id='tenant1', distributed=None,58                       ctx=None):59        if ctx is None:60            ctx = self.admin_ctx61        ctx.tenant_id = tenant_id62        router = {'name': 'router1', 'admin_state_up': True}63        if ha is not None:64            router['ha'] = ha65        if distributed is not None:66            router['distributed'] = distributed67        return self.plugin.create_router(ctx, {'router': router})68    def _update_router(self, router_id, ha=True, distributed=None, ctx=None):69        if ctx is None:70            ctx = self.admin_ctx71        data = {'ha': ha} if ha is not None else {}72        if distributed is not None:73            data['distributed'] = distributed74        return self.plugin._update_router_db(ctx, router_id,75                                             data, None)76    def _bind_router(self, router_id):77        with self.admin_ctx.session.begin(subtransactions=True):78            bindings = self.plugin.get_ha_router_port_bindings(self.admin_ctx,79                                                               [router_id])80            for agent_id, binding in zip(81                    [self.agent1['id'], self.agent2['id']], bindings):82                binding.l3_agent_id = agent_id83class L3HATestCase(L3HATestFramework):84    def test_verify_configuration_succeed(self):85        # Default configuration should pass86        self.plugin._verify_configuration()87    def test_verify_configuration_l3_ha_net_cidr_is_not_a_cidr(self):88        cfg.CONF.set_override('l3_ha_net_cidr', 'not a cidr')89        self.assertRaises(90            l3_ext_ha_mode.HANetworkCIDRNotValid,91            self.plugin._verify_configuration)92    def test_verify_configuration_l3_ha_net_cidr_is_not_a_subnet(self):93        cfg.CONF.set_override('l3_ha_net_cidr', '10.0.0.1/8')94        self.assertRaises(95            l3_ext_ha_mode.HANetworkCIDRNotValid,96            self.plugin._verify_configuration)97    def test_verify_configuration_min_l3_agents_per_router_below_minimum(self):98        cfg.CONF.set_override('min_l3_agents_per_router', 0)99        self.assertRaises(100            l3_ext_ha_mode.HAMinimumAgentsNumberNotValid,101            self.plugin._verify_configuration)102    def test_verify_configuration_max_l3_agents_below_min_l3_agents(self):103        cfg.CONF.set_override('max_l3_agents_per_router', 3)104        cfg.CONF.set_override('min_l3_agents_per_router', 4)105        self.assertRaises(106            l3_ext_ha_mode.HAMaximumAgentsNumberNotValid,107            self.plugin._verify_configuration)108    def test_ha_router_create(self):109        router = self._create_router()110        self.assertTrue(router['ha'])111    def test_ha_router_create_with_distributed(self):112        self.assertRaises(l3_ext_ha_mode.DistributedHARouterNotSupported,113                          self._create_router,114                          distributed=True)115    def test_no_ha_router_create(self):116        router = self._create_router(ha=False)117        self.assertFalse(router['ha'])118    def test_router_create_with_ha_conf_enabled(self):119        cfg.CONF.set_override('l3_ha', True)120        router = self._create_router(ha=None)121        self.assertTrue(router['ha'])122    def test_migration_from_ha(self):123        router = self._create_router()124        self.assertTrue(router['ha'])125        router = self._update_router(router['id'], ha=False)126        self.assertFalse(router.extra_attributes['ha'])127        self.assertIsNone(router.extra_attributes['ha_vr_id'])128    def test_migration_to_ha(self):129        router = self._create_router(ha=False)130        self.assertFalse(router['ha'])131        router = self._update_router(router['id'], ha=True)132        self.assertTrue(router.extra_attributes['ha'])133        self.assertIsNotNone(router.extra_attributes['ha_vr_id'])134    def test_migrate_ha_router_to_distributed(self):135        router = self._create_router()136        self.assertTrue(router['ha'])137        self.assertRaises(l3_ext_ha_mode.DistributedHARouterNotSupported,138                          self._update_router,139                          router['id'],140                          distributed=True)141    def test_l3_agent_routers_query_interface(self):142        router = self._create_router()143        self._bind_router(router['id'])144        routers = self.plugin.get_ha_sync_data_for_host(self.admin_ctx,145                                                        self.agent1['host'])146        self.assertEqual(1, len(routers))147        router = routers[0]148        self.assertIsNotNone(router.get('ha'))149        interface = router.get(constants.HA_INTERFACE_KEY)150        self.assertIsNotNone(interface)151        self.assertEqual(constants.DEVICE_OWNER_ROUTER_HA_INTF,152                         interface['device_owner'])153        self.assertEqual(cfg.CONF.l3_ha_net_cidr, interface['subnet']['cidr'])154    def test_update_state(self):155        router = self._create_router()156        self._bind_router(router['id'])157        routers = self.plugin.get_ha_sync_data_for_host(self.admin_ctx,158                                                        self.agent1['host'])159        state = routers[0].get(constants.HA_ROUTER_STATE_KEY)160        self.assertEqual('standby', state)161        self.plugin.update_router_state(self.admin_ctx, router['id'], 'active',162                                        self.agent1['host'])163        routers = self.plugin.get_ha_sync_data_for_host(self.admin_ctx,164                                                        self.agent1['host'])165        state = routers[0].get(constants.HA_ROUTER_STATE_KEY)166        self.assertEqual('active', state)167    def test_unique_ha_network_per_tenant(self):168        tenant1 = _uuid()169        tenant2 = _uuid()170        self._create_router(tenant_id=tenant1)171        self._create_router(tenant_id=tenant2)172        ha_network1 = self.plugin.get_ha_network(self.admin_ctx, tenant1)173        ha_network2 = self.plugin.get_ha_network(self.admin_ctx, tenant2)174        self.assertNotEqual(175            ha_network1['network_id'], ha_network2['network_id'])176    def _deployed_router_change_ha_flag(self, to_ha):177        self._create_router(ha=not to_ha)178        routers = self.plugin.get_ha_sync_data_for_host(self.admin_ctx)179        router = routers[0]180        interface = router.get(constants.HA_INTERFACE_KEY)181        if to_ha:182            self.assertIsNone(interface)183        else:184            self.assertIsNotNone(interface)185        self._update_router(router['id'], to_ha)186        routers = self.plugin.get_ha_sync_data_for_host(self.admin_ctx)187        router = routers[0]188        interface = router.get(constants.HA_INTERFACE_KEY)189        if to_ha:190            self.assertIsNotNone(interface)191        else:192            self.assertIsNone(interface)193    def test_deployed_router_can_have_ha_enabled(self):194        self._deployed_router_change_ha_flag(to_ha=True)195    def test_deployed_router_can_have_ha_disabled(self):196        self._deployed_router_change_ha_flag(to_ha=False)197    def test_create_ha_router_notifies_agent(self):198        self._create_router()199        self.assertTrue(self.notif_m.called)200    def test_update_router_to_ha_notifies_agent(self):201        router = self._create_router(ha=False)202        self.notif_m.reset_mock()203        self._update_router(router['id'], ha=True)204        self.assertTrue(self.notif_m.called)205    def test_unique_vr_id_between_routers(self):206        self._create_router()207        self._create_router()208        routers = self.plugin.get_ha_sync_data_for_host(self.admin_ctx)209        self.assertEqual(2, len(routers))210        self.assertNotEqual(routers[0]['ha_vr_id'], routers[1]['ha_vr_id'])211    @mock.patch('neutron.db.l3_hamode_db.VR_ID_RANGE', new=set(range(1, 1)))212    def test_vr_id_depleted(self):213        self.assertRaises(l3_ext_ha_mode.NoVRIDAvailable, self._create_router)214    @mock.patch('neutron.db.l3_hamode_db.VR_ID_RANGE', new=set(range(1, 2)))215    def test_vr_id_unique_range_per_tenant(self):216        self._create_router()217        self._create_router(tenant_id=_uuid())218        routers = self.plugin.get_ha_sync_data_for_host(self.admin_ctx)219        self.assertEqual(2, len(routers))220        self.assertEqual(routers[0]['ha_vr_id'], routers[1]['ha_vr_id'])221    @mock.patch('neutron.db.l3_hamode_db.MAX_ALLOCATION_TRIES', new=2)222    def test_vr_id_allocation_contraint_conflict(self):223        router = self._create_router()224        network = self.plugin.get_ha_network(self.admin_ctx,225                                             router['tenant_id'])226        with mock.patch.object(self.plugin, '_get_allocated_vr_id',227                               return_value=set()) as alloc:228            self.assertRaises(l3_ext_ha_mode.MaxVRIDAllocationTriesReached,229                              self.plugin._allocate_vr_id, self.admin_ctx,230                              network.network_id, router['id'])231            self.assertEqual(2, len(alloc.mock_calls))232    def test_vr_id_allocation_delete_router(self):233        router = self._create_router()234        network = self.plugin.get_ha_network(self.admin_ctx,235                                             router['tenant_id'])236        allocs_before = self.plugin._get_allocated_vr_id(self.admin_ctx,237                                                         network.network_id)238        router = self._create_router()239        allocs_current = self.plugin._get_allocated_vr_id(self.admin_ctx,240                                                          network.network_id)241        self.assertNotEqual(allocs_before, allocs_current)242        self.plugin.delete_router(self.admin_ctx, router['id'])243        allocs_after = self.plugin._get_allocated_vr_id(self.admin_ctx,244                                                        network.network_id)245        self.assertEqual(allocs_before, allocs_after)246    def test_vr_id_allocation_router_migration(self):247        router = self._create_router()248        network = self.plugin.get_ha_network(self.admin_ctx,249                                             router['tenant_id'])250        allocs_before = self.plugin._get_allocated_vr_id(self.admin_ctx,251                                                         network.network_id)252        router = self._create_router()253        self._update_router(router['id'], ha=False)254        allocs_after = self.plugin._get_allocated_vr_id(self.admin_ctx,255                                                        network.network_id)256        self.assertEqual(allocs_before, allocs_after)257    def test_one_ha_router_one_not(self):258        self._create_router(ha=False)259        self._create_router()260        routers = self.plugin.get_ha_sync_data_for_host(self.admin_ctx)261        ha0 = routers[0]['ha']262        ha1 = routers[1]['ha']263        self.assertNotEqual(ha0, ha1)264    def test_add_ha_port_binding_failure_rolls_back_port(self):265        router = self._create_router()266        device_filter = {'device_id': [router['id']]}267        ports_before = self.core_plugin.get_ports(268            self.admin_ctx, filters=device_filter)269        network = self.plugin.get_ha_network(self.admin_ctx,270                                             router['tenant_id'])271        with mock.patch.object(self.plugin, '_create_ha_port_binding',272                               side_effect=ValueError):273            self.assertRaises(ValueError, self.plugin.add_ha_port,274                              self.admin_ctx, router['id'], network.network_id,275                              router['tenant_id'])276        ports_after = self.core_plugin.get_ports(277            self.admin_ctx, filters=device_filter)278        self.assertEqual(ports_before, ports_after)279    def test_create_ha_network_binding_failure_rolls_back_network(self):280        networks_before = self.core_plugin.get_networks(self.admin_ctx)281        with mock.patch.object(self.plugin,282                               '_create_ha_network_tenant_binding',283                               side_effect=ValueError):284            self.assertRaises(ValueError, self.plugin._create_ha_network,285                              self.admin_ctx, _uuid())286        networks_after = self.core_plugin.get_networks(self.admin_ctx)287        self.assertEqual(networks_before, networks_after)288    def test_create_ha_network_subnet_failure_rolls_back_network(self):289        networks_before = self.core_plugin.get_networks(self.admin_ctx)290        with mock.patch.object(self.plugin, '_create_ha_subnet',291                               side_effect=ValueError):292            self.assertRaises(ValueError, self.plugin._create_ha_network,293                              self.admin_ctx, _uuid())294        networks_after = self.core_plugin.get_networks(self.admin_ctx)295        self.assertEqual(networks_before, networks_after)296    def test_create_ha_interfaces_binding_failure_rolls_back_ports(self):297        router = self._create_router()298        network = self.plugin.get_ha_network(self.admin_ctx,299                                             router['tenant_id'])300        device_filter = {'device_id': [router['id']]}301        ports_before = self.core_plugin.get_ports(302            self.admin_ctx, filters=device_filter)303        router_db = self.plugin._get_router(self.admin_ctx, router['id'])304        with mock.patch.object(self.plugin, '_create_ha_port_binding',305                               side_effect=ValueError):306            self.assertRaises(ValueError, self.plugin._create_ha_interfaces,307                              self.admin_ctx, router_db, network)308        ports_after = self.core_plugin.get_ports(309            self.admin_ctx, filters=device_filter)310        self.assertEqual(ports_before, ports_after)311    def test_create_router_db_ha_attribute_failure_rolls_back_router(self):312        routers_before = self.plugin.get_routers(self.admin_ctx)313        for method in ('_set_vr_id',314                       '_create_ha_interfaces',315                       '_notify_ha_interfaces_updated'):316            with mock.patch.object(self.plugin, method,317                                   side_effect=ValueError):318                self.assertRaises(ValueError, self._create_router)319        routers_after = self.plugin.get_routers(self.admin_ctx)320        self.assertEqual(routers_before, routers_after)321class L3HAUserTestCase(L3HATestFramework):322    def setUp(self):323        super(L3HAUserTestCase, self).setUp()324        self.user_ctx = context.Context('', _uuid())325    def test_create_ha_router(self):326        self._create_router(ctx=self.user_ctx)327    def test_update_router(self):328        router = self._create_router(ctx=self.user_ctx)329        self._update_router(router['id'], ha=False, ctx=self.user_ctx)330    def test_delete_router(self):331        router = self._create_router(ctx=self.user_ctx)...

test_l3_conntrack_helper.py

Source:test_l3_conntrack_helper.py

...22        if not self.is_extension_enabled('l3-conntrack-helper'):23            self.skipTest("No l3-conntrack-helper extension present")24        if not self.is_extension_enabled('expose-l3-conntrack-helper'):25            self.skipTest("No expose-l3-conntrack-helper extension present")26    def _create_router(self):27        router_name = uuid.uuid4().hex28        json_output = json.loads(self.openstack(29            'router create -f json ' + router_name30        ))31        self.assertIsNotNone(json_output['id'])32        router_id = json_output['id']33        self.addCleanup(self.openstack, 'router delete ' + router_id)34        return router_id35    def _create_helpers(self, router_id, helpers):36        created_helpers = []37        for helper in helpers:38            output = json.loads(self.openstack(39                'network l3 conntrack helper create %(router)s '40                '--helper %(helper)s --protocol %(protocol)s --port %(port)s '41                '-f json' % {'router': router_id,42                             'helper': helper['helper'],43                             'protocol': helper['protocol'],44                             'port': helper['port']}))45            self.assertEqual(helper['helper'], output['helper'])46            self.assertEqual(helper['protocol'], output['protocol'])47            self.assertEqual(helper['port'], output['port'])48            created_helpers.append(output)49        return created_helpers50    def test_l3_conntrack_helper_create_and_delete(self):51        """Test create, delete multiple"""52        helpers = [53            {54                'helper': 'tftp',55                'protocol': 'udp',56                'port': 6957            }, {58                'helper': 'ftp',59                'protocol': 'tcp',60                'port': 2161            }62        ]63        router_id = self._create_router()64        created_helpers = self._create_helpers(router_id, helpers)65        ct_ids = " ".join([ct['id'] for ct in created_helpers])66        raw_output = self.openstack(67            '--debug network l3 conntrack helper delete %(router)s '68            '%(ct_ids)s' % {69                'router': router_id, 'ct_ids': ct_ids})70        self.assertOutput('', raw_output)71    def test_l3_conntrack_helper_list(self):72        helpers = [73            {74                'helper': 'tftp',75                'protocol': 'udp',76                'port': 6977            }, {78                'helper': 'ftp',79                'protocol': 'tcp',80                'port': 2181            }82        ]83        expected_helpers = [84            {85                'Helper': 'tftp',86                'Protocol': 'udp',87                'Port': 6988            }, {89                'Helper': 'ftp',90                'Protocol': 'tcp',91                'Port': 2192            }93        ]94        router_id = self._create_router()95        self._create_helpers(router_id, helpers)96        output = json.loads(self.openstack(97            'network l3 conntrack helper list %s -f json ' % router_id98        ))99        for ct in output:100            self.assertEqual(router_id, ct.pop('Router ID'))101            ct.pop("ID")102            self.assertIn(ct, expected_helpers)103    def test_l3_conntrack_helper_set_and_show(self):104        helper = {105            'helper': 'tftp',106            'protocol': 'udp',107            'port': 69}108        router_id = self._create_router()109        created_helper = self._create_helpers(router_id, [helper])[0]110        output = json.loads(self.openstack(111            'network l3 conntrack helper show %(router_id)s %(ct_id)s '112            '-f json' % {113                'router_id': router_id, 'ct_id': created_helper['id']}))114        self.assertEqual(helper['helper'], output['helper'])115        self.assertEqual(helper['protocol'], output['protocol'])116        self.assertEqual(helper['port'], output['port'])117        raw_output = self.openstack(118            'network l3 conntrack helper set %(router_id)s %(ct_id)s '119            '--port %(port)s ' % {120                'router_id': router_id,121                'ct_id': created_helper['id'],122                'port': helper['port'] + 1})...

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.