summaryrefslogtreecommitdiffstats
path: root/test/features/environment.py
diff options
context:
space:
mode:
Diffstat (limited to 'test/features/environment.py')
-rw-r--r--test/features/environment.py176
1 files changed, 176 insertions, 0 deletions
diff --git a/test/features/environment.py b/test/features/environment.py
new file mode 100644
index 0000000..1ea0f08
--- /dev/null
+++ b/test/features/environment.py
@@ -0,0 +1,176 @@
+import os
+import shutil
+import sys
+from tempfile import mkstemp
+
+import db_utils as dbutils
+import fixture_utils as fixutils
+import pexpect
+
+from steps.wrappers import run_cli, wait_prompt
+
+test_log_file = os.path.join(os.environ['HOME'], '.mycli.test.log')
+
+
+SELF_CONNECTING_FEATURES = (
+ 'test/features/connection.feature',
+)
+
+
+MY_CNF_PATH = os.path.expanduser('~/.my.cnf')
+MY_CNF_BACKUP_PATH = f'{MY_CNF_PATH}.backup'
+MYLOGIN_CNF_PATH = os.path.expanduser('~/.mylogin.cnf')
+MYLOGIN_CNF_BACKUP_PATH = f'{MYLOGIN_CNF_PATH}.backup'
+
+
+def get_db_name_from_context(context):
+ return context.config.userdata.get(
+ 'my_test_db', None
+ ) or "mycli_behave_tests"
+
+
+
+def before_all(context):
+ """Set env parameters."""
+ os.environ['LINES'] = "100"
+ os.environ['COLUMNS'] = "100"
+ os.environ['EDITOR'] = 'ex'
+ os.environ['LC_ALL'] = 'en_US.UTF-8'
+ os.environ['PROMPT_TOOLKIT_NO_CPR'] = '1'
+ os.environ['MYCLI_HISTFILE'] = os.devnull
+
+ test_dir = os.path.abspath(os.path.dirname(os.path.dirname(__file__)))
+ login_path_file = os.path.join(test_dir, 'mylogin.cnf')
+# os.environ['MYSQL_TEST_LOGIN_FILE'] = login_path_file
+
+ context.package_root = os.path.abspath(
+ os.path.dirname(os.path.dirname(os.path.dirname(__file__))))
+
+ os.environ["COVERAGE_PROCESS_START"] = os.path.join(context.package_root,
+ '.coveragerc')
+
+ context.exit_sent = False
+
+ vi = '_'.join([str(x) for x in sys.version_info[:3]])
+ db_name = get_db_name_from_context(context)
+ db_name_full = '{0}_{1}'.format(db_name, vi)
+
+ # Store get params from config/environment variables
+ context.conf = {
+ 'host': context.config.userdata.get(
+ 'my_test_host',
+ os.getenv('PYTEST_HOST', 'localhost')
+ ),
+ 'port': context.config.userdata.get(
+ 'my_test_port',
+ int(os.getenv('PYTEST_PORT', '3306'))
+ ),
+ 'user': context.config.userdata.get(
+ 'my_test_user',
+ os.getenv('PYTEST_USER', 'root')
+ ),
+ 'pass': context.config.userdata.get(
+ 'my_test_pass',
+ os.getenv('PYTEST_PASSWORD', None)
+ ),
+ 'cli_command': context.config.userdata.get(
+ 'my_cli_command', None) or
+ sys.executable + ' -c "import coverage ; coverage.process_startup(); import mycli.main; mycli.main.cli()"',
+ 'dbname': db_name,
+ 'dbname_tmp': db_name_full + '_tmp',
+ 'vi': vi,
+ 'pager_boundary': '---boundary---',
+ }
+
+ _, my_cnf = mkstemp()
+ with open(my_cnf, 'w') as f:
+ f.write(
+ '[client]\n'
+ 'pager={0} {1} {2}\n'.format(
+ sys.executable, os.path.join(context.package_root,
+ 'test/features/wrappager.py'),
+ context.conf['pager_boundary'])
+ )
+ context.conf['defaults-file'] = my_cnf
+ context.conf['myclirc'] = os.path.join(context.package_root, 'test',
+ 'myclirc')
+
+ context.cn = dbutils.create_db(context.conf['host'], context.conf['port'],
+ context.conf['user'],
+ context.conf['pass'],
+ context.conf['dbname'])
+
+ context.fixture_data = fixutils.read_fixture_files()
+
+
+def after_all(context):
+ """Unset env parameters."""
+ dbutils.close_cn(context.cn)
+ dbutils.drop_db(context.conf['host'], context.conf['port'],
+ context.conf['user'], context.conf['pass'],
+ context.conf['dbname'])
+
+ # Restore env vars.
+ #for k, v in context.pgenv.items():
+ # if k in os.environ and v is None:
+ # del os.environ[k]
+ # elif v:
+ # os.environ[k] = v
+
+
+def before_step(context, _):
+ context.atprompt = False
+
+
+def before_scenario(context, arg):
+ with open(test_log_file, 'w') as f:
+ f.write('')
+ if arg.location.filename not in SELF_CONNECTING_FEATURES:
+ run_cli(context)
+ wait_prompt(context)
+
+ if os.path.exists(MY_CNF_PATH):
+ shutil.move(MY_CNF_PATH, MY_CNF_BACKUP_PATH)
+
+ if os.path.exists(MYLOGIN_CNF_PATH):
+ shutil.move(MYLOGIN_CNF_PATH, MYLOGIN_CNF_BACKUP_PATH)
+
+
+def after_scenario(context, _):
+ """Cleans up after each test complete."""
+ with open(test_log_file) as f:
+ for line in f:
+ if 'error' in line.lower():
+ raise RuntimeError(f'Error in log file: {line}')
+
+ if hasattr(context, 'cli') and not context.exit_sent:
+ # Quit nicely.
+ if not context.atprompt:
+ user = context.conf['user']
+ host = context.conf['host']
+ dbname = context.currentdb
+ context.cli.expect_exact(
+ '{0}@{1}:{2}>'.format(
+ user, host, dbname
+ ),
+ timeout=5
+ )
+ context.cli.sendcontrol('c')
+ context.cli.sendcontrol('d')
+ context.cli.expect_exact(pexpect.EOF, timeout=5)
+
+ if os.path.exists(MY_CNF_BACKUP_PATH):
+ shutil.move(MY_CNF_BACKUP_PATH, MY_CNF_PATH)
+
+ if os.path.exists(MYLOGIN_CNF_BACKUP_PATH):
+ shutil.move(MYLOGIN_CNF_BACKUP_PATH, MYLOGIN_CNF_PATH)
+ elif os.path.exists(MYLOGIN_CNF_PATH):
+ # This file was moved in `before_scenario`.
+ # If it exists now, it has been created during a test
+ os.remove(MYLOGIN_CNF_PATH)
+
+
+# TODO: uncomment to debug a failure
+# def after_step(context, step):
+# if step.status == "failed":
+# import ipdb; ipdb.set_trace()