diff options
author | Daniel Baumann <daniel.baumann@progress-linux.org> | 2021-02-08 11:28:14 +0000 |
---|---|---|
committer | Daniel Baumann <daniel.baumann@progress-linux.org> | 2021-02-08 11:28:14 +0000 |
commit | b678a621c57a6d3fdfac14bdbbef0ed743ab1742 (patch) | |
tree | 5481c14ce75dfda9c55721de033992b45ab0e1dc /test/utils.py | |
parent | Initial commit. (diff) | |
download | mycli-b678a621c57a6d3fdfac14bdbbef0ed743ab1742.tar.xz mycli-b678a621c57a6d3fdfac14bdbbef0ed743ab1742.zip |
Adding upstream version 1.22.2.upstream/1.22.2
Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>
Diffstat (limited to 'test/utils.py')
-rw-r--r-- | test/utils.py | 94 |
1 files changed, 94 insertions, 0 deletions
diff --git a/test/utils.py b/test/utils.py new file mode 100644 index 0000000..66b4194 --- /dev/null +++ b/test/utils.py @@ -0,0 +1,94 @@ +import os +import time +import signal +import platform +import multiprocessing + +import pymysql +import pytest + +from mycli.main import special + +PASSWORD = os.getenv('PYTEST_PASSWORD') +USER = os.getenv('PYTEST_USER', 'root') +HOST = os.getenv('PYTEST_HOST', 'localhost') +PORT = int(os.getenv('PYTEST_PORT', 3306)) +CHARSET = os.getenv('PYTEST_CHARSET', 'utf8') +SSH_USER = os.getenv('PYTEST_SSH_USER', None) +SSH_HOST = os.getenv('PYTEST_SSH_HOST', None) +SSH_PORT = os.getenv('PYTEST_SSH_PORT', 22) + + +def db_connection(dbname=None): + conn = pymysql.connect(user=USER, host=HOST, port=PORT, database=dbname, + password=PASSWORD, charset=CHARSET, + local_infile=False) + conn.autocommit = True + return conn + + +try: + db_connection() + CAN_CONNECT_TO_DB = True +except: + CAN_CONNECT_TO_DB = False + +dbtest = pytest.mark.skipif( + not CAN_CONNECT_TO_DB, + reason="Need a mysql instance at localhost accessible by user 'root'") + + +def create_db(dbname): + with db_connection().cursor() as cur: + try: + cur.execute('''DROP DATABASE IF EXISTS _test_db''') + cur.execute('''CREATE DATABASE _test_db''') + except: + pass + + +def run(executor, sql, rows_as_list=True): + """Return string output for the sql to be run.""" + result = [] + + for title, rows, headers, status in executor.run(sql): + rows = list(rows) if (rows_as_list and rows) else rows + result.append({'title': title, 'rows': rows, 'headers': headers, + 'status': status}) + + return result + + +def set_expanded_output(is_expanded): + """Pass-through for the tests.""" + return special.set_expanded_output(is_expanded) + + +def is_expanded_output(): + """Pass-through for the tests.""" + return special.is_expanded_output() + + +def send_ctrl_c_to_pid(pid, wait_seconds): + """Sends a Ctrl-C like signal to the given `pid` after `wait_seconds` + seconds.""" + time.sleep(wait_seconds) + system_name = platform.system() + if system_name == "Windows": + os.kill(pid, signal.CTRL_C_EVENT) + else: + os.kill(pid, signal.SIGINT) + + +def send_ctrl_c(wait_seconds): + """Create a process that sends a Ctrl-C like signal to the current process + after `wait_seconds` seconds. + + Returns the `multiprocessing.Process` created. + + """ + ctrl_c_process = multiprocessing.Process( + target=send_ctrl_c_to_pid, args=(os.getpid(), wait_seconds) + ) + ctrl_c_process.start() + return ctrl_c_process |