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

How to use create_resource method in localstack

Best Python code snippet using localstack_python

router.py

Source:router.py

...39class APIRouter(cinder.api.openstack.APIRouter):40    """Routes requests on the API to the appropriate controller and method."""41    ExtensionManager = extensions.ExtensionManager42    def _setup_routes(self, mapper, ext_mgr):43        self.resources['versions'] = versions.create_resource()44        mapper.connect("versions", "/",45                       controller=self.resources['versions'],46                       action='index')47        mapper.redirect("", "/")48        self.resources['volumes'] = volumes.create_resource(ext_mgr)49        mapper.resource("volume", "volumes",50                        controller=self.resources['volumes'],51                        collection={'detail': 'GET', 'summary': 'GET'},52                        member={'action': 'POST'})53        self.resources['messages'] = messages.create_resource(ext_mgr)54        mapper.resource("message", "messages",55                        controller=self.resources['messages'],56                        collection={'detail': 'GET'})57        self.resources['clusters'] = clusters.create_resource()58        mapper.resource('cluster', 'clusters',59                        controller=self.resources['clusters'],60                        collection={'detail': 'GET'})61        self.resources['types'] = types.create_resource()62        mapper.resource("type", "types",63                        controller=self.resources['types'],64                        member={'action': 'POST'})65        self.resources['group_types'] = group_types.create_resource()66        mapper.resource("group_type", "group_types",67                        controller=self.resources['group_types'],68                        member={'action': 'POST'})69        self.resources['group_specs'] = group_specs.create_resource()70        mapper.resource("group_spec", "group_specs",71                        controller=self.resources['group_specs'],72                        parent_resource=dict(member_name='group_type',73                                             collection_name='group_types'))74        self.resources['groups'] = groups.create_resource()75        mapper.resource("group", "groups",76                        controller=self.resources['groups'],77                        collection={'detail': 'GET'},78                        member={'action': 'POST'})79        mapper.connect("groups",80                       "/{project_id}/groups/{id}/action",81                       controller=self.resources["groups"],82                       action="action",83                       conditions={"action": ["POST"]})84        mapper.connect("groups/action",85                       "/{project_id}/groups/action",86                       controller=self.resources["groups"],87                       action="action",88                       conditions={"action": ["POST"]})89        self.resources['group_snapshots'] = (group_snapshots.create_resource())90        mapper.resource("group_snapshot", "group_snapshots",91                        controller=self.resources['group_snapshots'],92                        collection={'detail': 'GET'},93                        member={'action': 'POST'})94        self.resources['snapshots'] = snapshots.create_resource(ext_mgr)95        mapper.resource("snapshot", "snapshots",96                        controller=self.resources['snapshots'],97                        collection={'detail': 'GET'},98                        member={'action': 'POST'})99        self.resources['limits'] = limits.create_resource()100        mapper.resource("limit", "limits",101                        controller=self.resources['limits'])102        self.resources['snapshot_metadata'] = \103            snapshot_metadata.create_resource()104        snapshot_metadata_controller = self.resources['snapshot_metadata']105        mapper.resource("snapshot_metadata", "metadata",106                        controller=snapshot_metadata_controller,107                        parent_resource=dict(member_name='snapshot',108                                             collection_name='snapshots'))109        mapper.connect("metadata",110                       "/{project_id}/snapshots/{snapshot_id}/metadata",111                       controller=snapshot_metadata_controller,112                       action='update_all',113                       conditions={"method": ['PUT']})114        self.resources['volume_metadata'] = \115            volume_metadata.create_resource()116        volume_metadata_controller = self.resources['volume_metadata']117        mapper.resource("volume_metadata", "metadata",118                        controller=volume_metadata_controller,119                        parent_resource=dict(member_name='volume',120                                             collection_name='volumes'))121        mapper.connect("metadata",122                       "/{project_id}/volumes/{volume_id}/metadata",123                       controller=volume_metadata_controller,124                       action='update_all',125                       conditions={"method": ['PUT']})126        self.resources['consistencygroups'] = (127            consistencygroups.create_resource())128        mapper.resource("consistencygroup", "consistencygroups",129                        controller=self.resources['consistencygroups'],130                        collection={'detail': 'GET'},131                        member={'action': 'POST'})132        self.resources['manageable_volumes'] = volume_manage.create_resource()133        mapper.resource("manageable_volume", "manageable_volumes",134                        controller=self.resources['manageable_volumes'],135                        collection={'detail': 'GET'})136        self.resources['manageable_snapshots'] = \137            snapshot_manage.create_resource()138        mapper.resource("manageable_snapshot", "manageable_snapshots",139                        controller=self.resources['manageable_snapshots'],140                        collection={'detail': 'GET'})141        self.resources['backups'] = (142            backups.create_resource())143        mapper.resource("backup", "backups",144                        controller=self.resources['backups'],...

test_query_resources_utils.py

Source:test_query_resources_utils.py

...24        )25        self.auxtestResource(26            compressed_resource=compressed_entity,27            decompressed_resource=decompressed_entity,28            resource_1=Resource.create_resource(compressed_entity),29            resource_2=Resource.create_resource(decompressed_entity)30        )31    def testWikidataProperty(self):32        compressed_property = "wdt:P184"33        decompressed_property = "<http://www.wikidata.org/prop/direct/P184>"34        self.auxtestResource(35            compressed_resource=compressed_property,36            decompressed_resource=decompressed_property,37            resource_1=WikidataDirectProperty(compressed_property),38            resource_2=WikidataDirectProperty(decompressed_property)39        )40        self.auxtestResource(41            compressed_resource=compressed_property,42            decompressed_resource=decompressed_property,43            resource_1=Resource.create_resource(compressed_property),44            resource_2=Resource.create_resource(decompressed_property)45        )46    def testDBpediaEntity(self):47        compressed_resource = "dbr:Grand_Prix_(Cannes_Film_Festival)"48        decompressed_resource = "<http://dbpedia.org/resource/Grand_Prix_(Cannes_Film_Festival)>"49        self.auxtestResource(50            compressed_resource=compressed_resource,51            decompressed_resource=decompressed_resource,52            resource_1=DBpediaEntity(compressed_resource),53            resource_2=DBpediaEntity(decompressed_resource)54        )55        self.auxtestResource(56            compressed_resource=compressed_resource,57            decompressed_resource=decompressed_resource,58            resource_1=Resource.create_resource(compressed_resource),59            resource_2=Resource.create_resource(decompressed_resource)60        )61    @unittest.expectedFailure62    def testDBpediaAlternativeEntity(self):63        compressed_resource = "res:Game_of_Thrones"64        decompressed_resource = "<http://dbpedia.org/resource/Game_of_Thrones>"65        self.auxtestResource(66            compressed_resource=compressed_resource,67            decompressed_resource=decompressed_resource,68            resource_1=DBpediaAlternativeEntity(compressed_resource),69            resource_2=DBpediaAlternativeEntity(decompressed_resource)70        )71        self.auxtestResource(72            compressed_resource=compressed_resource,73            decompressed_resource=decompressed_resource,74            resource_1=Resource.create_resource(compressed_resource),75            resource_2=Resource.create_resource(decompressed_resource)76        )77    def testDBpediaProperty(self):78        compressed_property = "dbp:equivalentProperty"79        decompressed_property = "<http://dbpedia.org/property/equivalentProperty>"80        self.auxtestResource(81            compressed_resource=compressed_property,82            decompressed_resource=decompressed_property,83            resource_1=DBpediaProperty(compressed_property),84            resource_2=DBpediaProperty(decompressed_property)85        )86        self.auxtestResource(87            compressed_resource=compressed_property,88            decompressed_resource=decompressed_property,89            resource_1=Resource.create_resource(compressed_property),90            resource_2=Resource.create_resource(decompressed_property)91        )92    def testDBpediaOntology(self):93        compressed_ontology = "dbo:award"94        decompressed_ontology = "<http://dbpedia.org/ontology/award>"95        self.auxtestResource(96            compressed_resource=compressed_ontology,97            decompressed_resource=decompressed_ontology,98            resource_1=DBpediaOntology(compressed_ontology),99            resource_2=DBpediaOntology(decompressed_ontology)100        )101        self.auxtestResource(102            compressed_resource=compressed_ontology,103            decompressed_resource=decompressed_ontology,104            resource_1=Resource.create_resource(compressed_ontology),105            resource_2=Resource.create_resource(decompressed_ontology)106        )107    def testResourceEqual(self):108        compressed_entity = Resource.create_resource("wd:Q4072104")109        decompressed_entity = Resource.create_resource("<http://www.wikidata.org/entity/Q4072104>")110        entity_set = {compressed_entity, decompressed_entity}111        self.assertTrue(len(entity_set) == 1)112        resource = entity_set.pop()113        self.assertEqual(str(compressed_entity), str(resource))114        self.assertEqual(str(decompressed_entity), str(resource))115    def testIsWikidata(self):116        wikidata_entity = WikidataEntity("wd:Q4072104")117        wikidata_property = WikidataDirectProperty("wdt:P184")118        self.assertTrue(wikidata_entity.is_wikidata())119        self.assertTrue(wikidata_property.is_wikidata())120        self.assertFalse(wikidata_entity.is_dbpedia())121        self.assertFalse(wikidata_property.is_dbpedia())122    def testIsDBpedia(self):123        dbpedia_resource = DBpediaEntity("dbr:Grand_Prix_(Cannes_Film_Festival)")...

test_strict_compliance.py

Source:test_strict_compliance.py

...11            extra_config_knobs=[('NEUTRON', 'strict_compliance', True)])12    # end setUpClass13    def _create_floatingip_and_associate_port_without_ext_gw(self, proj_id):14        # external network15        net_q = self.create_resource(16            'network', proj_id, extra_res_fields={17                'router:external': True})18        self.create_resource(19            'subnet',20            proj_id,21            extra_res_fields={22                'network_id': net_q['id'],23                'cidr': '10.2.0.0/24',24                'ip_version': 4})25        # private network26        pvt_net_q = self.create_resource('network', proj_id)27        pvt_subnet_q = self.create_resource(28            'subnet',29            proj_id,30            extra_res_fields={31                'network_id': pvt_net_q['id'],32                'cidr': '10.1.0.0/24',33                'ip_version': 4})34        port_q = self.create_resource(35            'port', proj_id, extra_res_fields={36                'network_id': pvt_subnet_q['network_id']})37        return self.create_resource(38            'floatingip',39            proj_id,40            extra_res_fields={41                'floating_network_id': net_q['id'],42                'port_id': port_q['id']})43    def _create_floatingip_and_associate_port_with_ext_gw(self, proj_id):44        # external network45        net_q = self.create_resource(46            'network', proj_id, extra_res_fields={47                'router:external': True})48        self.create_resource(49            'subnet',50            proj_id,51            extra_res_fields={52                'network_id': net_q['id'],53                'cidr': '10.2.0.0/24',54                'ip_version': 4})55        router_q = self.create_resource('router', proj_id)56        # private network57        pvt_net_q = self.create_resource('network', proj_id)58        pvt_subnet_q = self.create_resource(59            'subnet',60            proj_id,61            extra_res_fields={62                'network_id': pvt_net_q['id'],63                'cidr': '10.1.0.0/24',64                'ip_version': 4})65        port_q = self.create_resource(66            'port', proj_id, extra_res_fields={67                'network_id': pvt_subnet_q['network_id']})68        port2_q = self.create_resource(69            'port', proj_id, extra_res_fields={70                'network_id': pvt_subnet_q['network_id']})71        # External gateway72        router_q = self.update_resource(73            'router', router_q['id'], proj_id, extra_res_fields={74                'external_gateway_info': {75                    'network_id': net_q['id']}})76        router_q = self.add_router_interface(77            router_q['id'], proj_id, extra_res_fields={78                'port_id': port2_q['id']})79        return self.create_resource(80            'floatingip',81            proj_id,82            extra_res_fields={83                'floating_network_id': net_q['id'],84                'port_id': port_q['id']})85    # test when strict_compliance is ON86    def test_create_fip_and_associate_port_without_ext_gw(self):87        proj_obj = self._vnc_lib.project_read(88            fq_name=['default-domain', 'default-project'])89        res_q = self.create_resource('security_group', proj_obj.uuid)90        self.list_resource(91            'security_group',92            proj_uuid=proj_obj.uuid,93            req_filters={94                'name': res_q['name']})95        with ExpectedException(webtest.app.AppError):96            self._create_floatingip_and_associate_port_without_ext_gw(97                proj_obj.uuid)98    # test when strict_compliance is ON99    def test_create_fip_and_associate_port_with_ext_gw(self):100        proj_obj = self._vnc_lib.project_read(101            fq_name=['default-domain', 'default-project'])102        res_q = self.create_resource('security_group', proj_obj.uuid)103        self.list_resource(104            'security_group',105            proj_uuid=proj_obj.uuid,106            req_filters={107                'name': res_q['name']})108        self._create_floatingip_and_associate_port_with_ext_gw(proj_obj.uuid)109# end class TestStrictCompON110class TestStrictCompOff(test_case.NeutronBackendTestCase):111    @classmethod112    def setUpClass(cls):113        super(TestStrictCompOff, cls).setUpClass(114            extra_config_knobs=[('NEUTRON', 'strict_compliance', False)])115    # end setUpClass116    def _create_floatingip_and_associate_port_without_ext_gw(self, proj_id):117        # external network118        net_q = self.create_resource(119            'network', proj_id, extra_res_fields={120                'router:external': True})121        self.create_resource(122            'subnet',123            proj_id,124            extra_res_fields={125                'network_id': net_q['id'],126                'cidr': '10.2.0.0/24',127                'ip_version': 4})128        # private network129        pvt_net_q = self.create_resource('network', proj_id)130        pvt_subnet_q = self.create_resource(131            'subnet',132            proj_id,133            extra_res_fields={134                'network_id': pvt_net_q['id'],135                'cidr': '10.1.0.0/24',136                'ip_version': 4})137        port_q = self.create_resource(138            'port', proj_id, extra_res_fields={139                'network_id': pvt_subnet_q['network_id']})140        return self.create_resource(141            'floatingip',142            proj_id,143            extra_res_fields={144                'floating_network_id': net_q['id'],145                'port_id': port_q['id']})146    # test when strict_compliance is OFF147    def test_create_fip_and_associate_port_without_ext_gw(self):148        proj_obj = self._vnc_lib.project_read(149            fq_name=['default-domain', 'default-project'])150        res_q = self.create_resource('security_group', proj_obj.uuid)151        self.list_resource(152            'security_group',153            proj_uuid=proj_obj.uuid,154            req_filters={155                'name': res_q['name']})156        self._create_floatingip_and_associate_port_without_ext_gw(157            proj_obj.uuid)...

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.