How to use required_collections method in molecule

Best Python code snippet using molecule_python

test_database.py

Source:test_database.py

1import pytest2import os3from database import Database4class TestDatabase():5    # This scenario initializes the database, so run it first.6    @pytest.mark.order(1)7    def test_create_file(self):8        try:9            db = Database(init=True)10        except Exception as e:11            assert False, f"'test_create_file' raised an exception: {e}"12    def test_create_connection_failed(self):13        with pytest.raises(RuntimeError):14            Database(file='tests/resources/dbconfig_connection_failed.json')15    def test_create_file_not_found(self):16        with pytest.raises(FileNotFoundError):17            Database(file='does-not-exist')18    19    def test_create_invalid_data(self):20        with pytest.raises(ValueError):21            Database(file='tests/resources/dbconfig_invalid_data.json')22    def test_create_malformed_json(self):23        with pytest.raises(Exception):24            Database(file='tests/resources/dbconfig_malformed.json')25    def test_create_missing_data(self):26        with pytest.raises(ValueError):27            Database(file='tests/resources/dbconfig_missing_data.json')28    def test_create_missing_key(self):29        with pytest.raises(KeyError):30            Database(file='tests/resources/dbconfig_missing_keys.json')31    def test_create_nonexistent_db(self):32        with pytest.raises(KeyError):33            Database(file='tests/resources/dbconfig_nonexistent_db.json')34    def test_collections_exist(self):35        db = Database()36        required_collections = ['Redemptions', 'Members', 'Settings', 'Awards']...

__init__.py

Source:__init__.py

...3import pymongo, redis4from flask import Flask5from flask_socketio import SocketIO6from flask_login import LoginManager7def create_required_collections(required_collections, database_name="flashdb"):8    current_collections_in_db = mongo_client[database_name].list_collection_names()9    print("database_name :: ", database_name)10    for collection in required_collections:11        if collection not in current_collections_in_db:12            test_data = { "initialiser_data": "TEST" }13            db_collection = mongo_client[database_name][collection]14            inserted_data = db_collection.insert_one(test_data)15            if inserted_data is not None:16                print(f"Collection '{collection}' is created")17        else:18            print(f"Collection '{collection}' was already present")19def create_app(debug=True):20    """Create the app."""21    app = Flask(__name__)22    app.debug = debug23    app.config['SECRET_KEY'] = 'gjr39dkjn344_!67#'24    from .main import main as main_blueprint25    app.register_blueprint(main_blueprint)26    socketio.init_app(app)27    return app28redis_client = redis.Redis()29# redis_pubsub = redis_client.pubsub()30print(redis_client.ping())31socketio = SocketIO()32# Establish db connection.33mongo_client = pymongo.MongoClient("mongodb://localhost:27017/")34print(mongo_client)35master_flash_db_client = mongo_client["flashdb"] 36print(master_flash_db_client)37print(mongo_client.list_database_names())38required_collections = ["clients", "projects", "test", "clientdb_project_mapping"]...

vamBackgroundCHecker.py

Source:vamBackgroundCHecker.py

1import pymongo2from pymongo import MongoClient3from pymongo import errors4import os5import sys6from colorama import Fore7project_path = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))8sys.path.append(project_path)9class BotStructureCheck:10    def __init__(self):11        self.connection = MongoClient("mongodb://localhost:27017")12        self.vam = self.connection["VirtualAirMarshal"]13        self.required_collections = ["jailedMembers",  # Collection storing all jailled members14                                     "jailProfiles",  # Collection storing all communities applied for jair service15                                     "spamProfiles",  # Collection storing all communities applied for spam preventions16                                     "supportProfiles",  # Collection storing all communities applied for support system17                                     "loggerSystem"  # Collection storing logger data18                                     ]19        20    def check_collections(self):21        """ Check bots backend on load"""22        bot_collections = self.vam.list_collection_names()23        print(Fore.GREEN + "Checking Virtual Air Marshal backend integrity")24        for collection in self.required_collections:25            if collection not in bot_collections:26                self.vam.create_collection(collection)27                print(Fore.YELLOW + f"{collection.upper()} has been created!")28            else:29                print(Fore.GREEN + f'{collection.upper()} already exists')...

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.