def init_mysql_server()

in mysqlx-connector-python/unittests.py [0:0]

• 124 lines of code
• 16 McCabe index (conditional complexity)

def init_mysql_server(port, options):
    """Initialize a MySQL Server"""
    name = "server{0}".format(len(tests.MYSQL_SERVERS) + 1)
    extra_args = [
        {
            "version": (5, 7, 17),
            "options": {
                "mysqlx_bind_address": "mysqlx_bind_address={0}".format(
                    "::" if tests.IPV6_AVAILABLE else "0.0.0.0"
                )
            },
        },
        {
            "version": (8, 0, 30),
            "options": {
                "innodb_log_file_size": "1Gb",
            },
        },
    ]

    if options.secure_file_priv is not None:
        extra_args += [
            {
                "version": (5, 5, 53),
                "options": {
                    "secure_file_priv": "secure_file_priv = %s"
                    % options.secure_file_priv
                },
            }
        ]
    else:
        extra_args += [{"version": (5, 5, 53), "options": {"secure_file_priv": ""}}]

    try:
        mysql_server = mysqld.MySQLServer(
            basedir=options.mysql_basedir,
            topdir=os.path.join(options.mysql_topdir, "cpy_" + name),
            cnf=MY_CNF,
            bind_address=options.bind_address,
            port=port,
            mysqlx_port=options.mysqlx_port,
            unix_socket_folder=options.unix_socket_folder,
            ssl_folder=os.path.abspath(tests.SSL_DIR),
            ssl_ca="tests_CA_cert.pem",
            ssl_cert="tests_server_cert.pem",
            ssl_key="tests_server_key.pem",
            name=name,
            extra_args=extra_args,
            sharedir=options.mysql_sharedir,
        )
    except tests.mysqld.MySQLBootstrapError as err:
        LOGGER.error(
            "Failed initializing MySQL server "
            "'{name}': {error}".format(name=name, error=str(err))
        )
        sys.exit(1)

    if os.name == "posix" and len(mysql_server.unix_socket) > 110:
        LOGGER.error(
            "Unix socket file is to long for mysqld (>110). "
            "Consider using --unix-socket"
        )
        sys.exit(1)

    mysql_server._debug = options.debug

    have_to_bootstrap = True
    if options.force:
        # Force removal of previous test data
        if mysql_server.check_running():
            mysql_server.stop()
            if not mysql_server.wait_down():
                LOGGER.error(
                    "Failed shutting down the MySQL server '{name}'".format(name=name)
                )
                sys.exit(1)
        mysql_server.remove()
    else:
        if mysql_server.check_running():
            LOGGER.info(
                "Reusing previously bootstrapped MySQL server '{name}'".format(
                    name=name
                )
            )
            have_to_bootstrap = False
        else:
            LOGGER.warning(
                "Can not connect to previously bootstrapped "
                "MySQL Server '{name}'; forcing bootstrapping".format(name=name)
            )
            mysql_server.remove()

    tests.MYSQL_VERSION = mysql_server.version
    tests.MYSQL_LICENSE = mysql_server.license
    tests.MYSQL_VERSION_TXT = ".".join([str(i) for i in mysql_server.version])
    tests.MYSQL_SERVERS.append(mysql_server)

    mysql_server.client_config = {
        "host": options.host,
        "port": port,
        "unix_socket": mysql_server.unix_socket,
        "user": "root",
        "password": "",
        "database": "myconnpy",
        "connection_timeout": 60,
    }

    mysql_server.xplugin_config = {
        "host": options.host,
        "port": options.mysqlx_port,
        "user": "root",
        "password": "",
        "schema": "myconnpy",
    }

    if mysql_server.version >= (5, 7, 15):
        mysql_server.xplugin_config["socket"] = mysql_server.mysqlx_unix_socket
        os.environ["MYSQLX_UNIX_PORT"] = mysql_server.mysqlx_unix_socket

    # Bootstrap and start a MySQL server
    if have_to_bootstrap:
        LOGGER.info("Bootstrapping MySQL server '{name}'".format(name=name))
        try:
            mysql_server.bootstrap()
        except tests.mysqld.MySQLBootstrapError as exc:
            LOGGER.error(
                "Failed bootstrapping MySQL server '{name}': "
                "{error}".format(name=name, error=str(exc))
            )
            sys.exit(1)
        mysql_server.start()
        if not mysql_server.wait_up():
            LOGGER.error(
                "Failed to start the MySQL server '{name}'. "
                "Check error log.".format(name=name)
            )
            sys.exit(1)