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

How to use validate_collection method in Slash

Best Python code snippet using slash

test_database.py

Source:test_database.py

...95        await test_db.drop_collection('test')96        assert 'test' not in (await test_db.collection_names())97        await test_db.drop_collection(test_db.test.doesnotexist)98    @pytest.mark.asyncio99    async def test_validate_collection(self, test_db):100        with pytest.raises(TypeError):101            await test_db.validate_collection(5)102        with pytest.raises(TypeError):103            await test_db.validate_collection(None)104        await test_db.test.insert_one({'dummy': 'object'})105        with pytest.raises(OperationFailure):106            await test_db.validate_collection('test.doesnotexist"')107        with pytest.raises(OperationFailure):108            await test_db.validate_collection(test_db.test.doesnotexist)109        assert await test_db.validate_collection('test')110        assert await test_db.validate_collection(test_db.test)111        assert await test_db.validate_collection(test_db.test, full=True)112        assert await test_db.validate_collection(test_db.test, scandata=True)113        assert await test_db.validate_collection(test_db.test, full=True, scandata=True)114        assert await test_db.validate_collection(test_db.test, True, True)115    @pytest.mark.asyncio116    async def test_profiling_levels(self, mongo, test_db):117        connection = await mongo.get_connection()118        if connection.is_mongos:119            pytest.skip('Profiling works only without mongos.')120            return121        assert await test_db.profiling_level() == OFF  # default122        with pytest.raises(ValueError):123            await test_db.set_profiling_level(5.5)124        with pytest.raises(ValueError):125            await test_db.set_profiling_level(None)126        with pytest.raises(ValueError):127            await test_db.set_profiling_level(-1)128        with pytest.raises(TypeError):...

crud.py

Source:crud.py

...59        api_response.total = 160    else:61        raise EntityNotFoundException(f'Collection id {params.id} does not exist')62def get_items_per_collection(params: GETCollectionItems, api_response: APIResponse):63    validate_collection(collection_id=params.id)64    try:65        custom_sort = getattr(ItemModel, params.sorting).asc()66        if params.order == 'desc':67            custom_sort = getattr(ItemModel, params.sorting).desc()68    except Exception:69        raise BadRequestException(f'Cannot sort by {params.sorting}')70    item_query = (71        db.session.query(ItemModel, StorageModel, ExtendedModel).join(StorageModel, ExtendedModel,72                                                                      ItemsCollectionsModel)73        .filter(ItemsCollectionsModel.collection_id == params.id, ItemModel.archived == params.archived)74        .order_by(ItemModel.type, custom_sort)75    )76    paginate(params, api_response, item_query, combine_item_tables)77def create_collection(data: POSTCollection, api_response: APIResponse):78    collection_query = (79        db.session.query(CollectionsModel).filter(CollectionsModel.owner == data.owner,80                                                  CollectionsModel.container_code == data.container_code)81    )82    collection_result = collection_query.all()83    if len(collection_result) == ConfigClass.MAX_COLLECTIONS:84        raise BadRequestException(f'Cannot create more than {ConfigClass.MAX_COLLECTIONS} collections')85    elif data.name in (collection.name for collection in collection_result):86        raise DuplicateRecordException(f'Collection {data.name} already exists')87    else:88        model_data = {'id': data.id, 'owner': data.owner, 'container_code': data.container_code,89                      'name': data.name}90        collection = CollectionsModel(**model_data)91        db.session.add(collection)92        db.session.commit()93        db.session.refresh(collection)94        api_response.result = collection.to_dict()95        api_response.total = 196def update_collection(data: PUTCollections, api_response: APIResponse):97    collections = {c.id: c.name for c in data.collections}98    query = (99        db.session.query(CollectionsModel).filter(CollectionsModel.owner == data.owner,100                                                  CollectionsModel.container_code == data.container_code)101    )102    query_result = query.all()103    exist_collections = {c.id: c.name for c in query_result}104    for c_id in collections:105        if c_id not in exist_collections:106            raise EntityNotFoundException(f'Collection id: {c_id} does not exist')107        elif collections[c_id] in exist_collections.values():108            raise DuplicateRecordException(f'Collection name: {collections[c_id]} already exists')109    for collection in data.collections:110        db.session.merge(CollectionsModel(id=collection.id, name=collection.name,111                                          container_code=data.container_code, owner=data.owner,112                                          last_updated_time=datetime.utcnow()))113        db.session.commit()114    result = json.loads(data.json())115    api_response.result = result116    api_response.total = len(data.collections)117def add_items(data: POSTCollectionItems, api_response: APIResponse):118    validate_collection(collection_id=data.id)119    # foreign key constraint ensures collection_id and item_id must exist in collection and items tables120    for item_id in data.item_ids:121        db.session.merge(ItemsCollectionsModel(collection_id=data.id, item_id=item_id))122    db.session.commit()123    result = json.loads(data.json())124    api_response.result = result125    api_response.total = len(data.item_ids)126def remove_items(data: DELETECollectionItems):127    validate_collection(collection_id=data.id)128    # foreign key constraint ensures collection_id and item_id must exist in collection and items tables129    db.session.query(ItemsCollectionsModel).filter(ItemsCollectionsModel.collection_id == data.id,130                                                   ItemsCollectionsModel.item_id.in_(data.item_ids)).delete()131    db.session.commit()132def remove_collection(collection_id: UUID, api_response: APIResponse):133    validate_collection(collection_id=collection_id)134    db.session.query(CollectionsModel).filter(CollectionsModel.id == collection_id).delete()135    db.session.commit()136    api_response.total = 0...

MongoDBManager.py

Source:MongoDBManager.py

...39            return True40        except ConnectionFailure as e:41            raise MongoDBException("Server not available. {0}".format(e.message))42# Method to Check if collection exist in database43    def validate_collection(self, collection_name, database_name):44        database = self.client[database_name]45        collection_names = database.collection_names()46        if collection_name in collection_names:47            return True48        raise MongoDBException("Collection {0} NOT exist in {1} database. Please try again.".format(collection_name, database_name))49# Method to Check if specific database exist50    def validate_database(self, database_name):51        databases = self.client.database_names()52        if database_name in databases:53            return True54        raise MongoDBException("Database {0} NOT exist. Please try again.".format(database_name))55# Method to Query MongoDB for documents.56    def execute_query(self, query, database_, collection_):57        try:58            data = []59            60            # Check if Authentication required61            if self.use_auth:62                client = self.authenticate(database_)63                64                self.validate_database(database_)65                database = client[database_]66                67                # Check if collection exist in db68                self.validate_collection(collection_, database_)69                collection = database[collection_]70                71            else:72                # Check if database exist73                self.validate_database(database_)74                database = self.client[database_]75    76                # Check if collection exist in db77                self.validate_collection(collection_, database_)78                collection = database[collection_]79            # Query the database. Returns a Cursor instance80            results = collection.find(query)81            # doc = record82            for doc in results:83                for field_value in doc:84                    # ObjectId is not JSON serializable85                    if isinstance(doc[field_value], ObjectId):86                        doc[field_value] = str(doc[field_value])87                data.append(doc)88            return data89        except Exception as e:90            raise MongoDBException(e)91# Method to insert documents into MongoDB.92    def insert_documents(self, database, collection, document):93        94        # Check if Authentication required95        if self.use_auth:96            conn = self.authenticate(database)97            db = conn[database]98            99            val_db = self.validate_database(database)100            val_col = self.validate_collection(collection, database)101            102        else:103            conn = self.client104            db = conn[database]105            106            val_db = self.validate_database(database)107            val_col = self.validate_collection(collection, database)108            109        if val_db and val_col:110            try:111                if isinstance(document, list):112                    add_method = db[collection].insert_many113                    res = add_method(document)114                    ids = res.inserted_ids115                    return res, ids116                elif isinstance(document, dict):117                    add_method = db[collection].insert_one118                    res = add_method(document)119                    ids = res.inserted_id120                    return res, ids121                else:122                    raise MongoDBException("Input data must be either dict or list")123            124            except Exception as err:125                raise MongoDBException(err)126                127                128# Method to update documents in MongoDB.129    def update_documents(self, database, collection, query, document):130        # Check if Authentication required131        if self.use_auth:132            conn = self.authenticate(database)133            db = conn[database]134            135            val_db = self.validate_database(database)136            val_col = self.validate_collection(collection, database)137            138        else:139            conn = self.client140            db = conn[database]141            142            val_db = self.validate_database(database)143            val_col = self.validate_collection(collection, database)144        145        if val_db and val_col:146            try:147                result = db[collection].update_many(query, document)148                return result149            except Exception as err:150                raise MongoDBException(err)151# Method to delete documents in MongoDB.152    def delete_documents(self, database, collection, query):153        # Check if Authentication required154        if self.use_auth:155            conn = self.authenticate(database)156            db = conn[database]157            158            val_db = self.validate_database(database)159            val_col = self.validate_collection(collection, database)160            161        else:162            conn = self.client163            db = conn[database]164            165            val_db = self.validate_database(database)166            val_col = self.validate_collection(collection, database)167        168        if val_db and val_col:169            try:170                result = db[collection].delete_many(query)171                return result172            except Exception as err:173                raise MongoDBException(err)174    def close_connection(self):175        """176        Close the connection177        """...

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.