How to use replicate - 10 common examples

export_rpl test.
"""

import os

import mutlib
import replicate

from mysql.utilities.exception import MUTLibError, UtilError


_RPL_MODES = ["master", "slave", "both"]
_LOCKTYPES = ['no-locks', 'lock-all', 'snapshot']


class test(replicate.test):
    """check --rpl parameter for export utility
    This test executes a series of export database operations on a single
    server using a variety of --rpl and --locking options. It uses the
    replicate test as a parent for setup and teardown methods.
    """

    server3 = None
    s3_serverid = None

    def check_prerequisites(self):
        # Check MySQL server version - Must be 5.1.0 or higher
        if not self.servers.get_server(0).check_version_compat(5, 1, 0):
            raise MUTLibError("Test requires server version 5.1.0 or higher")
        self.check_gtid_unsafe()
        return replicate.test.check_prerequisites(self)

#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
#

"""
replicate_parameters test.
"""

import replicate

from mysql.utilities.exception import MUTLibError, UtilError


class test(replicate.test):
    """check parameters for the replicate utility
    This test executes the replicate utility parameters. It uses the
    replicate test as a parent for setup and teardown methods.
    """

    def check_prerequisites(self):
        return replicate.test.check_prerequisites(self)

    def setup(self):
        return replicate.test.setup(self)

    def run(self):
        self.mask_global = False  # Turn off global masks
        self.res_fname = "result.txt"

        test_num = 1

# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
#

"""
replicate_innodb test.
"""

import os

import replicate

from mysql.utilities.exception import MUTLibError, UtilError


class test(replicate.test):
    """setup replication
    This test attempts to replicate among a master and slave whose
    innodb settings are different. It uses the replicate test for
    inherited methods.
    """

    server5 = None
    s5_serverid = None

    def check_prerequisites(self):
        if self.servers.get_server(0).check_version_compat(5, 5, 0):
            raise MUTLibError("Test requires server version 5.1.")
        return self.check_num_servers(1)

    def setup(self):
        self.server0 = self.servers.get_server(0)

# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
#

"""
replicate_errors test.
"""

import socket

import replicate
import mutlib

from mysql.utilities.exception import MUTLibError


class test(replicate.test):
    """check error conditions
    This test ensures the known error conditions are tested. It uses the
    cloneuser test as a parent for setup and teardown methods.
    """

    server3 = None
    port3 = None

    def check_prerequisites(self):
        return replicate.test.check_prerequisites(self)

    def setup(self):
        self.server3 = None
        return replicate.test.setup(self)

    def run(self):

def run_and_record_test(self, comment, options):
        """Execute the test and record the results.

        comment[in]     Comment.
        options[in]     Options for running rpl test.
        """
        res = replicate.test.run_rpl_test(self, self.server1, self.server2,
                                          self.s1_serverid, comment, options,
                                          False, 0, False)
        if not res:
            raise MUTLibError("{0}: failed".format(comment))

        self.wait_for_slave(10)

        # Record the results
        self.get_table_rows(comment)

        # Stop slave
        self.stop_slave(comment, self.server1)

        return True

def cleanup(self):
        return replicate.test.cleanup(self)

def check_prerequisites(self):
        return replicate.test.check_prerequisites(self)

def setup(self):
        self.server0 = self.servers.get_server(0)
        self.server5 = None
        self.s5_serverid = None

        replicate.test.setup(self)

        index = self.servers.find_server_by_name("rep_slave_no_innodb")
        if index >= 0:
            self.server5 = self.servers.get_server(index)
            try:
                res = self.server5.show_server_variable("server_id")
            except MUTLibError as err:
                raise MUTLibError("Cannot get replication slave "
                                  "server_id: {0}".format(err.errmsg))
            self.s5_serverid = int(res[0][1])
        else:
            self.s5_serverid = self.servers.get_next_id()
            res = self.servers.spawn_new_server(
                self.server0, self.s5_serverid, "rep_slave_no_innodb",
                ' --mysqld="--log-bin=mysql-bin  --skip-innodb '
                '--default-storage-engine=MyISAM"')

def setup(self):
        self.res_fname = "result.txt"
        result = replicate.test.setup(self)
        if not result:
            return False

        master_str = "--master={0}".format(
            self.build_connection_string(self.server1))
        slave_str = " --slave={0}".format(
            self.build_connection_string(self.server2))
        conn_str = master_str + slave_str
        self.server2.exec_query("STOP SLAVE")
        self.server2.exec_query("RESET SLAVE")
        self.server1.exec_query("STOP SLAVE")
        self.server1.exec_query("RESET SLAVE")

        data_file = os.path.normpath("./std_data/basic_data.sql")
        try:
            self.server1.exec_query("DROP DATABASE IF EXISTS util_test")

def cleanup(self):
        return replicate.test.cleanup(self)