KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use _execute_migration method in autotest

Best Python code snippet using autotest_python

migration_manager.py

Source:migration_manager.py

...43            return44        Log.i("Schema version table doesn't exist, creating")45        if len(MigrationManager._migrations) == 0:46            raise Exception("Base migration wasn't found")47        MigrationManager._execute_migration(db, MigrationManager._migrations[0])48    @staticmethod49    def _check_for_failed_migrations(db):50        c = db.cursor()51        c.execute("select id, name from schema_versions where success = 0")52        failed = c.fetchone()53        c.close()54        if failed is not None:55            raise Exception("Database contains a failed migration: {}. {}".format(failed[0], failed[1]))56    @staticmethod57    def _execute_missing_migrations(db):58        existing = [x[0] for x in db.execute("select id from schema_versions")]59        for m in MigrationManager._migrations:60            if m.number not in existing:61                MigrationManager._execute_migration(db, m)62    @staticmethod63    def _load_migration_files():64        migrations = map(lambda x: MigrationFile(x), MigrationManager._migration_file_names)65        migrations = sorted(migrations, key=lambda x: x.number)66        numbers = []67        for m in migrations:68            if m.number in numbers:69                raise Exception(70                    "Error while loading migration {}: Migration with number {} already exists".format(m.name,71                                                                                                       m.number))72        MigrationManager._migrations = migrations73    @staticmethod74    def _execute_migration(db, migration):75        Log.i("Executing migration {}. {}".format(migration.number, migration.name))76        exc = None77        statements = MigrationManager._split_statements(migration.content)78        start = time.time()79        try:80            for statement in statements:81                MigrationManager._execute_statement(db, statement)82        except sqlite3.OperationalError as e:83            Log.e("Error while executing migration {}. {}".format(migration.number, migration.name))84            exc = e85        finally:86            end = time.time()87        execution_time = math.floor((end - start) * 1000)88        MigrationManager._save_migration_status(db, migration, execution_time, exc is None)...

migrator.py

Source:migrator.py

...20            current_schema_version = self.dataschema.current_schema()21            migrations = self.collector.migrations()22            migrations_to_execute = self._select_migrations(migrations, current_schema_version)23            for migration in migrations_to_execute:24                self._execute_migration(migration)25        except MigrationExecutionError as exc:26            self.logger.critical("Error executing migration {}".format(exc.dest))27            raise28    def _select_migrations(self, migrations, current_version):29        return migrations[self._index_for_version(migrations, current_version): None]30    def _index_for_version(self, migrations, version):31        if version is None:32            return 033        for index, migration in enumerate(migrations):34            if version in migration:35                return index + 136        return len(migrations)37    def _execute_migration(self, migration):38        return_value = self.subprocess_module.call(['python', migration])39        target = self.extract_version_from_file(migration)40        if return_value == 0:41            self.dataschema.set_schema_version(target)42            self.logger.info("Migration {} successfully executed".format(target))43        else:44            raise MigrationExecutionError(target)45class TestMigrator(object):46    def __init__(self, database, migrator, logger=logging.getLogger()):47        self.database = database48        self.migrator = migrator49        self.logger = logger50    def test_migrate(self):51        self._execute_database_manipulation_command(self.database.remove_test_database)...

__init__.py

Source:__init__.py

...20        if self.version > Credentials.DESIRED_VERSION:21            LOGGER.fatal('credentials.db file is using an unsupported version of the schema.')22            exit(1)23        for num in range(self.version + 1, Credentials.DESIRED_VERSION + 1):24            self._execute_migration(os.path.join(Credentials.MIGRATIONS_DIR, f'v{num}.sql'))25            self._db.execute(f'PRAGMA user_version={num}')26            self._db.commit()27    @property28    def version(self):29        return self._db.execute('PRAGMA user_version').fetchone()[0]30    def _execute_migration(self, path):31        with open(path) as file:32            script = file.read()33        self._db.executescript(script)34    def save_token(self, alias, id_token):35        self._validate_token(alias, id_token)36        self._db.execute('INSERT OR REPLACE INTO tokens(provider, id_token) VALUES (?, ?)', (alias, id_token))37        self._db.commit()38    def get_token(self, alias):39        result = self._db.execute(40                'SELECT id_token FROM tokens WHERE provider = ?',41                (alias,)42            ).fetchone()43        if result:44            try:...

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.