#!/usr/bin/python3

from __future__ import print_function

import argparse

import datetime

import os

import re

import six

import subprocess

import sys

import unittest

import yaml

from distutils.spawn import find_executable

LIBDIRS = 'src/utils/.libs:src/plugins/.libs:src/plugins/fs/.libs:src/lib/.libs'

GIDIR = 'src/lib'

SKIP_CONFIG = 'skip.yml'

def _get_tests_from_suite(suite, tests):

    """ Extract tests from the test suite """

    # 'tests' we get from 'unittest.defaultTestLoader.discover' are "wrapped"

    # in multiple 'unittest.suite.TestSuite' classes/lists so we need to "unpack"

    # the indivudual test cases

    for test in suite:

        if isinstance(test, unittest.suite.TestSuite):

            _get_tests_from_suite(test, tests)

        if isinstance(test, unittest.TestCase):

            tests.append(test)

    return tests

def _get_test_tags(test):

    """ Get test tags for single test case """

    tags = []

    # test failed to load, usually some ImportError or something really broken

    # in the test file, just return empty list and let it fail

    # with python2 the loader will raise an exception directly without returning

    # a "fake" FailedTest test case

    if six.PY3 and isinstance(test, unittest.loader._FailedTest):

        return tags

    test_fn = getattr(test, test._testMethodName)

    # it is possible to either tag a test funcion or the class so we need to

    # check both for the tag

    if getattr(test_fn, "slow", False) or getattr(test_fn.__self__, "slow", False):

        tags.append(TestTags.SLOW)

    if getattr(test_fn, "unstable", False) or getattr(test_fn.__self__, "unstable", False):

        tags.append(TestTags.UNSTABLE)

    if getattr(test_fn, "unsafe", False) or getattr(test_fn.__self__, "unsafe", False):

        tags.append(TestTags.UNSAFE)

    if getattr(test_fn, "core", False) or getattr(test_fn.__self__, "core", False):

        tags.append(TestTags.CORE)

    if getattr(test_fn, "nostorage", False) or getattr(test_fn.__self__, "nostorage", False):

        tags.append(TestTags.NOSTORAGE)

    if getattr(test_fn, "extradeps", False) or getattr(test_fn.__self__, "extradeps", False):

        tags.append(TestTags.EXTRADEPS)

    if getattr(test_fn, "regression", False) or getattr(test_fn.__self__, "regression", False):

        tags.append(TestTags.REGRESSION)

    if getattr(test_fn, "sourceonly", False) or getattr(test_fn.__self__, "sourceonly", False):

        tags.append(TestTags.SOURCEONLY)

    return tags

def parse_args():

    """ Parse cmdline arguments """

    argparser = argparse.ArgumentParser(description='libblockdev test suite')

    argparser.add_argument('testname', nargs='*', help='name of test class or '

                           'method (e. g. "CryptoTestFormat", '

                           '"GenericResize.test_ext2_generic_resize")')

    argparser.add_argument('-f', '--fast', dest='fast', help='skip slow tests',

                           action='store_true')

    argparser.add_argument('-l', '--lucky', dest='lucky',

                           help='run also potentially dangerous/failing tests',

                           action='store_true')

    argparser.add_argument('-j', '--jenkins', dest='jenkins',

                           help='run also tests that should run only in a CI environment',

                           action='store_true')

    argparser.add_argument('-c', '--core', dest='core',

                           help='run tests that cover basic functionality of the library and regression tests',

                           action='store_true')

    argparser.add_argument('-s', '--stop', dest='stop',

                           help='stop executing after first failed test',

                           action='store_true')

    argparser.add_argument('-i', '--installed', dest='installed',

                           help='run tests against installed version of libblockdev',

                           action='store_true')

    args = argparser.parse_args()

    if args.fast:

        os.environ['SKIP_SLOW'] = ''

    if args.lucky:

        os.environ['FEELINGLUCKY'] = ''

    if args.jenkins:

        os.environ['JENKINS_HOME'] = ''

    # read the environmental variables for backwards compatibility

    if 'JENKINS_HOME' in os.environ:

        args.jenkins = True

    if 'SKIP_SLOW' in os.environ:

        args.fast = True

    if 'FEELINGLUCKY' in os.environ:

        args.lucky = True

    return args

def _split_test_id(test_id):

    # test.id() looks like 'crypto_test.CryptoTestResize.test_luks2_resize'

    # and we want to print 'test_luks2_resize (crypto_test.CryptoTestResize)'

    test_desc = test.id().split(".")

    test_name = test_desc[-1]

    test_module = ".".join(test_desc[:-1])

    return test_name, test_module

def _print_skip_message(test, skip_tag):

    test_id = test.id()

    test_module, test_name = _split_test_id(test_id)

    if skip_tag == TestTags.SLOW:

        reason = "skipping slow tests"

    elif skip_tag == TestTags.UNSTABLE:

        reason = "skipping unstable tests"

    elif skip_tag == TestTags.UNSAFE:

        reason = "skipping test that modifies system configuration"

    elif skip_tag == TestTags.EXTRADEPS:

        reason = "skipping test that requires special configuration"

    elif skip_tag == TestTags.CORE:

        reason = "skipping non-core test"

    elif skip_tag == TestTags.SOURCEONLY:

        reason = "skipping test that can run only against library compiled from source"

    else:

        reason = "unknown reason"  # just to be sure there is some default value

    if test._testMethodDoc:

        print("%s (%s)\n%s ... skipped '%s'" % (test_name, test_module, test._testMethodDoc, reason),

              file=sys.stderr)

    else:

        print("%s (%s) ... skipped '%s'" % (test_name, test_module, reason),

              file=sys.stderr)

def _should_skip(distro=None, version=None, arch=None, reason=None):

    # all these can be lists or a single value, so covert everything to list

    if distro is not None and type(distro) is not list:

        distro = [distro]

    if version is not None and type(version) is not list:

        version = [version]

    if arch is not None and type(arch) is not list:

        arch = [arch]

    # DISTRO, VERSION and ARCH variables are set in main, we don't need to

    # call hostnamectl etc. for every test run

    if (distro is None or DISTRO in distro) and (version is None or VERSION in version) and \

       (arch is None or ARCH in arch):

        return True

    return False

def _parse_skip_config(config):

    with open(config) as f:

        data = f.read()

    parsed = yaml.load(data, Loader=yaml.SafeLoader)

    skipped_tests = dict()

    for entry in parsed:

        for skip in entry["skip_on"]:

            if _should_skip(**skip):

                skipped_tests[entry["test"]] = skip["reason"]

    return skipped_tests

if __name__ == '__main__':

    testdir = os.path.abspath(os.path.dirname(__file__))

    projdir = os.path.abspath(os.path.normpath(os.path.join(testdir, '..')))

    args = parse_args()

    if args.installed:

        os.environ['LIBBLOCKDEV_TESTS_SKIP_OVERRIDE'] = ''

        os.environ['LIBBLOCKDEV_CONFIG_DIR'] = '/etc/libblockdev/conf.d/'

    else:

        if 'LD_LIBRARY_PATH' not in os.environ and 'GI_TYPELIB_PATH' not in os.environ:

            os.environ['LD_LIBRARY_PATH'] = LIBDIRS

            os.environ['GI_TYPELIB_PATH'] = GIDIR

            os.environ['LIBBLOCKDEV_CONFIG_DIR'] = os.path.join(testdir, 'default_config')

            try:

                pyver = 'python3' if six.PY3 else 'python'

                os.execv(sys.executable, [pyver] + sys.argv)

            except OSError as e:

                print('Failed re-exec with a new LD_LIBRARY_PATH and GI_TYPELIB_PATH: %s' % str(e))

                sys.exit(1)

        sys.path.append(testdir)

        sys.path.append(projdir)

        sys.path.append(os.path.join(projdir, 'src/python'))

    start_time = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")

    loader = unittest.defaultTestLoader

    suite = unittest.TestSuite()

    if args.testname:

        test_cases = loader.loadTestsFromNames(args.testname)

    else:

        test_cases = loader.discover(start_dir=testdir, pattern='*_test*.py')

    # extract list of test classes so we can check/run them manually one by one

    tests = []

    tests = _get_tests_from_suite(test_cases, tests)

    # get distro and arch here so we don't have to do this for every test

    from utils import get_version

    DISTRO, VERSION = get_version()

    ARCH = os.uname()[-1]

    # get list of tests to skip from the config file

    skipping = _parse_skip_config(os.path.join(testdir, SKIP_CONFIG))

    # for some reason overrides_hack will fail if we import this at the start

    # of the file

    from utils import TestTags

    for test in tests:

        test_id = test.id()

        # get tags and (possibly) skip the test

        tags = _get_test_tags(test)

        if TestTags.SLOW in tags and args.fast:

            _print_skip_message(test, TestTags.SLOW)

            continue

        if TestTags.UNSTABLE in tags and not args.lucky:

            _print_skip_message(test, TestTags.UNSTABLE)

            continue

        if TestTags.UNSAFE in tags or TestTags.EXTRADEPS in tags and not args.jenkins:

            _print_skip_message(test, TestTags.UNSAFE)

            continue

        if TestTags.EXTRADEPS in tags and not args.jenkins:

            _print_skip_message(test, TestTags.EXTRADEPS)

            continue

        if TestTags.SOURCEONLY in tags and args.installed:

            _print_skip_message(test, TestTags.SOURCEONLY)

            continue

        if args.core and TestTags.CORE not in tags and TestTags.REGRESSION not in tags:

            _print_skip_message(test, TestTags.CORE)

            continue

        # check if the test is in the list of tests to skip

        skip_id = next((test_pattern for test_pattern in skipping.keys() if re.search(test_pattern, test_id)), None)

        if skip_id:

            test_name, test_module = _split_test_id(test_id)

            reason = "not supported on this distribution in this version and arch: %s" % skipping[skip_id]

            print("%s (%s)\n%s ... skipped '%s'" % (test_name, test_module,

                                                    test._testMethodDoc, reason),

                  file=sys.stderr)

            continue

        # finally add the test to the suite

        suite.addTest(test)

    result = unittest.TextTestRunner(verbosity=2, failfast=args.stop).run(suite)

    # dump cropped journal to log file

    if find_executable('journalctl'):

        with open('journaldump.log', 'w') as outfile:

            subprocess.call(['journalctl', '-S', start_time], stdout=outfile)

    if result.wasSuccessful():

        sys.exit(0)

    else:

        sys.exit(1)