_testutils.py

from __future__ import annotations

import os
import subprocess
import sys
import time
from enum import Enum
from typing import Iterator

from swet._build_utils import _recompile_sweb
from swet._utils import find_files
from swet._utils import get_injected_user_progs
from swet._utils import TestFile


class Result(Enum):
    passed = 'PASSED'
    failed = 'FAILED'
    skipped = 'SKIPPED'


class Colors(Enum):
    red = '\033[41m'
    green = '\033[42m'
    yellow = '\033[43;30m'
    normal = '\033[m'


TEST_RESULT_COLORS = {
    Result.passed: Colors.green,
    Result.failed: Colors.red,
    Result.skipped: Colors.yellow,
}

QEMU_EXECUTE_COMMAND = [
    'qemu-system-x86_64',

    '-m',
    '8M',

    '-cpu',
    'qemu64',

    '-nographic',

    '-drive',
    'file=SWEB.qcow2,index=0,media=disk',

    '-monitor',
    'none',
]


def _print_testresult(
    testname: str,
    test_status: Result,
    use_colors=True,
) -> None:

    test_color = TEST_RESULT_COLORS[test_status].value
    result = test_status.value
    termreset = Colors.normal.value

    if use_colors:
        print(f'{testname}{test_color}{result}{termreset}')
    else:
        print(f'{testname}{result}')


def follow(
    file,
    sleep_sec: float = 0.5,
    timeout: int = 10,
    use_color: bool = True,
) -> Iterator[str]:
    # SRC: https://stackoverflow.com/a/54263201
    line = ''
    idle_time: float = 0.0

    sys.stdout.write('Executing test ........... ')
    sys.stdout.flush()
    while True:
        tmp = file.readline()

        if tmp != '':
            line += tmp
            if line.endswith('\n'):
                yield line
                line = ''
                idle_time = 0
        else:
            time.sleep(sleep_sec)
            idle_time += sleep_sec
            if idle_time > timeout:
                if use_color:
                    sys.stdout.write(
                        (
                            f'{Colors.green.value}Executed!'
                            f'{Colors.normal.value}\n'
                        ),
                    )
                else:
                    sys.stdout.write('Executed!\n')

                return


def execute_test(
    testname: TestFile,
    build_path: str,
    code_path: str,
    use_colors: bool,
    testreport: bool = False,
) -> int:

    if testreport:
        # TODO: Implement the option to specify if the programm should create
        # a JUNIT XML File
        raise NotImplementedError('This functionality is not jet implemented')

    # TODO: Testen, ob das File schon existiert, wenn ja dann muss entsprechend
    # behandeln
    open(f'{build_path}/{testname.filename_without_ending}.log', 'w').close()

    qemu_command = QEMU_EXECUTE_COMMAND + [
        '-debugcon',
        f'file:{build_path}/{testname.filename_without_ending}.log',
    ]

    qemu_prozess = subprocess.Popen(
        qemu_command,
        cwd=build_path,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
    )

    test_result = Result.passed

    with open(
        f'{build_path}/{testname.filename_without_ending}.log',
    ) as logfile:
        for logline in follow(logfile, use_color=use_colors):
            if 'Assertion failed' in logline:
                test_result = Result.failed

            elif 'KERNEL PANIC' in logline:
                test_result = Result.failed

    qemu_prozess.terminate()

    _print_testresult(
        testname.filename_without_ending.ljust(30, '.'),
        test_status=test_result,
        use_colors=use_colors,
    )

    if test_result == Result.failed:
        return -1
    else:
        return 0


def _run_tests(
    tests_to_run: list[TestFile],
    build_path: str,
    use_colors: bool,
    testreport: bool,
) -> int:

    user_progs_files = find_files('.', 'user_progs.h')  # should only be one

    if len(user_progs_files) == 0:
        print(
            'No user_progs.h-File found!',
            file=sys.stderr,
        )
        return -1

    if len(user_progs_files) > 1:
        found_files = ', '.join(user_progs_files)
        print(
            (
                'There where multiple files found'
                f'for user_progs.h: {found_files}'
            ),
            file=sys.stderr,
        )
        return -1

    user_progs_file = user_progs_files[0]

    # make sure the build-dir exists/create
    # equivalent to mkdir -p <foldername>
    if not os.path.isdir(build_path):
        print(
            f'Path, {build_path} does not exist and will be created',
            file=sys.stderr,
        )

    os.makedirs(build_path, exist_ok=True)

    # Make sure, that there is a folder called "userspace" and an folder called
    # common in the current Directory
    code_path = os.path.realpath('.')

    if not (
        os.path.isdir(f'{code_path}/common') and
        os.path.isdir(f'{code_path}/userspace')
    ):
        print(
            'Please run sweb-tester inside the sweb-repository',
            file=sys.stderr,
        )

    ret_val = 0
    for test in tests_to_run:
        print(f'Running Test {test.filename_without_ending}')

        patched_user_progs = get_injected_user_progs(test)

        with open(user_progs_file, 'w') as user_progs_fileoperator:
            user_progs_fileoperator.write(patched_user_progs)

        compile_ret_val = _recompile_sweb(build_path, code_path)

        if compile_ret_val == -1:
            # Cancel the execution
            print(
                'Abort the test, because the compile process failed',
                file=sys.stderr,
            )
            return -1

        # TODO: pass headless and testreport to the function
        exec_test_ret_val = execute_test(
            test,
            build_path,
            code_path,
            use_colors,
        )
        ret_val |= exec_test_ret_val

    return ret_val