in mysql-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 mysql.connector logger, and filter out warnings
mysql_connector_logger = logging.getLogger("mysql.connector")
tests.setup_logger(
mysql_connector_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,
)
# If Django was supplied, add Django to PYTHONPATH
if options.django_path:
sys.path.insert(0, options.django_path)
try:
import django
tests.DJANGO_VERSION = django.VERSION[0:3]
except ImportError:
msg = "Could not find django package at {0}".format(options.django_path)
LOGGER.error(msg)
sys.exit(1)
if sys.version_info[0] == 3 and tests.DJANGO_VERSION < (1, 5):
LOGGER.error("Django older than v1.5 will not work with Python 3")
sys.exit(1)
# 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.client_config = {
"host": options.host,
"port": options.port,
"user": options.user,
"password": options.password,
"database": "myconnpy",
"connection_timeout": 60,
"unix_socket": options.unix_socket_folder,
}
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 mysql.connector
with mysql.connector.connect(
host=options.host,
port=options.port,
user=options.user,
password=options.password,
) as cnx:
with cnx.cursor() as cur:
cur = cnx.cursor()
# Create database for testing
LOGGER.info("Creating 'myconnpy' database")
cur.execute("DROP DATABASE IF EXISTS myconnpy")
cur.execute("CREATE DATABASE myconnpy CHARACTER SET utf8mb4")
# Version
cur.execute("SELECT VERSION()")
res = cur.fetchone()
match = version_re.match(res[0])
if not match:
raise ValueError("Invalid version number '{}'".format(res[0]))
ver = tuple(map(int, match.groups()))
# License
cur.execute("SHOW VARIABLES LIKE 'license'")
res = cur.fetchone()
lic = res[1]
mysql_server.version = ver
mysql_server.license = lic
# Socket
cur.execute("SHOW VARIABLES LIKE 'socket'")
res = cur.fetchone()
mysql_server.unix_socket = res[1]
# MySQL X Socket
cur.execute("SHOW VARIABLES LIKE 'mysqlx_socket'")
res = cur.fetchone()
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)
tests.MYSQL_CAPI = options.mysql_capi
if not options.skip_install:
openssl_include_dir = options.openssl_include_dir or os.environ.get(
"OPENSSL_INCLUDE_DIR"
)
openssl_lib_dir = options.openssl_lib_dir or os.environ.get("OPENSSL_LIB_DIR")
tests.install_connector(
_TOPDIR,
tests.TEST_BUILD_DIR,
openssl_include_dir,
openssl_lib_dir,
options.mysql_capi,
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:
# Run test cases
if options.stats:
if options.stats_host:
stats_db_info = {
"host": options.stats_host,
"port": options.stats_port,
"user": options.stats_user,
"password": options.stats_password,
"database": options.stats_db,
}
cnxstats = mysql.connector.connect(**stats_db_info)
setup_stats_db(cnxstats)
else:
cnxstats = None
result = StatsTestRunner(verbosity=options.verbosity, dbcnx=cnxstats).run(
testsuite
)
elif 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)