How to use basedir method in avocado

Best Python code snippet using avocado_python

take_backup.py

Source:take_backup.py

1from master_backup_script.backuper import Backup2from prepare_env_test_mode.run_benchmark import RunBenchmark3from time import sleep4import os, signal5import shutil6import logging7import concurrent.futures8from shlex import split9from subprocess import Popen, getstatusoutput10from general_conf import path_config11logger = logging.getLogger(__name__)12class WrapperForBackupTest(Backup):13    # The Backup class child to do some staff for backup in --test_mode14    def __init__(self, config=path_config.config_path_file, full_dir=None, inc_dir=None, basedir=None):15        self.conf = config16        super().__init__(config=self.conf)17        if full_dir is not None:18            self.full_dir = full_dir19        if inc_dir is not None:20            self.inc_dir = inc_dir21        if basedir is not None:22            self.basedir = basedir23    @staticmethod24    def general_tablespace_rel(basedir):25        directory = "{}/relative_path".format(basedir)26        if os.path.exists(directory):27            try:28                logger.debug("Removing relative_path...")29                shutil.rmtree(directory)30            except Exception as err:31                logger.debug("FAILED: Removing relative_path")32                logger.error(err)33        try:34            logger.debug("Creating relative_path...")35            os.makedirs(directory)36        except Exception as err:37            logger.debug("FAILED: Creating relative_path")38            logger.error(err)39    @staticmethod40    def parallel_sleep_queries(basedir, sql, sock):41        dir_path = os.path.dirname(os.path.realpath(__file__))42        bash_command = "{}/run_sql_queries.sh {} {} '{}'".format(dir_path, basedir, sock, sql)43        try:44            process = Popen(45                split(bash_command),46                stdin=None,47                stdout=None,48                stderr=None)49        except Exception as e:50            print(e)51    @staticmethod52    def run_ddl_test_sh(basedir, sock):53        logger.debug("Trying to run call_ddl_test.sh")54        dir_path = os.path.dirname(os.path.realpath(__file__))55        bash_command = "{}/call_ddl_test.sh {} {} {}".format(dir_path, dir_path, basedir, sock)56        try:57            process = Popen(58                split(bash_command),59                stdin=None,60                stdout=None,61                stderr=None)62        except Exception as e:63            print(e)64    @staticmethod65    def run_temp_table_test_sh(basedir, sock):66        # Static method for calling call_temp_table.sh bash file.67        logger.debug("Trying to run call_temp_table_test.sh")68        dir_path = os.path.dirname(os.path.realpath(__file__))69        bash_command = "{}/call_temp_table_test.sh {} {} {}".format(dir_path, dir_path, basedir, sock)70        try:71            process = Popen(72                split(bash_command),73                stdin=None,74                stdout=None,75                stderr=None)76        except Exception as e:77            print(e)78    @staticmethod79    def run_call_create_index_temp_sh(basedir, sock):80        # Static method for calling call_create_index_temp.sh bash file.81        logger.debug("Trying to run call_temp_table_test.sh")82        dir_path = os.path.dirname(os.path.realpath(__file__))83        bash_command = "{}/call_create_index_temp.sh {} {} {}".format(dir_path, dir_path, basedir, sock)84        try:85            process = Popen(86                split(bash_command),87                stdin=None,88                stdout=None,89                stderr=None)90        except Exception as e:91            print(e)92    @staticmethod93    def run_call_innodb_online_alter_encryption_sql_sh(basedir, sock):94        logger.debug("Trying to run call_innodb_online_alter_encryption_sql.sh")95        dir_path = os.path.dirname(os.path.realpath(__file__))96        bash_command = "{}/call_innodb_online_alter_encryption_sql.sh {} {} {}".format(dir_path,97                                                                                   basedir,98                                                                                   sock,99                                                                                   dir_path)100        status, output = getstatusoutput(bash_command)101        if status == 0:102            logger.debug("Running call_innodb_online_alter_encryption_sql.sh - OK")103        else:104            logger.error("Failed to run")105            logger.error(output)106    @staticmethod107    def run_call_innodb_online_alter_encryption_alters_sh(basedir, sock):108        logger.debug("Trying to run call_innodb_online_alter_encryption_alters.sh")109        dir_path = os.path.dirname(os.path.realpath(__file__))110        bash_command = "{}/call_innodb_online_alter_encryption_alters.sh {} {} {}".format(dir_path,111                                                                                       basedir,112                                                                                       sock,113                                                                                       dir_path)114        try:115            process = Popen(116                split(bash_command),117                stdin=None,118                stdout=None,119                stderr=None)120        except Exception as e:121            print(e)122    @staticmethod123    def check_kill_process(pstring):124        # Static method for killing given processes125        for line in os.popen("ps ax | grep " + pstring + " | grep -v grep"):126            fields = line.split()127            pid = fields[0]128            if pid:129                os.kill(int(pid), signal.SIGKILL)130    @staticmethod131    def create_million_tables(basedir):132        for i in range(1000000):133            sql_create = "create table sysbench_test_db.ddl_table{}(id int not null)"134            RunBenchmark.run_sql_statement(basedir=basedir, sql_statement=sql_create.format(i))135        #sql_create_run = '{} -e \"{}\"'.format(RunBenchmark.get_mysql_conn(basedir), sql_create.format(i))136        # try:137        #     process = Popen(138        #         split(sql_create_run),139        #         stdin=None,140        #         stdout=None,141        #         stderr=None)142        # except Exception as e:143        #     print(e)144        #insert_into = "insert into sysbench_test_db.ddl_table{}(id) values(1),(2),(3),(4),(5)"145        #RunBenchmark.run_sql_statement(basedir=basedir, sql_statement=insert_into.format(i))146    def run_all_backup(self):147        # Method for taking backups using master_backup_script.backuper.py::all_backup()148        RunBenchmark().run_sysbench_prepare(basedir=self.basedir)149        # Fix for https://github.com/ShahriyarR/MySQL-AutoXtraBackup/issues/245150        # Disabled for now151        # self.create_million_tables(basedir=self.basedir)152        # with concurrent.futures.ProcessPoolExecutor(max_workers=100) as pool:153        #         for i in range(1000000):154        #             pool.submit(self.create_million_tables(basedir=self.basedir, i=i))155        if '5.7' in self.basedir:156            # Fix for https://github.com/ShahriyarR/MySQL-AutoXtraBackup/issues/205157            # Adding compression column with predefined dictionary.158            sql_create_dictionary = "CREATE COMPRESSION_DICTIONARY numbers('08566691963-88624912351-16662227201-46648573979-64646226163-77505759394-75470094713-41097360717-15161106334-50535565977')"159            RunBenchmark.run_sql_statement(basedir=self.basedir, sql_statement=sql_create_dictionary)160            # Fix for https://github.com/ShahriyarR/MySQL-AutoXtraBackup/issues/229161            # Creating encrypted general tablespace162            sql_create_tablespace = "create tablespace ts3_enc add datafile 'ts3_enc.ibd' encryption='Y'"163            RunBenchmark.run_sql_statement(basedir=self.basedir, sql_statement=sql_create_tablespace)164            # Fix for https://github.com/ShahriyarR/MySQL-AutoXtraBackup/issues/271165            # Preparing env here166            self.run_call_innodb_online_alter_encryption_sql_sh(basedir=self.basedir, sock="{}/socket.sock".format(self.basedir))167            self.run_call_innodb_online_alter_encryption_alters_sh(basedir=self.basedir, sock="{}/socket.sock".format(self.basedir))168            # Fix for https://github.com/ShahriyarR/MySQL-AutoXtraBackup/issues/268169            # Running create statement170            sql_create_table = "CREATE TABLE sysbench_test_db.t10 (a INT AUTO_INCREMENT PRIMARY KEY, b INT)"171            RunBenchmark.run_sql_statement(basedir=self.basedir, sql_statement=sql_create_table)172            for _ in range(10):173                insert_rand = "INSERT INTO sysbench_test_db.t10 (b) VALUES (FLOOR(RAND() * 10000)), (FLOOR(RAND() * 10000)), (FLOOR(RAND() * 10000))"174                RunBenchmark.run_sql_statement(basedir=self.basedir, sql_statement=insert_rand)175            for _ in range(5):176                insert_select = "INSERT INTO sysbench_test_db.t10 (b) SELECT b FROM sysbench_test_db.t10"177                RunBenchmark.run_sql_statement(basedir=self.basedir, sql_statement=insert_select)178            # Fix for https://github.com/ShahriyarR/MySQL-AutoXtraBackup/issues/268179            self.run_temp_table_test_sh(basedir=self.basedir, sock="{}/socket.sock".format(self.basedir))180            self.run_call_create_index_temp_sh(basedir=self.basedir, sock="{}/socket.sock".format(self.basedir))181            for i in range(1, 5):182                sql_encrypt = "alter table sysbench_test_db.sbtest{} encryption='Y'".format(i)183                RunBenchmark.run_sql_statement(basedir=self.basedir, sql_statement=sql_encrypt)184                # Compression related issue -> https://bugs.launchpad.net/percona-xtrabackup/+bug/1641745185                # Disabling for now186                # TODO: Enable this after #1641745 is fixed. Or disable 64K page size for MySQL;disabled.187                sql_compress = "alter table sysbench_test_db.sbtest{} compression='lz4'".format(i)188                RunBenchmark.run_sql_statement(basedir=self.basedir, sql_statement=sql_compress)189                sql_optimize = "optimize table sysbench_test_db.sbtest{}".format(i)190                RunBenchmark.run_sql_statement(basedir=self.basedir, sql_statement=sql_optimize)191                # Fix for https://github.com/ShahriyarR/MySQL-AutoXtraBackup/issues/196192                # Adding JSON + virtual + stored columns here193                sql_virtual_column = "alter table sysbench_test_db.sbtest{} add column json_test_v json generated always as (json_array(k,c,pad)) virtual".format(i)194                RunBenchmark.run_sql_statement(basedir=self.basedir, sql_statement=sql_virtual_column)195                sql_stored_column = "alter table sysbench_test_db.sbtest{} add column json_test_s json generated always as (json_array(k,c,pad)) stored".format(i)196                RunBenchmark.run_sql_statement(basedir=self.basedir, sql_statement=sql_stored_column)197                sql_create_json_column = "alter table sysbench_test_db.sbtest{} add column json_test_index varchar(255) generated always as (json_array(k,c,pad)) stored".format(i)198                RunBenchmark.run_sql_statement(basedir=self.basedir, sql_statement=sql_create_json_column)199                sql_alter_add_index = "alter table sysbench_test_db.sbtest{} add index(json_test_index)".format(i)200                RunBenchmark.run_sql_statement(basedir=self.basedir, sql_statement=sql_alter_add_index)201            general_tablespace = "create tablespace ts1 add datafile 'ts1.ibd' engine=innodb"202            RunBenchmark.run_sql_statement(basedir=self.basedir, sql_statement=general_tablespace)203            outside_tablespace_full_path = '{}/out_ts1.ibd'.format(self.basedir)204            if os.path.isfile(outside_tablespace_full_path):205                os.remove(outside_tablespace_full_path)206            # Fix for https://github.com/ShahriyarR/MySQL-AutoXtraBackup/issues/219207            general_out_tablespace = "create tablespace out_ts1 add datafile '{}' engine=innodb".format(outside_tablespace_full_path)208            RunBenchmark.run_sql_statement(basedir=self.basedir, sql_statement=general_out_tablespace)209            # Create general tablespace with relative path210            # TODO: enable this after fix for https://bugs.launchpad.net/percona-xtrabackup/+bug/1736380211            # self.general_tablespace_rel(self.basedir)212            # general_out_relative = "create tablespace out_rel_ts1 add datafile '../relative_path/out_rel_ts1.ibd' engine=innodb"213            # RunBenchmark.run_sql_statement(basedir=self.basedir, sql_statement=general_out_relative)214            for i in range(5, 10):215                sql_compress = "alter table sysbench_test_db.sbtest{} compression='zlib'".format(i)216                RunBenchmark.run_sql_statement(basedir=self.basedir, sql_statement=sql_compress)217                sql_optimize = "optimize table sysbench_test_db.sbtest{}".format(i)218                RunBenchmark.run_sql_statement(basedir=self.basedir, sql_statement=sql_optimize)219                sql_alter_compression_dict = "alter table sysbench_test_db.sbtest{} modify c varchar(250) column_format compressed with compression_dictionary numbers".format(i)220                RunBenchmark.run_sql_statement(basedir=self.basedir, sql_statement=sql_alter_compression_dict)221            for i in range(10, 15):222                # Fix for https://github.com/ShahriyarR/MySQL-AutoXtraBackup/issues/206223                # Altering some tables to use general tablespace.224                sql_virtual_column = "alter table sysbench_test_db.sbtest{} add column json_test_v json generated always as (json_array(k,c,pad)) virtual".format(i)225                RunBenchmark.run_sql_statement(basedir=self.basedir, sql_statement=sql_virtual_column)226                sql_stored_column = "alter table sysbench_test_db.sbtest{} add column json_test_s json generated always as (json_array(k,c,pad)) stored".format(i)227                RunBenchmark.run_sql_statement(basedir=self.basedir, sql_statement=sql_stored_column)228                sql_create_json_column = "alter table sysbench_test_db.sbtest{} add column json_test_index varchar(255) generated always as (json_array(k,c,pad)) stored".format(i)229                RunBenchmark.run_sql_statement(basedir=self.basedir, sql_statement=sql_create_json_column)230                sql_alter_add_index = "alter table sysbench_test_db.sbtest{} add index(json_test_index)".format(i)231                RunBenchmark.run_sql_statement(basedir=self.basedir, sql_statement=sql_alter_add_index)232                # Decrypting tables for -> https://github.com/ShahriyarR/MySQL-AutoXtraBackup/issues/264233                sql_encrypt = "alter table sysbench_test_db.sbtest{} encryption='N'".format(i)234                RunBenchmark.run_sql_statement(basedir=self.basedir, sql_statement=sql_encrypt)235                sql_alter_tablespace = "alter table sysbench_test_db.sbtest{} tablespace=ts1".format(i)236                RunBenchmark.run_sql_statement(basedir=self.basedir, sql_statement=sql_alter_tablespace)237            for i in range(15, 20):238                sql_virtual_column = "alter table sysbench_test_db.sbtest{} add column json_test_v json generated always as (json_array(k,c,pad)) virtual".format(i)239                RunBenchmark.run_sql_statement(basedir=self.basedir, sql_statement=sql_virtual_column)240                sql_stored_column = "alter table sysbench_test_db.sbtest{} add column json_test_s json generated always as (json_array(k,c,pad)) stored".format(i)241                RunBenchmark.run_sql_statement(basedir=self.basedir, sql_statement=sql_stored_column)242                sql_create_json_column = "alter table sysbench_test_db.sbtest{} add column json_test_index varchar(255) generated always as (json_array(k,c,pad)) stored".format(i)243                RunBenchmark.run_sql_statement(basedir=self.basedir, sql_statement=sql_create_json_column)244                sql_alter_add_index = "alter table sysbench_test_db.sbtest{} add index(json_test_index)".format(i)245                RunBenchmark.run_sql_statement(basedir=self.basedir, sql_statement=sql_alter_add_index)246                # Decrypting tables for -> https://github.com/ShahriyarR/MySQL-AutoXtraBackup/issues/264247                sql_encrypt = "alter table sysbench_test_db.sbtest{} encryption='N'".format(i)248                RunBenchmark.run_sql_statement(basedir=self.basedir, sql_statement=sql_encrypt)249                sql_alter_tablespace = "alter table sysbench_test_db.sbtest{} tablespace=out_ts1".format(i)250                RunBenchmark.run_sql_statement(basedir=self.basedir, sql_statement=sql_alter_tablespace)251            for i in range(25, 30):252                # Altering encrypted tables to use encrypted general tablespace253                sql_encrypt = "alter table sysbench_test_db.sbtest{} encryption='Y'".format(i)254                RunBenchmark.run_sql_statement(basedir=self.basedir, sql_statement=sql_encrypt)255                sql_virtual_column = "alter table sysbench_test_db.sbtest{} add column json_test_v json generated always as (json_array(k,c,pad)) virtual".format(256                    i)257                RunBenchmark.run_sql_statement(basedir=self.basedir, sql_statement=sql_virtual_column)258                sql_stored_column = "alter table sysbench_test_db.sbtest{} add column json_test_s json generated always as (json_array(k,c,pad)) stored".format(259                    i)260                RunBenchmark.run_sql_statement(basedir=self.basedir, sql_statement=sql_stored_column)261                sql_create_json_column = "alter table sysbench_test_db.sbtest{} add column json_test_index varchar(255) generated always as (json_array(k,c,pad)) stored".format(262                    i)263                RunBenchmark.run_sql_statement(basedir=self.basedir, sql_statement=sql_create_json_column)264                sql_alter_add_index = "alter table sysbench_test_db.sbtest{} add index(json_test_index)".format(i)265                RunBenchmark.run_sql_statement(basedir=self.basedir, sql_statement=sql_alter_add_index)266                sql_alter_tablespace = "alter table sysbench_test_db.sbtest{} tablespace=ts3_enc".format(i)267                RunBenchmark.run_sql_statement(basedir=self.basedir, sql_statement=sql_alter_tablespace)268            # TODO: enable this after fix for https://bugs.launchpad.net/percona-xtrabackup/+bug/1736380269            # for i in range(20, 25):270            #     sql_virtual_column = "alter table sysbench_test_db.sbtest{} add column json_test_v json generated always as (json_array(k,c,pad)) virtual".format(271            #         i)272            #     RunBenchmark.run_sql_statement(basedir=self.basedir, sql_statement=sql_virtual_column)273            #     sql_stored_column = "alter table sysbench_test_db.sbtest{} add column json_test_s json generated always as (json_array(k,c,pad)) stored".format(274            #         i)275            #     RunBenchmark.run_sql_statement(basedir=self.basedir, sql_statement=sql_stored_column)276            #     sql_create_json_column = "alter table sysbench_test_db.sbtest{} add column json_test_index varchar(255) generated always as (json_array(k,c,pad)) stored".format(277            #         i)278            #     RunBenchmark.run_sql_statement(basedir=self.basedir, sql_statement=sql_create_json_column)279            #     sql_alter_add_index = "alter table sysbench_test_db.sbtest{} add index(json_test_index)".format(i)280            #     RunBenchmark.run_sql_statement(basedir=self.basedir, sql_statement=sql_alter_add_index)281            #     sql_alter_tablespace = "alter table sysbench_test_db.sbtest{} tablespace=out_rel_ts1".format(i)282            #     RunBenchmark.run_sql_statement(basedir=self.basedir, sql_statement=sql_alter_tablespace)283            # NOTE: PXB will ignore rocksdb tables, which is going to break pt-table-checksum284            # for i in range(10, 15):285            #     sql_alter = "alter table sysbench_test_db.sbtest{} engine=rocksdb".format(i)286            #     RunBenchmark.run_sql_statement(basedir=self.basedir, sql_statement=sql_alter)287        # NOTE: PXB will ignore tokudb tables, which is going to break pt-table-checksum288        # for i in range(15, 20):289        #     sql_alter = "alter table sysbench_test_db.sbtest{} engine=tokudb".format(i)290        #     RunBenchmark.run_sql_statement(basedir=self.basedir, sql_statement=sql_alter)291        if '5.5' in self.basedir:292            for i in range(1, 5):293                sql_alter = "alter table sysbench_test_db.sbtest{} modify c varchar(120) CHARACTER SET utf8 COLLATE utf8_general50_ci".format(i)294                RunBenchmark.run_sql_statement(basedir=self.basedir, sql_statement=sql_alter)295        # Altering some of the table engines from innodb to myisam296        # Disabled based on -> https://bugs.mysql.com/bug.php?id=89977297        # for i in range(20, 25):298        #     sql_alter_engine = "alter table sysbench_test_db.sbtest{} engine=myisam".format(i)299        #     RunBenchmark.run_sql_statement(basedir=self.basedir, sql_statement=sql_alter_engine)300        # Fix for https://github.com/ShahriyarR/MySQL-AutoXtraBackup/issues/222301        # Creating table with data directory option302        if '5.6' in self.basedir or '5.7' in self.basedir:303            if os.path.exists('{}/{}'.format(self.basedir, 'sysbench_test_db')):304                shutil.rmtree('{}/{}'.format(self.basedir, 'sysbench_test_db'))305            sql_create_table = "create table sysbench_test_db.t1(c varchar(255)) data directory='{}'".format(self.basedir)306            RunBenchmark.run_sql_statement(basedir=self.basedir, sql_statement=sql_create_table)307            sql_insert_data = "insert into sysbench_test_db.t1 select c from sysbench_test_db.sbtest1"308            RunBenchmark.run_sql_statement(basedir=self.basedir, sql_statement=sql_insert_data)309        flush_tables = "flush tables"310        RunBenchmark.run_sql_statement(basedir=self.basedir, sql_statement=flush_tables)311        sleep(10)312        try:313            for _ in range(int(self.incremental_count) + 1):314                # RunBenchmark().run_sysbench_run(basedir=self.basedir)315                # TODO: enable when you pass --lock-ddl-per-table or --lock-ddl; disabled by default316                # Fix for https://github.com/ShahriyarR/MySQL-AutoXtraBackup/issues/243317                # Calling here ddl_test.sh file for running some DDLs.318                # self.run_ddl_test_sh(basedir=self.basedir, sock="{}/socket.sock".format(self.basedir))319                # Disabled based on -> https://bugs.mysql.com/bug.php?id=89977320                # Concurrently running select on myisam based tables.321                # with concurrent.futures.ProcessPoolExecutor(max_workers=50) as pool:322                #     for _ in range(10):323                #         for i in range(20, 25):324                #             pool.submit(325                #                 self.parallel_sleep_queries(basedir=self.basedir,326                #                                             sock="{}/socket.sock".format(self.basedir),327                #                                             sql="select benchmark(9999999, md5(c)) from sysbench_test_db.sbtest{}".format(328                #                                                 i)))329                self.all_backup()330                    # self.check_kill_process('call_ddl_test')331        except Exception as err:332            print(err)333            raise334        else:335            if os.path.isfile('{}/out_ts1.ibd'.format(self.basedir)):336                os.remove('{}/out_ts1.ibd'.format(self.basedir))337            if os.path.isfile('{}/sysbench_test_db/t1.ibd'.format(self.basedir)):338                os.remove('{}/sysbench_test_db/t1.ibd'.format(self.basedir))339            # TODO: enable this after fix for https://bugs.launchpad.net/percona-xtrabackup/+bug/1736380340            # self.general_tablespace_rel(self.basedir)341        finally:342            # self.check_kill_process('call_ddl_test')343            self.check_kill_process('call_temp_table_test')344            self.check_kill_process('call_create_index_temp')345            self.check_kill_process('call_innodb_alter_encryption_alters')346            self.check_kill_process('call_innodb_alter_encryption_sql')347            pass...

test_runner_test_mode.py

Source:test_runner_test_mode.py

1import pytest2from prepare_env_test_mode.run_benchmark import RunBenchmark3class TestRunnerTestMode:4    # No matter here which fixture to choose I need an object5    @pytest.mark.usefixtures("return_runner_test_mode_obj_5_6_xb_2_3")6    def test_get_free_tcp_port(self, return_runner_test_mode_obj_5_6_xb_2_3):7        return_runner_test_mode_obj_5_6_xb_2_3.get_free_tcp_port()8    @pytest.mark.usefixtures("return_runner_test_mode_obj_5_6_xb_2_3")9    def test_create_dsns_table(self, return_runner_test_mode_obj_5_6_xb_2_3):10        for basedir in return_runner_test_mode_obj_5_6_xb_2_3.basedirs:11            if '5.6' in basedir:12                mysql_master_client_cmd = RunBenchmark(config=return_runner_test_mode_obj_5_6_xb_2_3.conf).get_mysql_conn(13                    basedir=basedir)14                assert return_runner_test_mode_obj_5_6_xb_2_3.create_dsns_table(mysql_master_client_cmd)15    @pytest.mark.usefixtures("return_runner_test_mode_obj_5_6_xb_2_3")16    def test_populate_dsns_table(self, return_runner_test_mode_obj_5_6_xb_2_3):17        for basedir in return_runner_test_mode_obj_5_6_xb_2_3.basedirs:18            if '5.6' in basedir:19                mysql_master_client_cmd = RunBenchmark(config=return_runner_test_mode_obj_5_6_xb_2_3.conf).get_mysql_conn(20                    basedir=basedir)21                file_name = "cl_node0"22                mysql_slave_client_cmd = RunBenchmark(config=return_runner_test_mode_obj_5_6_xb_2_3.conf).get_mysql_conn(basedir=basedir,23                                                                                       file_name=file_name)24                # Get slave port here25                sql = "{} -e 'select @@port'".format(mysql_slave_client_cmd)26                port = return_runner_test_mode_obj_5_6_xb_2_3.run_sql_command(sql_command=sql)27                #slave_sock = "{}/sock0.sock".format(basedir)28                assert return_runner_test_mode_obj_5_6_xb_2_3.populate_dsns_table(sql_conn=mysql_master_client_cmd, slave_port=port[7:])29    @pytest.mark.usefixtures("return_runner_test_mode_obj_5_6_xb_2_3")30    def test_run_pt_table_checksum(self, return_runner_test_mode_obj_5_6_xb_2_3):31        for basedir in return_runner_test_mode_obj_5_6_xb_2_3.basedirs:32            if '5.6' in basedir:33                # socket = return_run_benchmark_obj.get_sock(basedir=basedir)34                # conn_options = "--user={} --socket={}".format('root', socket)35                assert return_runner_test_mode_obj_5_6_xb_2_3.run_pt_table_checksum(basedir=basedir)36    @pytest.mark.usefixtures("return_runner_test_mode_obj_5_6_xb_2_3")37    def test_run_change_master(self, return_runner_test_mode_obj_5_6_xb_2_3):38        for basedir in return_runner_test_mode_obj_5_6_xb_2_3.basedirs:39            if '5.6' in basedir:40                assert return_runner_test_mode_obj_5_6_xb_2_3.run_change_master(basedir=basedir, file_name='cl_node0')41    @pytest.mark.usefixtures("return_runner_test_mode_obj_5_6_xb_2_3")42    def test_drop_blank_mysql_users(self, return_runner_test_mode_obj_5_6_xb_2_3):43        for basedir in return_runner_test_mode_obj_5_6_xb_2_3.basedirs:44            if '5.6' in basedir:45                mysql_master_client_cmd = RunBenchmark(config=return_runner_test_mode_obj_5_6_xb_2_3.conf).get_mysql_conn(basedir=basedir)46                assert return_runner_test_mode_obj_5_6_xb_2_3.drop_blank_mysql_users(mysql_master_client_cmd)47    @pytest.mark.usefixtures("return_runner_test_mode_obj_5_6_xb_2_3")48    def test_check_slave_status(self, return_runner_test_mode_obj_5_6_xb_2_3):49        for basedir in return_runner_test_mode_obj_5_6_xb_2_3.basedirs:50            if '5.6' in basedir:51                mysql_slave_client_cmd = RunBenchmark(config=return_runner_test_mode_obj_5_6_xb_2_3.conf).get_mysql_conn(basedir=basedir, file_name="cl_node0")52                show_slave_status = "{} -e 'show slave status\G'"53                assert return_runner_test_mode_obj_5_6_xb_2_3.check_slave_status(show_slave_status.format(mysql_slave_client_cmd)) is None54    @pytest.mark.usefixtures("return_runner_test_mode_obj_5_6_xb_2_3")55    def test_wipe_backup_prepare_copyback_5_6_xb_2_3(self, return_runner_test_mode_obj_5_6_xb_2_3):56        for basedir in return_runner_test_mode_obj_5_6_xb_2_3.basedirs:57            if '5.6' in basedir:58                return_runner_test_mode_obj_5_6_xb_2_3.wipe_backup_prepare_copyback(basedir=basedir)59    @pytest.mark.usefixtures("return_runner_test_mode_obj_5_6_xb_2_4")60    def test_wipe_backup_prepare_copyback_5_6_xb_2_4(self, return_runner_test_mode_obj_5_6_xb_2_4):61        for basedir in return_runner_test_mode_obj_5_6_xb_2_4.basedirs:62            if '5.6' in basedir:63                return_runner_test_mode_obj_5_6_xb_2_4.wipe_backup_prepare_copyback(basedir=basedir)64    @pytest.mark.usefixtures("return_runner_test_mode_obj_5_7_xb_2_4")65    def test_wipe_backup_prepare_copyback_5_7_xb_2_4(self, return_runner_test_mode_obj_5_7_xb_2_4):66        for basedir in return_runner_test_mode_obj_5_7_xb_2_4.basedirs:67            if '5.7' in basedir:...

test_clone_build_start_server.py

Source:test_clone_build_start_server.py

...15    def test_build_server(self, return_clone_obj):16        assert return_clone_obj.build_server() is True17    def test_rename_basedirs(self, return_clone_obj):18        assert return_clone_obj.rename_basedirs() is True19    def test_get_basedir(self, return_clone_obj):20        assert isinstance(return_clone_obj.get_basedir(), list)21        assert len(return_clone_obj.get_basedir()) > 022    @pytest.mark.usefixtures("return_basedir")23    def test_prepare_startup(self, return_clone_obj, return_basedir):24        basedir_list = return_basedir25        for basedir in basedir_list:26            assert return_clone_obj.prepare_startup(basedir_path=basedir) is True27    @pytest.mark.usefixtures("return_basedir")28    def test_prepare_start_dynamic(self, return_clone_obj, return_basedir):29        basedir_list = return_basedir30        for basedir in basedir_list:31            assert return_clone_obj.prepare_start_dynamic(basedir_path=basedir) is True32    @pytest.mark.usefixtures("return_basedir")33    def test_start_server(self, return_clone_obj, return_basedir):34        basedirs = return_basedir35        for basedir in basedirs:...

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.