#!/usr/bin/env python3
# encoding: utf-8
"""
This script runs a sequence of wml unit test scenarios.
"""

import argparse, enum, os, re, subprocess, sys

class UnexpectedTestStatusException(Exception):
    """Exception raised when a unit test doesn't return the expected result."""
    pass

class UnitTestResult(enum.Enum):
    """Enum corresponding to game_launcher.hpp's unit_test_result"""
    PASS = 0
    FAIL = 1
    TIMEOUT = 2
    FAIL_LOADING_REPLAY = 3
    FAIL_PLAYING_REPLAY = 4
    FAIL_BROKE_STRICT = 5
    FAIL_WML_EXCEPTION = 6

class TestCase:
    """Represents a single line of the wml_test_schedule."""
    def __init__(self, status, name):
        self.status = status
        self.name = name

    def __str__(self):
        return "TestCase<{status}, {name}>".format(status=self.status, name=self.name)

class TestListParser:
    """Each line in the list of tests should be formatted:
        <expected return code><space><name of unit test scenario>

    For example:
        0 test_functionality

    Lines beginning # are treated as comments.
    """
    def __init__(self, options):
        self.verbose = options.verbose
        self.filename = options.list

    def get(self):
        status_name_re = re.compile(r"^(\d+) ([\w-]+)$")
        test_list = []
        for line in open(self.filename, mode="rt"):
            line = line.strip()
            if line == "" or line.startswith("#"):
                continue

            x = status_name_re.match(line)
            if x is None:
                print("Could not parse test list file: ", line)

            t = TestCase(UnitTestResult(int(x.groups()[0])), x.groups()[1])
            if self.verbose > 1:
                print(t)
            test_list.append(t)
        return test_list

def get_output_filename(args):
    for i,arg in enumerate(args):
        if arg == "-u":
            return "test-output-"+args[i+1]
    raise RuntimeError("No -u option found!")

def run_with_rerun_for_sdl_video(args, timeout):
    """A wrapper for subprocess.run with a workaround for the issue of travis+18.04
    intermittently failing to initialise SDL.
    """
    # Sanity check on the number of retries. It's a rare failure, a single retry would probably
    # be enough.
    sdl_retries = 0
    while sdl_retries < 10:
        # For compatibility with Ubuntu 16.04 LTS, this has to run on Python3.5,
        # so the capture_output argument is not available.
        filename = get_output_filename(args)
        res = subprocess.run(args, timeout=timeout, stdout=open(filename, "w"), stderr=subprocess.STDOUT)
        retry = False
        with open(filename, "r") as output:
            if "Could not initialize SDL_video" in output.read():
                retry = True
        if not retry:
        	  return res
        sdl_retries += 1
        print("Could not initialise SDL_video error, attempt", sdl_retries)

class WesnothRunner:
    def __init__(self, options):
        self.verbose = options.verbose
        if options.path is None:
            path = os.path.split(os.path.realpath(sys.argv[0]))[0]
        elif options.path in ["XCode", "xcode", "Xcode"]:
            import glob
            path_list = []
            for build in ["Debug", "Release"]:
                pattern = os.path.join("~/Library/Developer/XCode/DerivedData/Wesnoth*",
                    build, "Build/Products/Release/Wesnoth.app/Contents/MacOS/")
                path_list.extend(glob.glob(os.path.expanduser(pattern)))
            if len(path_list) == 0:
                raise FileNotFoundError("Couldn't find your xcode build dir")
            if len(path_list) > 1:
                # seems better than choosing one at random
                raise RuntimeError("Found more than one xcode build dir")
            path = path_list[0]
        else:
            path = options.path
        path += "/wesnoth"
        if options.debug_bin:
            path += "-debug"
        self.common_args = [path]
        if options.strict_mode:
            self.common_args.append("--log-strict=warning")
        if options.clean:
            self.common_args.append("--noaddons")
        if options.additional_arg is not None:
            self.common_args.extend(options.additional_arg)
        self.timeout = options.timeout
        self.batch_timeout = options.batch_timeout
        if self.verbose > 1:
            print("Options that will be used for all Wesnoth instances:", repr(self.common_args))

    def run_tests(self, test_list):
        """Run all of the tests in a single instance of Wesnoth"""
        if len(test_list) == 0:
            raise ValueError("Running an empty test list")
        if len(test_list) > 1:
            for test in test_list:
                if test.status != UnitTestResult.PASS:
                    raise NotImplementedError("run_tests doesn't yet support batching tests with non-zero statuses")
        expected_result = test_list[0].status
        args = self.common_args.copy()
        for test in test_list:
            args.append("-u")
            args.append(test.name)
        if self.timeout == 0:
            timeout = None
        else:
            if len(test_list) == 1:
                timeout = self.timeout
            else:
                timeout = self.batch_timeout
        if len(test_list) == 1:
            print("Running test", test_list[0].name)
        else:
            print("Running {count} tests ({names})".format(count=len(test_list),
                names=", ".join([test.name for test in test_list])))
        if self.verbose > 1:
            print(repr(args))
        try:
            res = run_with_rerun_for_sdl_video(args, timeout)
        except subprocess.TimeoutExpired as e:
            print("Timed out (killed by Python timeout implementation)")
            res = subprocess.CompletedProcess(args, UnitTestResult.TIMEOUT.value)
        if self.verbose > 1:
            print("Result:", res.returncode)
        if res.returncode < 0:
            print("Wesnoth exited because of signal", -res.returncode)
            if options.backtrace:
                print("Launching GDB for a backtrace...")
                gdb_args = ["gdb", "-q", "-batch", "-ex", "start", "-ex", "continue", "-ex", "bt", "-ex", "quit", "--args"]
                gdb_args.extend(args)
                subprocess.run(gdb_args, timeout=240)
            raise UnexpectedTestStatusException()
        if res.returncode != expected_result.value:
            with open(get_output_filename(args), "r") as output:
                for line in output.readlines():
                	print(line)
            print("Failure, Wesnoth returned", res.returncode, "but we expected", expected_result.value)
            raise UnexpectedTestStatusException()

def test_batcher(test_list):
    """A generator function that collects tests into batches which a single
    instance of Wesnoth can run.
    """
    expected_to_pass = []
    for test in test_list:
        if test.status == UnitTestResult.PASS:
            expected_to_pass.append(test)
        else:
            yield [test]
    yield expected_to_pass

if __name__ == '__main__':
    ap = argparse.ArgumentParser()
    # The options that are mandatory to support (because they're used in the Travis script)
    # are the one-letter forms of verbose, clean, timeout and backtrace.
    ap.add_argument("-v", "--verbose", action="count", default=0,
        help="Verbose mode. Use -v twice for very verbose mode.")
    ap.add_argument("-c", "--clean", action="store_true",
        help="Clean mode. (Don't load any add-ons. Used for mainline tests.)")
    ap.add_argument("-a", "--additional_arg", action="append",
        help="Additional arguments to go to wesnoth. For options that start with a hyphen, '--add_argument --data-dir' will give an error, use '--add_argument=--data-dir' instead.")
    ap.add_argument("-t", "--timeout", type=int, default=10,
        help="New timer value to use, instead of 10s as default. The value 0 means no timer, and also skips tests that expect timeout.")
    ap.add_argument("-bt", "--batch-timeout", type=int, default=300,
        help="New timer value to use for batched tests, instead of 300s as default.")
    ap.add_argument("-s", "--no-strict", dest="strict_mode", action="store_false",
        help="Disable strict mode. By default, we run wesnoth with the option --log-strict=warning to ensure errors result in a failed test.")
    ap.add_argument("-d", "--debug_bin", action="store_true",
        help="Run wesnoth-debug binary instead of wesnoth.")
    ap.add_argument("-g", "--backtrace", action="store_true",
        help="If we encounter a crash, generate a backtrace using gdb. Must have gdb installed for this option.")
    ap.add_argument("-p", "--path", metavar="dir",
        help="Path to wesnoth binary. By default assume it is with this script.")
    ap.add_argument("-l", "--list", metavar="filename",
        help="Loads list of tests from the given file.",
    default="wml_test_schedule")

    # Workaround for argparse not accepting option values that start with a hyphen,
    # for example "-a --user-data-dir". https://bugs.python.org/issue9334
    # New callers can use "-a=--user-data-dir", but compatibility with the old version
    # of run_wml_tests needs support for "-a --user-data-dir".
    try:
        while True:
            i = sys.argv.index("-a")
            sys.argv[i] = "=".join(["-a", sys.argv.pop(i + 1)])
    except IndexError:
        pass
    except ValueError:
        pass

    options = ap.parse_args()

    if options.verbose > 1:
        print(repr(options))

    test_list = TestListParser(options).get()
    runner = WesnothRunner(options)

    a_test_failed = False
    for batch in test_batcher(test_list):
        try:
            runner.run_tests(batch)
        except UnexpectedTestStatusException:
            a_test_failed = True

    if a_test_failed:
        sys.exit(1)