~widelands-dev/widelands/trunk

#!/usr/bin/env python3
# encoding: utf-8

from glob import glob
import argparse
import os
import re
import shutil
import subprocess
import sys
import tempfile
import concurrent.futures as cf
import time
import datetime
import multiprocessing

get_time = time.monotonic

# psutil is not part of the standard library.
# It's only used to get the amount of memory for -j 0
try:
    import psutil
except:
    psutil = None

def datadir():
    return os.path.join(os.path.dirname(__file__), "data")

def datadir_for_testing():
    return os.path.relpath(".", os.path.dirname(__file__))

ansi_colors = {
    'black':   '\033[30m',
    'red':     '\033[31m',
    'green':   '\033[32m',
    'yellow':  '\033[33m',
    'blue':    '\033[34m',
    'purple':  '\033[35m',
    'cyan':    '\033[36m',
    'white':   '\033[37m',
    'default': '\033[39m'
}

info_color      = 'cyan'
success_color   = 'green'
warning_color   = 'yellow'
error_color     = 'red'
separator_color = 'purple'

log_colors = {
    # order in decreasing priority
    'ERROR':   error_color,
    'WARNING': warning_color,
}

use_colors = True

def colorize(text, color):
    if not use_colors:
        return text
    return f'{ansi_colors[color]}{text}{ansi_colors["default"]}'

def colorize_log(text):
    if not use_colors:
        return text
    for key,color in log_colors.items():
        if key in text:
            return colorize(text, color)
    return text

group_start = '\n'
group_end = ''
if os.getenv('GITHUB_ACTION'):
    group_start = '\n::group::'
    group_end = '\n::endgroup::\n'


class WidelandsTestCase():
    do_use_random_directory = True
    path_to_widelands_binary = None
    keep_output_around = False
    ignore_error_code = False
    timeout = 600
    total_tests = 0

    statuses = {
        'Starting':  'Start',
        'Loading':   'Load ',
        'Passed':    'Done ',
        'FAILED':    'FAIL ',
        'TIMED OUT': 'TMOUT',
        'SKIPPED':   'SKIP ',
        'IGNORED':   ' IGN ',
        'Info':      'Info '
    }

    status_colors = {
        'Done ': success_color,
        'FAIL ': error_color,
        'TMOUT': error_color,
        'SKIP ': warning_color,
        ' IGN ': warning_color,
        'Info ': info_color
    }

    annotate = [ 'SKIP ', ' IGN ' ]

    def __init__(self, test_script, **wlargs):
        self.test_script = test_script
        self._shortname = os.path.basename(test_script)
        self._wlargs = wlargs
        self.success = None
        self.result = 'NOT EXECUTED'
        self.report_header = None
        # 'which_time' is an integer defining the number of times Widelands has run this test
        # case (i.e. because we might load a saved game from an earlier run). This will impact
        # the filenames for stdout.txt.
        self.which_time = 0
        self.outputs = []
        WidelandsTestCase.total_tests += 1
        self._test_number = WidelandsTestCase.total_tests

    def run_widelands(self, wlargs):
        """Run Widelands with the given 'wlargs'.

        Returns the stdout filename."""
        stdout_filename = os.path.join(self.run_dir, f'stdout_{self.which_time:02d}.txt')
        if (os.path.exists(stdout_filename)):
            os.unlink(stdout_filename)

        with open(stdout_filename, 'a') as stdout_file:
            args = [self.path_to_widelands_binary,
                    '--verbose',
                    f'--datadir={datadir()}',
                    f'--datadir_for_testing={datadir_for_testing()}',
                    f'--homedir={self.run_dir}',
                    '--nosound',
                    '--fail-on-lua-error',
                    '--fail-on-errors',
                    '--language=en' ]
            args += [ f'--{key}={value}' for key, value in wlargs.items() ]

            stdout_file.write('Running widelands binary: ')
            for anarg in args:
              stdout_file.write(anarg)
              stdout_file.write(" ")
            stdout_file.write("\n")
            stdout_file.flush()

            start_time = get_time()
            widelands = subprocess.Popen(
                    args, shell=False, stdout=stdout_file, stderr=subprocess.STDOUT)
            try:
                widelands.communicate(timeout = self.timeout)
            except subprocess.TimeoutExpired:
                widelands.kill()
                widelands.communicate()
                self.wl_timed_out = True
                stdout_file.write('\n')
                stdout_file.write(colorize('Timed out.', error_color))
                stdout_file.write('\n')
            end_time = get_time()
            stdout_file.flush()
            self.duration = datetime.timedelta(seconds = end_time - start_time)
            stdout_file.write('\n')
            stdout_file.write(
                colorize(f'{self.step_name()}: Returned from Widelands in {self.duration}, ' \
                f'return code is {widelands.returncode:d}', info_color))
            stdout_file.write('\n')
            self.widelands_returncode = widelands.returncode
        self.outputs.append(stdout_filename)
        return stdout_filename

    def run(self):
        num_w = len(str(self.total_tests))
        self._progress = str(self._test_number).rjust(num_w) + '/' + str(self.total_tests)

        if self.do_use_random_directory:
            self.run_dir = tempfile.mkdtemp(prefix="widelands_regression_test_")
        else:
            self.run_dir = os.path.join(tempfile.gettempdir(), "widelands_regression_test", self.__class__.__name__)
            if os.path.exists(self.run_dir):
                if not self.keep_output_around:
                    shutil.rmtree(self.run_dir)
                    os.makedirs(self.run_dir)
            else:
                os.makedirs(self.run_dir)

        self.widelands_returncode = 0
        self.wl_timed_out = False
        self.current_step = ""

        self.out_status('Start', f'{self.test_script} starting ...')
        stdout_filename = self.run_widelands(self._wlargs)
        stdout = open(stdout_filename, "r").read()
        self.verify_success(stdout, stdout_filename)

        find_saves = lambda stdout: re.findall(r'Script requests save to: (\w+)$', stdout, re.M)
        savegame_done = { fn: False for fn in find_saves(stdout) }
        while self.success and not all(savegame_done.values()):
            self.which_time += 1
            for savegame in sorted(savegame_done):
                if not savegame_done[savegame]: break
            self.current_step = f"{savegame}.wgf"
            self.out_status("Load ", "loading savegame ...")
            stdout_filename = self.run_widelands({ "loadgame": os.path.join(
                self.run_dir, "save", f"{savegame}.wgf")})
            stdout = open(stdout_filename, "r").read()
            for new_save in find_saves(stdout):
                if new_save not in savegame_done:
                    savegame_done[new_save] = False
            savegame_done[savegame] = True
            self.verify_success(stdout, stdout_filename)

        if self.report_header == None and not self.keep_output_around:
            shutil.rmtree(self.run_dir)

    def step_name(self):
        if self.which_time == 0:
            return self._shortname
        return f'{self._shortname} / {self.current_step}'

    def out_status(self, status, message):
        # Force writing to main test log immediately

        # On github, also create an annotation for non-fatal problems that would
        # otherwise go unnoticed
        if (status in self.annotate) and os.getenv('GITHUB_ACTION'):
            sys.stdout.write(f'::warning::{self.step_name()}: {message}\n')
            sys.stdout.flush()

        if use_colors and status in self.status_colors.keys():
            color = self.status_colors[status]
            status = colorize(status, color)
            message = colorize(message, color)
        sys.stdout.write(f'{self._progress} {status} {self.step_name()}: {message}\n')
        sys.stdout.flush()

    def get_result_color(self):
        if not use_colors:
            return 'default'
        status = self.statuses[self.result]
        if status in self.status_colors.keys():
            return self.status_colors[status]
        return 'default'

    def out_result(self, stdout_filename):
        self.out_status(self.statuses[self.result], f'{self.result} in {self.duration}')
        if self.keep_output_around:
            self.out_status('Info ', f'stdout: {stdout_filename}')

    def fail(self, short, long, stdout_filename):
        self.success = False
        self.result = short
        self.out_result(stdout_filename)
        long = colorize(long, self.status_colors['FAIL '])
        self.report_header = f'{long} Analyze the files in {self.run_dir} to see why this test case failed.\n'
        self.report_header += colorize(f'Stdout is: {stdout_filename}', info_color)

    def step_success(self, stdout_filename):
        old_result = self.result
        self.result = "Passed"
        self.out_result(stdout_filename)
        if self.which_time == 0:
            self.success = True
        else:
            self.result = old_result

    def verify_success(self, stdout, stdout_filename):
        # Catch instabilities with SDL in CI environment
        if self.widelands_returncode == 2:
            # Print stdout in the final summary with this header
            self.result = "SKIPPED"
            self.report_header = 'SDL initialization failed. TEST SKIPPED.'
            self.out_status('SKIP ', self.report_header)
            if use_colors:
                self.report_header = colorize(self.report_header, self.status_colors['SKIP '])
            if self.which_time == 0:  # must set it for the first run, later just ignore
                self.success = True
        else:
            if self.wl_timed_out:
                self.fail("TIMED OUT", "The test timed out.", stdout_filename)
                return
            if self.widelands_returncode != 0:
                if self.widelands_returncode == 1 and self.ignore_error_code:
                    self.out_status(' IGN ', f'IGNORING error code 1')
                else:
                    self.fail("FAILED", "Widelands exited abnormally.", stdout_filename)
                    return
            if not "All Tests passed" in stdout or "lua_errors.cc" in stdout:
                self.fail("FAILED", "Not all tests pass.", stdout_filename)
                return
            self.step_success(stdout_filename)

    def print_report(self):
        print(f'{colorize(self.result, self.get_result_color())}: {self.test_script}\n')
        print(self.report_header)
        print(group_start, end='')
        print(colorize("stdout:", info_color))
        for stdout_fn in self.outputs:
            with open(stdout_fn, "r") as stdout:
                for line in stdout:
                    line = colorize_log(line)
                    print(line, end='')
        print(group_end, end='')


# For parallel execution of tests
def recommended_workers(binary):
    cpu_count = multiprocessing.cpu_count()

    # Widelands uses 2 threads, but the logic thread is much more heavy,
    # so we give each Widelands instance one full CPU for the logic thread
    # while we let the graphic threads of up to 3 instances share one CPU.
    max_threads_cpu = 1
    if cpu_count > 2 :
        max_threads_cpu = cpu_count - 1 - (cpu_count - 1) // 4

    # Also test memory as a limiting factor
    if psutil != None:
        mem_per_instance = 1500 # MB - default for debug builds

        widelands = subprocess.run([binary, '--version'], shell=False, encoding='utf-8', capture_output=True)
        firstline = widelands.stdout.splitlines(keepends=False)[0]
        firstline_OK = False
        if firstline.startswith('This is Widelands version'):
            if firstline.endswith('Release'):
                mem_per_instance = 800 # MB
                firstline_OK = True
            elif firstline.endswith('Debug'):
                firstline_OK = True
        if not firstline_OK:
            print('Cannot parse build type from stdout:', firstline)

        max_threads_mem = max(1, psutil.virtual_memory().available // (mem_per_instance * 1000 * 1000))
        mem_avail = psutil.virtual_memory().available / 1000 / 1000 / 1000 # GB, not GiB :)
        mem_total = psutil.virtual_memory().total / 1000 / 1000 / 1000
        print(f"{cpu_count} CPUs, {mem_avail:.1f}/{mem_total:.1f} gigabytes memory free")
        return min(max_threads_cpu, max_threads_mem)
    else:
        return max_threads_cpu

def find_binary():
    # Prefer binary from source directory
    for potential_binary in (
        glob(os.path.join(os.curdir, "widelands")) +
        glob(os.path.join(os.path.dirname(__file__), "widelands")) +
        glob(os.path.join("build", "src", "widelands")) +
        glob(os.path.join("src", "widelands")) +
        glob(os.path.join("..", "*", "src", "widelands"))
    ):
        if os.access(potential_binary, os.X_OK):
            return potential_binary

    # Fall back to binary in $PATH if possible
    return shutil.which("widelands")

    return None

def check_binary(binary):
    # Prefer binary from source directory instead of PATH
    if os.path.dirname(binary) == '' and os.access(binary, os.X_OK):
        return os.path.join(os.curdir, binary)

    if shutil.which(binary) != None:
        return binary

    if os.path.dirname(binary) != '' and os.access(binary, os.X_OK):
        return binary

    for potential_path in [ os.curdir, os.path.dirname(__file__) ]:
        fullpath = os.path.join(potential_path, binary)
        if os.access(fullpath, os.X_OK):
            return fullpath

    return None

def parse_args():
    p = argparse.ArgumentParser(description=
        "Run the regression tests suite."
    )

    p.add_argument("-r", "--regexp", type=str,
        help = "Run only the tests from the files which filename matches."
    )
    p.add_argument("-n", "--nonrandom", action="store_true", default = False,
        help = "Do not randomize the directories for the tests. This is useful "
        "if you want to run a test more often than once and not reopen stdout.txt "
        "in your editor."
    )
    p.add_argument("-k", "--keep-around", action="store_true", default = False,
        help = "Keep the output files around even when a test terminates successfully."
    )
    p.add_argument("-b", "--binary", type=str,
        help = "Run this binary as Widelands. Otherwise some default paths are searched."
    )
    p.add_argument("-i", "--ignore-error-code", action="store_true", default = False,
        help = "Assume success on return code 1, to allow running the tests "
        "without ASan reporting false positives."
    )
    p.add_argument("-j", "--workers", type=int, default = 0,
        help = "Use this many parallel workers."
    )
    p.add_argument("-p", "--plain", "--nocolor", action="store_true", default = False,
        help = "Don't use ANSI color sequences in the output."
    )
    p.add_argument("-t", "--timeout", type=float, default = "10",
        help = "Set the timeout duration for test cases in minutes. Default is 10 minutes."
    )

    args = p.parse_args()

    if args.binary is None:
        args.binary = find_binary()
        if args.binary is None:
            p.error("No widelands binary found. Please specify with -b.")
    else:
        args.binary = check_binary(args.binary)
        if args.binary is None:
            p.error("The specified widelands binary is not found.")

    if args.nonrandom:
        if args.workers != 1:
            args.workers = 1
            print("Only one worker is possible with --nonrandom!")
        if args.keep_around:
            print("--nonrandom is not recommended with --keep-around, some files will be overwritten!")

    if args.workers == 0:
        args.workers = recommended_workers(args.binary)

    return args


def discover_loadgame_tests(regexp, suite):
    """Add all tests using --loadgame to the 'suite'."""
    # Savegames with custom scripts
    for fixture in sorted(glob(os.path.join("test", "save", "*"))):
        if not os.path.isdir(fixture):
            continue
        savegame = sorted(glob(os.path.join(fixture, "*.wgf")))[0]
        for test_script in sorted(glob(os.path.join(fixture, "test*.lua"))):
            if regexp is not None and not re.search(regexp, test_script):
                continue
            suite.append(
                    WidelandsTestCase(test_script,
                        loadgame=savegame, script=test_script))
    # Savegames without custom script, just test loading
    test_script = os.path.join("test", "scripting", "load_and_quit.lua")
    for savegame in sorted(glob(os.path.join("test", "save", "*.wgf"))):
        if regexp is not None and not re.search(regexp, savegame):
            continue
        suite.append(WidelandsTestCase(savegame, loadgame=savegame, script=test_script))

def discover_scenario_tests(regexp, suite):
    """Add all tests using --scenario to the 'suite'."""
    for wlmap in sorted(glob(os.path.join("test", "maps", "*"))):
        if not os.path.isdir(wlmap):
            continue
        for test_script in sorted(glob(os.path.join(wlmap, "scripting", "test*.lua"))):
            if regexp is not None and not re.search(regexp, test_script):
                continue
            suite.append(
                    WidelandsTestCase(test_script,
                        scenario=wlmap, script=test_script))

def discover_game_template_tests(regexp, suite):
    """Add all tests using --new_game_from_template to the 'suite'."""
    for templ in sorted(glob(os.path.join("test", "templates", "test*.wgt"))):
        if not os.path.isfile(templ):
            continue
        test_script = templ[:-3] + 'lua'
        if not os.path.isfile(test_script):
            print(f"WARNING: Game template test { templ }: corresponding script { test_script } not found - Skipping.")
            continue
        if regexp is not None and not re.search(regexp, test_script):
            continue
        suite.append(
                WidelandsTestCase(test_script,
                    new_game_from_template=templ, script=test_script))

def discover_editor_tests(regexp, suite):
    """Add all tests needing --editor to the 'suite'."""
    for wlmap in sorted(glob(os.path.join("test", "maps", "*"))):
        if not os.path.isdir(wlmap):
            continue
        for test_script in sorted(glob(os.path.join(wlmap, "scripting", "editor_test*.lua"))):
            if regexp is not None and not re.search(regexp, test_script):
                continue
            suite.append(
                    WidelandsTestCase(test_script,
                        editor=wlmap, script=test_script))

def main():
    args = parse_args()

    global use_colors
    use_colors = not args.plain
    WidelandsTestCase.path_to_widelands_binary = args.binary
    print(f"Using '{args.binary}' binary.")
    WidelandsTestCase.do_use_random_directory = not args.nonrandom
    WidelandsTestCase.keep_output_around = args.keep_around
    WidelandsTestCase.ignore_error_code = args.ignore_error_code
    WidelandsTestCase.timeout = args.timeout * 60

    test_cases = []
    discover_loadgame_tests(args.regexp, test_cases)
    discover_scenario_tests(args.regexp, test_cases)
    discover_game_template_tests(args.regexp, test_cases)
    discover_editor_tests(args.regexp, test_cases)

    print(f"Will run { len(test_cases) } tests with { args.workers } workers.\n")

    start_time = get_time()

    if args.workers == 1:
        # Single-threaded execution is special-cased for nicer grouping.
        for test_case in test_cases:
            test_case.run()
    else:
        # Parallel execution
        with cf.ThreadPoolExecutor(max_workers = args.workers) as executor:
            futures = {executor.submit(test_case.run): test_case for test_case in test_cases}
            for future in cf.as_completed(futures):
                if future.exception():
                    raise future.exception()

    end_time = get_time()

    separator = '\n' + \
        colorize('---------------------------------------------------------------------------',
            separator_color) + '\n'

    nr_errors = 0
    results = dict()
    for test_case in test_cases:
        # Skipped test cases are logged, but don't count as failure
        if test_case.report_header != None:
            print(separator)
            test_case.print_report()
            if test_case.result in results.keys():
                results[test_case.result].append(test_case.test_script)
            else:
                results[test_case.result] = [ test_case.test_script ]
        if not test_case.success:
            nr_errors += 1

    print(separator)

    summary_common = f'Ran {len(test_cases)} test cases in {(end_time - start_time):.3f} s,'

    if nr_errors == 0:
        print(summary_common, colorize('all tests passed.', success_color))
        return True

    for result,tests in results.items():
        print(f'{len(tests)} tests {result}:')
        for test_name in tests:
            print(f'     {test_name}')
    print(separator)
    print(summary_common, f'{len(test_cases) - nr_errors} tests passed,',
          colorize(f'{nr_errors} tests failed!', error_color))
    return False

if __name__ == '__main__':
    sys.exit(0 if main() else 1)