in mysqlx-connector-python/unittests.py [0:0]
def main():
parser = _get_arg_parser()
options = parser.parse_args()
tests.OPTIONS_INIT = True
if isinstance(options, tuple):
# Fallback to old optparse
options = options[0]
if options.show_tests:
sys.path.insert(0, os.path.join(os.getcwd(), "lib"))
for name, _, description in tests.get_test_modules():
print("{0:22s} {1}".format(name, description))
sys.exit()
# Setup tests logger
tests.setup_logger(LOGGER, debug=options.debug, logfile=options.logfile)
# Setup mysqlx logger, and filter out warnings
mysqlx_logger = logging.getLogger("mysqlx")
tests.setup_logger(
mysqlx_logger,
debug=options.debug,
logfile=options.logfile,
filter=warnings_filter,
)
LOGGER.info(
"MySQL Connector/Python unittest using Python v{0}".format(
".".join([str(v) for v in sys.version_info[0:3]])
)
)
# Check if we can test IPv6
if options.ipv6:
if not tests.IPV6_AVAILABLE:
LOGGER.error("Can not test IPv6: not available on your system")
sys.exit(1)
options.bind_address = "::"
options.host = "::1"
LOGGER.info("Testing using IPv6. Binding to :: and using host ::1")
else:
tests.IPV6_AVAILABLE = False
if not options.mysql_sharedir:
options.mysql_sharedir = os.path.join(options.mysql_basedir, "share")
LOGGER.debug("Setting default sharedir: %s", options.mysql_sharedir)
if options.mysql_topdir != MYSQL_DEFAULT_TOPDIR:
# Make sure the topdir is absolute
if not os.path.isabs(options.mysql_topdir):
options.mysql_topdir = os.path.join(
os.path.dirname(os.path.realpath(__file__)),
options.mysql_topdir,
)
# We have to at least run 1 MySQL server
if options.use_external_server:
name = "server{}".format(len(tests.MYSQL_SERVERS) + 1)
mysql_server = mysqld.MySQLExternalServer(MY_CNF, name)
mysql_server.xplugin_config = {
"host": options.host,
"port": options.mysqlx_port,
"user": options.user,
"password": options.password,
"schema": "myconnpy",
"socket": options.unix_socket_folder,
}
tests.MYSQL_SERVERS.append(mysql_server)
tests.MYSQL_EXTERNAL_SERVER = True
version_re = re.compile(r"^(\d+)\.(\d+)\.(\d+)+", re.ASCII)
try:
import mysqlx
with mysqlx.get_session(
host=options.host,
port=options.mysqlx_port,
user=options.user,
password=options.password,
) as session:
# Version
sql_res = session.sql("SELECT VERSION()").execute()
res = sql_res.fetch_one()
match = version_re.match(res[0])
if not match:
raise ValueError("Invalid version number '{}'".format(res[0]))
ver = tuple(map(int, match.groups()))
# License
sql_res = session.sql("SHOW VARIABLES LIKE 'license'").execute()
res = sql_res.fetch_one()
lic = res[1]
mysql_server.version = ver
mysql_server.license = lic
# MySQL X Socket
sql_res = session.sql("SHOW VARIABLES LIKE 'mysqlx_socket'").execute()
res = sql_res.fetch_one()
if res:
mysql_server.mysqlx_unix_socket = res[1]
tests.MYSQL_VERSION = mysql_server.version
tests.MYSQL_LICENSE = mysql_server.license
tests.MYSQL_VERSION_TXT = ".".join(map(str, mysql_server.version))
except Exception as err:
LOGGER.error("Failed connecting to the external MySQL server: %s", err)
sys.exit(1)
else:
# Bootstrap MySQL server
init_mysql_server(port=(options.port), options=options)
if not options.skip_install:
protobuf_include_dir = options.protobuf_include_dir or os.environ.get(
"MYSQLXPB_PROTOBUF_INCLUDE_DIR"
)
protobuf_lib_dir = options.protobuf_lib_dir or os.environ.get(
"MYSQLXPB_PROTOBUF_LIB_DIR"
)
protoc = options.protoc or os.environ.get("MYSQLXPB_PROTOC")
if any((protobuf_include_dir, protobuf_lib_dir, protoc)):
if not protobuf_include_dir:
LOGGER.error("Unable to find Protobuf include directory.")
sys.exit(1)
if not protobuf_lib_dir:
LOGGER.error("Unable to find Protobuf library directory.")
sys.exit(1)
if not protoc:
LOGGER.error("Unable to find Protobuf protoc binary.")
sys.exit(1)
tests.install_connector(
_TOPDIR,
tests.TEST_BUILD_DIR,
protobuf_include_dir,
protobuf_lib_dir,
protoc,
options.extra_compile_args,
options.extra_link_args,
options.debug,
)
# Which tests cases to run
testcases = []
if options.test_regex_pattern:
pattern = re.compile(options.test_regex_pattern)
testcases = [
module
for name, module, _ in tests.get_test_modules()
if pattern.match(name)
]
if not testcases:
LOGGER.error("No test matches the provided regex pattern")
elif options.testcase:
for name, module, _ in tests.get_test_modules():
if name == options.testcase or module == options.testcase:
LOGGER.info("Executing tests in module %s", module)
testcases = [module]
break
if not testcases:
LOGGER.error("Test case not valid; see --help-tests")
sys.exit(1)
elif options.onetest:
LOGGER.info("Executing test: %s", options.onetest)
testcases = [options.onetest]
else:
testcases = [mod[1] for mod in tests.get_test_modules()]
# Load tests
test_loader = unittest.TestLoader()
testsuite = None
if testcases:
# Check if we nee to test anything with the C Extension
if any(["cext" in case for case in testcases]):
# Try to load the C Extension, and try to load the MySQL library
tests.check_c_extension()
testsuite = test_loader.loadTestsFromNames(testcases)
else:
LOGGER.error("No test cases loaded.")
sys.exit(1)
# Initialize the other MySQL Servers
if not options.use_external_server:
for i in range(1, tests.MYSQL_SERVERS_NEEDED):
init_mysql_server(port=(options.port + i), options=options)
LOGGER.info(
"Using MySQL server version %s %s",
".".join([str(v) for v in tests.MYSQL_VERSION[0:3]]),
tests.MYSQL_LICENSE,
)
LOGGER.info("Starting unit tests")
was_successful = False
try:
if sys.version_info[0:2] == (2, 6):
result = Python26TestRunner(verbosity=options.verbosity).run(testsuite)
else:
result = BasicTestRunner(verbosity=options.verbosity).run(testsuite)
was_successful = result.wasSuccessful()
except KeyboardInterrupt:
LOGGER.info("Unittesting was interrupted")
was_successful = False
# Log messages added by test cases
for msg in tests.MESSAGES["WARNINGS"]:
LOGGER.warning(msg)
for msg in tests.MESSAGES["INFO"]:
LOGGER.info(msg)
# Show skipped tests
if len(tests.MESSAGES["SKIPPED"]):
LOGGER.info("Skipped tests: %d", len(tests.MESSAGES["SKIPPED"]))
for msg in tests.MESSAGES["SKIPPED"]:
LOGGER.info("Skipped: " + msg)
# Clean up
try:
tests.MYSQL_DUMMY_THREAD.join()
tests.MYSQL_DUMMY.shutdown()
tests.MYSQL_DUMMY.server_close()
except:
# Is OK when failed
pass
if not options.use_external_server:
for mysql_server in tests.MYSQL_SERVERS:
name = mysql_server.name
if not options.keep:
mysql_server.stop()
if not mysql_server.wait_down():
LOGGER.error("Failed stopping MySQL server '%s'", name)
else:
mysql_server.remove()
LOGGER.info(
"MySQL server '%s' stopped and cleaned up",
name,
)
elif not mysql_server.check_running():
mysql_server.start()
if not mysql_server.wait_up():
LOGGER.error(
"MySQL could not be kept running; failed to restart",
)
else:
LOGGER.info(
"MySQL server kept running on %s:%d",
mysql_server.bind_address,
mysql_server.port,
)
# Make sure the DEVNULL file is closed
try:
mysqld.DEVNULL.close()
except:
pass
txt = ""
if not was_successful:
txt = "not "
LOGGER.info("MySQL Connector/Python unittests were %ssuccessful", txt)
# Return result of tests as exit code
sys.exit(not was_successful)