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

How to use create_mapping method in tempest

Best Python code snippet using tempest_python

test_instance_mapping.py

Source:test_instance_mapping.py

...31        args['uuid'] = uuidutils.generate_uuid()32    args.update(kwargs)33    ctxt = context.RequestContext('fake-user', 'fake-project')34    return cell_mapping.CellMapping._create_in_db(ctxt, args)35def create_mapping(**kwargs):36    args = sample_mapping.copy()37    if 'instance_uuid' not in kwargs:38        args['instance_uuid'] = uuidutils.generate_uuid()39    args.update(kwargs)40    ctxt = context.RequestContext('fake-user', 'fake-project')41    return instance_mapping.InstanceMapping._create_in_db(ctxt, args)42class InstanceMappingTestCase(test.NoDBTestCase):43    USES_DB_SELF = True44    def setUp(self):45        super(InstanceMappingTestCase, self).setUp()46        self.useFixture(fixtures.Database(database='api'))47        self.context = context.RequestContext('fake-user', 'fake-project')48        self.mapping_obj = instance_mapping.InstanceMapping()49    def test_get_by_instance_uuid(self):50        cell_mapping = create_cell_mapping()51        mapping = create_mapping()52        db_mapping = self.mapping_obj._get_by_instance_uuid_from_db(53                self.context, mapping['instance_uuid'])54        for key in [key for key in self.mapping_obj.fields.keys()55                    if key != 'cell_mapping']:56            self.assertEqual(db_mapping[key], mapping[key])57        self.assertEqual(db_mapping['cell_mapping']['id'], cell_mapping['id'])58    def test_get_by_instance_uuid_not_found(self):59        self.assertRaises(exception.InstanceMappingNotFound,60                self.mapping_obj._get_by_instance_uuid_from_db, self.context,61                uuidutils.generate_uuid())62    def test_save_in_db(self):63        mapping = create_mapping()64        cell_mapping = create_cell_mapping()65        self.mapping_obj._save_in_db(self.context, mapping['instance_uuid'],66                {'cell_id': cell_mapping['id']})67        db_mapping = self.mapping_obj._get_by_instance_uuid_from_db(68                self.context, mapping['instance_uuid'])69        for key in [key for key in self.mapping_obj.fields.keys()70                    if key not in ['cell_id', 'cell_mapping', 'updated_at']]:71            self.assertEqual(db_mapping[key], mapping[key])72        self.assertEqual(db_mapping['cell_id'], cell_mapping['id'])73    def test_destroy_in_db(self):74        mapping = create_mapping()75        self.mapping_obj._get_by_instance_uuid_from_db(self.context,76                mapping['instance_uuid'])77        self.mapping_obj._destroy_in_db(self.context, mapping['instance_uuid'])78        self.assertRaises(exception.InstanceMappingNotFound,79                self.mapping_obj._get_by_instance_uuid_from_db, self.context,80                mapping['instance_uuid'])81    def test_cell_id_nullable(self):82        # Just ensure this doesn't raise83        create_mapping(cell_id=None)84    def test_modify_cell_mapping(self):85        inst_mapping = instance_mapping.InstanceMapping(context=self.context)86        inst_mapping.instance_uuid = uuidutils.generate_uuid()87        inst_mapping.project_id = self.context.project_id88        inst_mapping.cell_mapping = None89        inst_mapping.create()90        c_mapping = cell_mapping.CellMapping(91                self.context,92                uuid=uuidutils.generate_uuid(),93                name="cell0",94                transport_url="none:///",95                database_connection="fake:///")96        c_mapping.create()97        inst_mapping.cell_mapping = c_mapping98        inst_mapping.save()99        result_mapping = instance_mapping.InstanceMapping.get_by_instance_uuid(100                                    self.context, inst_mapping.instance_uuid)101        self.assertEqual(result_mapping.cell_mapping.id,102                         c_mapping.id)103class InstanceMappingListTestCase(test.NoDBTestCase):104    USES_DB_SELF = True105    def setUp(self):106        super(InstanceMappingListTestCase, self).setUp()107        self.useFixture(fixtures.Database(database='api'))108        self.context = context.RequestContext('fake-user', 'fake-project')109        self.list_obj = instance_mapping.InstanceMappingList()110    def test_get_by_project_id_from_db(self):111        project_id = 'fake-project'112        mappings = {}113        mapping = create_mapping(project_id=project_id)114        mappings[mapping['instance_uuid']] = mapping115        mapping = create_mapping(project_id=project_id)116        mappings[mapping['instance_uuid']] = mapping117        db_mappings = self.list_obj._get_by_project_id_from_db(118                self.context, project_id)119        for db_mapping in db_mappings:120            mapping = mappings[db_mapping.instance_uuid]121            for key in instance_mapping.InstanceMapping.fields.keys():122                self.assertEqual(db_mapping[key], mapping[key])123    def test_instance_mapping_list_get_by_cell_id(self):124        """Tests getting all of the InstanceMappings for a given CellMapping id125        """126        # we shouldn't have any instance mappings yet127        inst_mapping_list = (128            instance_mapping.InstanceMappingList.get_by_cell_id(129                self.context, sample_cell_mapping['id'])130        )131        self.assertEqual(0, len(inst_mapping_list))132        # now create an instance mapping in a cell133        db_inst_mapping1 = create_mapping()134        # let's also create an instance mapping that's not in a cell to make135        # sure our filtering is working136        db_inst_mapping2 = create_mapping(cell_id=None)137        self.assertIsNone(db_inst_mapping2['cell_id'])138        # now we should list out one instance mapping for the cell139        inst_mapping_list = (140            instance_mapping.InstanceMappingList.get_by_cell_id(141                self.context, db_inst_mapping1['cell_id'])142        )143        self.assertEqual(1, len(inst_mapping_list))144        self.assertEqual(db_inst_mapping1['id'], inst_mapping_list[0].id)145    def test_instance_mapping_get_by_instance_uuids(self):146        db_inst_mapping1 = create_mapping()147        db_inst_mapping2 = create_mapping(cell_id=None)148        # Create a third that we won't include149        create_mapping()150        uuids = [db_inst_mapping1.instance_uuid,151                 db_inst_mapping2.instance_uuid]152        mappings = instance_mapping.InstanceMappingList.get_by_instance_uuids(153            self.context, uuids + [uuidsentinel.deleted_instance])154        self.assertEqual(sorted(uuids),...

test_cell_mapping.py

Source:test_cell_mapping.py

...19SAMPLE_MAPPING = {'uuid': '',20                  'name': 'fake-cell',21                  'transport_url': 'rabbit:///',22                  'database_connection': 'mysql+pymysql:///'}23def create_mapping(**kwargs):24    args = SAMPLE_MAPPING.copy()25    if 'uuid' not in kwargs:26        args['uuid'] = uuidutils.generate_uuid()27    args.update(kwargs)28    ctxt = context.RequestContext()29    return cell_mapping.CellMapping._create_in_db(ctxt, args)30class CellMappingTestCase(test.NoDBTestCase):31    USES_DB_SELF = True32    def setUp(self):33        super(CellMappingTestCase, self).setUp()34        self.useFixture(fixtures.Database(database='api'))35        self.context = context.RequestContext('fake-user', 'fake-project')36        self.mapping_obj = cell_mapping.CellMapping()37    def test_get_by_uuid(self):38        mapping = create_mapping()39        db_mapping = self.mapping_obj._get_by_uuid_from_db(self.context,40                mapping['uuid'])41        for key in self.mapping_obj.fields.keys():42            self.assertEqual(db_mapping[key], mapping[key])43    def test_get_by_uuid_not_found(self):44        self.assertRaises(exception.CellMappingNotFound,45                self.mapping_obj._get_by_uuid_from_db, self.context,46                          uuidutils.generate_uuid())47    def test_save_in_db(self):48        mapping = create_mapping()49        self.mapping_obj._save_in_db(self.context, mapping['uuid'],50                {'name': 'meow'})51        db_mapping = self.mapping_obj._get_by_uuid_from_db(self.context,52                mapping['uuid'])53        self.assertNotEqual(db_mapping['name'], mapping['name'])54        for key in [key for key in self.mapping_obj.fields.keys()55                    if key not in ['name', 'updated_at']]:56            self.assertEqual(db_mapping[key], mapping[key])57    def test_destroy_in_db(self):58        mapping = create_mapping()59        self.mapping_obj._get_by_uuid_from_db(self.context, mapping['uuid'])60        self.mapping_obj._destroy_in_db(self.context, mapping['uuid'])61        self.assertRaises(exception.CellMappingNotFound,62                self.mapping_obj._get_by_uuid_from_db, self.context,63                mapping['uuid'])64    def test_destroy_in_db_not_found(self):65        self.assertRaises(exception.CellMappingNotFound,66                self.mapping_obj._destroy_in_db, self.context,67                uuidutils.generate_uuid())68class CellMappingListTestCase(test.NoDBTestCase):69    USES_DB_SELF = True70    def setUp(self):71        super(CellMappingListTestCase, self).setUp()72        self.useFixture(fixtures.Database(database='api'))73    def test_get_all(self):74        mappings = {}75        mapping = create_mapping()76        mappings[mapping['uuid']] = mapping77        mapping = create_mapping()78        mappings[mapping['uuid']] = mapping79        ctxt = context.RequestContext()80        db_mappings = cell_mapping.CellMappingList._get_all_from_db(ctxt)81        for db_mapping in db_mappings:82            mapping = mappings[db_mapping.uuid]83            for key in cell_mapping.CellMapping.fields.keys():84                self.assertEqual(db_mapping[key], mapping[key])85    def test_get_by_disabled(self):86        enabled_mapping = create_mapping(disabled=False)87        disabled_mapping = create_mapping(disabled=True)88        ctxt = context.RequestContext()89        mappings = cell_mapping.CellMappingList.get_all(ctxt)90        self.assertEqual(2, len(mappings))91        self.assertEqual(enabled_mapping['uuid'], mappings[0].uuid)92        self.assertEqual(disabled_mapping['uuid'], mappings[1].uuid)93        mappings = cell_mapping.CellMappingList.get_by_disabled(ctxt,94                                                                disabled=False)95        self.assertEqual(1, len(mappings))96        self.assertEqual(enabled_mapping['uuid'], mappings[0].uuid)97        mappings = cell_mapping.CellMappingList.get_by_disabled(ctxt,98                                                                disabled=True)99        self.assertEqual(1, len(mappings))100        self.assertEqual(disabled_mapping['uuid'], mappings[0].uuid)101    def test_get_by_project_id(self):102        ctxt = context.RequestContext()103        cell1 = objects.CellMapping.get_by_uuid(ctxt, create_mapping().uuid)104        cell2 = objects.CellMapping.get_by_uuid(ctxt, create_mapping().uuid)105        cell3 = objects.CellMapping.get_by_uuid(ctxt, create_mapping().uuid)106        cells = [cell1, cell2, cell3]107        # Proj1 is all in one cell108        for i in range(0, 5):109            uuid = uuidutils.generate_uuid()110            im = objects.InstanceMapping(context=ctxt,111                                         instance_uuid=uuid,112                                         cell_mapping=cell1,113                                         project_id='proj1')114            im.create()115        # Proj2 is in the first two cells116        for i in range(0, 5):117            uuid = uuidutils.generate_uuid()118            cell = cells[i % 2]119            im = objects.InstanceMapping(context=ctxt,...

elasticsearch_dao.py

Source:elasticsearch_dao.py

...30        actionList = []31        actionCount = 132        # Create mappings33        if create_mapping:34            self.create_mapping(index, type)35        for doc in doc_list:36            item = doc.copy()37            id = item.get('id')38            if actionCount % 1000 == 0:39                self._bulk_insert(index, type, actionList[:])40                actionList = []41            if upsert:42                actionList.append({"index": {43                    "_index": index,44                    "_type": type,45                    "_id": id46                }})47            else:48                actionList.append({"index": {49                    "_index": index,50                    "_type": type51                }})52            actionList.append(item)53            actionCount += 154        if len(actionList) > 0:55            self._bulk_insert(index, type, actionList[:])56    def insert_one(self, doc, index, type, id, upsert=True, create_mapping=True):57        # Create mappings58        if create_mapping:59            self.create_mapping(index, type)60        if not upsert:61            res = self.connection.index(index, type, doc)62        else:63            res = self.connection.index(index, type, doc, id)64        return res65    def create_mapping(self, index, type):66        mapping = {67            "mappings": {68                type: {69                    "properties": {70                        "data": {71                            "properties": {72                                "coordinates": {"type": "geo_point"}73                            }74                        }75                    }76                }77            }78        }79        return self.connection.indices.create(index=index, ignore=400, body=json.dumps(mapping))

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.