px4-firmware/test/mavsdk_tests/mavsdk_test_runner.py

#!/usr/bin/env python3

import argparse
import datetime
import fnmatch
import json
import math
import os
import psutil  # type: ignore
import re
import signal
import subprocess
import sys
import time
from logger_helper import color, colorize
import process_helper as ph
from typing import Any, Dict, List, NoReturn, TextIO, Optional
from types import FrameType

try:
    import requests
except ImportError as e:
    print("Failed to import requests: " + str(e))
    print("")
    print("You may need to install it using:")
    print("    pip3 install --user requests")
    print("")
    sys.exit(1)


def main() -> NoReturn:

    parser = argparse.ArgumentParser()
    parser.add_argument("--log-dir",
                        help="Directory for log files", default="logs")
    parser.add_argument("--speed-factor", default=1,
                        help="how fast to run the simulation")
    parser.add_argument("--iterations", type=int, default=1,
                        help="how often to run all tests")
    parser.add_argument("--abort-early", action='store_true',
                        help="abort on first unsuccessful test")
    parser.add_argument("--gui", default=False, action='store_true',
                        help="display the visualization for a simulation")
    parser.add_argument("--model", type=str, default='all',
                        help="only run tests for one model")
    parser.add_argument("--case", type=str, default='all',
                        help="only run tests for one case "
                             "(or multiple cases with wildcard '*')")
    parser.add_argument("--debugger", default="",
                        help="choice from valgrind, callgrind, gdb, lldb")
    parser.add_argument("--upload", default=False, action='store_true',
                        help="Upload logs to logs.px4.io")
    parser.add_argument("--verbose", default=False, action='store_true',
                        help="enable more verbose output")
    parser.add_argument("config_file", help="JSON config file to use")
    parser.add_argument("--build-dir", type=str,
                        default='build/px4_sitl_default/',
                        help="relative path where the built files are stored")
    args = parser.parse_args()

    with open(args.config_file) as json_file:
        config = json.load(json_file)

    if config["mode"] != "sitl" and args.gui:
        print("--gui is not compatible with the mode '{}'"
              .format(config["mode"]))
        sys.exit(1)

    if not is_everything_ready(config, args.build_dir):
        sys.exit(1)

    if args.verbose:
        print("Creating directory: {}".format(args.log_dir))
    os.makedirs(args.log_dir, exist_ok=True)

    tester = Tester(
        config,
        args.iterations,
        args.abort_early,
        args.speed_factor,
        args.model,
        args.case,
        args.debugger,
        args.log_dir,
        args.gui,
        args.verbose,
        args.upload,
        args.build_dir
    )
    signal.signal(signal.SIGINT, tester.sigint_handler)

    sys.exit(0 if tester.run() else 1)


def is_running(process_name: str) -> bool:
    for proc in psutil.process_iter(attrs=['name']):
        if proc.info['name'] == process_name:
            return True
    return False


def is_everything_ready(config: Dict[str, str], build_dir: str) -> bool:
    result = True

    if config['mode'] == 'sitl':
        if is_running('px4'):
            print("px4 process already running\n"
                  "run `killall px4` and try again")
            result = False
        if not os.path.isfile(os.path.join(build_dir, 'bin/px4')):
            print("PX4 SITL is not built\n"
                  "run `DONT_RUN=1 "
                  "make px4_sitl gazebo mavsdk_tests` or "
                  "`DONT_RUN=1 make px4_sitl_default gazebo mavsdk_tests`")
            result = False
        if config['simulator'] == 'gazebo':
            if is_running('gzserver'):
                print("gzserver process already running\n"
                      "run `killall gzserver` and try again")
                result = False
            if is_running('gzclient'):
                print("gzclient process already running\n"
                      "run `killall gzclient` and try again")
                result = False

    if not os.path.isfile(os.path.join(build_dir,
                                       'mavsdk_tests/mavsdk_tests')):
        print("Test runner is not built\n"
              "run `DONT_RUN=1 "
              "make px4_sitl gazebo mavsdk_tests` or "
              "`DONT_RUN=1 make px4_sitl_default gazebo mavsdk_tests`")
        result = False

    return result


class Tester:
    def __init__(self,
                 config: Dict[str, Any],
                 iterations: int,
                 abort_early: bool,
                 speed_factor: float,
                 model: str,
                 case: str,
                 debugger: str,
                 log_dir: str,
                 gui: bool,
                 verbose: bool,
                 upload: bool,
                 build_dir: str):
        self.config = config
        self.build_dir = build_dir
        self.active_runners: List[ph.Runner]
        self.iterations = iterations
        self.abort_early = abort_early
        self.tests = self.determine_tests(config['tests'], model, case)
        self.speed_factor = speed_factor
        self.debugger = debugger
        self.log_dir = log_dir
        self.gui = gui
        self.verbose = verbose
        self.upload = upload
        self.start_time = datetime.datetime.now()
        self.log_fd: Any[TextIO] = None

    @staticmethod
    def wildcard_match(pattern: str, potential_match: str) -> bool:
        return fnmatch.fnmatchcase(potential_match, pattern)

    @staticmethod
    def plural_s(n_items: int) -> str:
        if n_items > 1:
            return "s"
        else:
            return ""

    @staticmethod
    def query_test_cases(build_dir: str, filter: str) -> List[str]:
        cmd = os.path.join(build_dir, 'mavsdk_tests/mavsdk_tests')
        args = ["--list-test-names-only", filter]
        p = subprocess.Popen(
            [cmd] + args,
            stdin=subprocess.PIPE,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT)
        assert p.stdout is not None
        cases = str(p.stdout.read().decode("utf-8")).strip().split('\n')
        return cases

    def determine_tests(self,
                        tests: List[Dict[str, Any]],
                        model: str,
                        case: str) -> List[Dict[str, Any]]:
        for test in tests:
            test['selected'] = (model == 'all' or model == test['model'])
            test['cases'] = dict.fromkeys(
                self.query_test_cases(self.build_dir, test['test_filter']))
            for key in test['cases'].keys():
                test['cases'][key] = {
                    'selected': (test['selected'] and
                                 (case == 'all' or
                                  self.wildcard_match(case, key)))}
        return tests

    def run(self) -> bool:
        self.show_plans()
        self.prepare_for_results()
        self.run_tests()
        self.show_detailed_results()
        self.show_overall_result()
        return self.was_overall_pass()

    def show_plans(self) -> None:
        print()
        print(colorize(
            "About to run {} test case{} for {} selected model{} "
            "({} iteration{}):".
            format(self.num_cases(), self.plural_s(self.num_cases()),
                   self.num_models(), self.plural_s(self.num_models()),
                   self.iterations, self.plural_s(self.iterations)),
            color.BOLD))

        for test in self.tests:
            if not test['selected']:
                print(colorize(colorize(
                    "  - {} (not selected)".format(test['model']),
                    color.BOLD), color.GRAY))
                continue
            print(colorize("  - {}:".format(test['model']), color.BOLD))
            for key, case in test['cases'].items():
                if case['selected']:
                    print("    - '{}'".format(key))
                else:
                    print(colorize("    - '{}' (not selected)".format(key),
                                   color.GRAY))
        print()

    def num_models(self) -> int:
        return sum(map(
            lambda test: 1 if test['selected'] else 0,
            self.tests))

    def num_cases(self) -> int:
        n_cases = 0
        for test in self.tests:
            if not test['selected']:
                continue
            n_cases += self.num_cases_for_test(test)

        return n_cases

    def num_cases_for_test(self, test: Dict[str, Any]) -> int:
        n_cases = 0
        for case in test['cases'].values():
            if not case['selected']:
                continue
            n_cases += 1
        return n_cases

    def prepare_for_results(self) -> None:
        for test in self.tests:
            if not test['selected']:
                continue
            for key, value in test['cases'].items():
                if not value['selected']:
                    continue
                value['results'] = []

    def run_tests(self) -> None:
        for iteration in range(self.iterations):
            if self.iterations > 1:
                print(colorize("%%% Test iteration: {} / {}".
                      format(iteration + 1, self.iterations), color.BOLD))

            success = self.run_iteration(iteration)
            if not success:
                if self.abort_early:
                    break

    def run_iteration(self, iteration: int) -> bool:
        iteration_success = True
        for test in self.tests:
            if not test['selected']:
                continue

            print(colorize(
                "==> Running tests for {}".format(test['model']),
                color.BOLD))

            test_i = 0
            for key, case_value in test['cases'].items():
                if not case_value['selected']:
                    continue

                print("--> Test case {} of {}: '{}' running ...".
                      format(test_i+1,
                             self.num_cases_for_test(test),
                             key))

                log_dir = self.get_log_dir(iteration, test['model'], key)
                if self.verbose:
                    print("Creating log directory: {}"
                          .format(log_dir))
                os.makedirs(log_dir, exist_ok=True)

                was_success = self.run_test_case(test, key, log_dir)

                print("--- Test case {} of {}: '{}' {}."
                      .format(test_i+1,
                              self.num_cases_for_test(test),
                              key,
                              colorize("succeeded", color.GREEN)
                              if was_success
                              else colorize("failed", color.RED)))

                if not was_success:
                    iteration_success = False

                if not was_success and self.abort_early:
                    print("Aborting early")
                    return False

                test_i += 1

        return iteration_success

    def get_log_dir(self, iteration: int, model: str, case: str) -> str:
        date_and_time = self.start_time.strftime("%Y-%m-%dT%H-%M-%SZ")
        foldername = os.path.join(self.log_dir, date_and_time)

        if self.iterations > 1:
            # Use as many zeros in foldername as required.
            digits = math.floor(math.log10(self.iterations)) + 1
            foldername = os.path.join(foldername,
                                      str(iteration + 1).zfill(digits))

        foldername = os.path.join(foldername, model)
        foldername = os.path.join(foldername, case.replace(" ", "_"))

        return foldername

    def get_max_speed_factor(self, test: Dict[str, Any]) -> float:
        speed_factor = self.speed_factor
        if "max_speed_factor" in test:
            speed_factor = min(float(speed_factor), test["max_speed_factor"])
        return speed_factor

    def run_test_case(self, test: Dict[str, Any],
                      case: str, log_dir: str) -> bool:

        logfile_path = self.determine_logfile_path(log_dir, 'combined')
        self.start_combined_log(logfile_path)

        self.start_runners(log_dir, test, case)

        test_timeout_s = test['timeout_min']*60
        while self.active_runners[-1].time_elapsed_s() < test_timeout_s:
            returncode = self.active_runners[-1].poll()

            self.collect_runner_output()

            if returncode is not None:
                is_success = (returncode == 0)
                break

        else:
            print(colorize(
                "Test timeout of {} mins triggered!".
                format(test['timeout_min']),
                color.BOLD))
            is_success = False

        self.stop_runners()
        # Collect what was left in output buffers.
        self.collect_runner_output()
        self.stop_combined_log()

        result = {'success': is_success,
                  'logfiles': [runner.get_log_filename()
                               for runner in self.active_runners]}
        test['cases'][case]['results'].append(result)

        if not is_success:
            if not self.verbose:
                print(self.get_combined_log(logfile_path))
            print("Logfiles: ")
            print("  - {}".format(logfile_path))
            for runner in self.active_runners:
                print("  - {}".format(runner.get_log_filename()))

        if self.upload:
            ulog_file = self.parse_for_ulog_file(
                self.get_combined_log(logfile_path))
            self.upload_log(ulog_file, test['model'], case, is_success)

        return is_success

    def start_runners(self,
                      log_dir: str,
                      test: Dict[str, Any],
                      case: str) -> None:
        self.active_runners = []

        if self.config['mode'] == 'sitl':
            if self.config['simulator'] == 'gazebo':
                # Use RegEx to extract worldname.world from case name
                match = re.search(r'\((.*?\.world)\)', case)
                if match:
                    world_name = match.group(1)
                else:
                    world_name = 'empty.world'

                gzserver_runner = ph.GzserverRunner(
                    os.getcwd(),
                    log_dir,
                    test['vehicle'],
                    case,
                    self.get_max_speed_factor(test),
                    self.verbose,
                    self.build_dir,
                    world_name)
                self.active_runners.append(gzserver_runner)

                gzmodelspawn_runner = ph.GzmodelspawnRunner(
                    os.getcwd(),
                    log_dir,
                    test['vehicle'],
                    case,
                    self.verbose,
                    self.build_dir)
                self.active_runners.append(gzmodelspawn_runner)

                if self.gui:
                    gzclient_runner = ph.GzclientRunner(
                        os.getcwd(),
                        log_dir,
                        test['model'],
                        case,
                        self.verbose)
                    self.active_runners.append(gzclient_runner)

                # We must start the PX4 instance at the end, as starting
                # it in the beginning, then connecting Gazebo server freaks
                # out the PX4 (it needs to have data coming in when started),
                # and can lead to EKF to freak out, or the instance itself
                # to die unexpectedly.
                px4_runner = ph.Px4Runner(
                    os.getcwd(),
                    log_dir,
                    test['model'],
                    case,
                    self.get_max_speed_factor(test),
                    self.debugger,
                    self.verbose,
                    self.build_dir)
                for env_key in test.get('env', []):
                    px4_runner.env[env_key] = str(test['env'][env_key])
                self.active_runners.append(px4_runner)

        mavsdk_tests_runner = ph.TestRunner(
            os.getcwd(),
            log_dir,
            test['model'],
            case,
            self.config['mavlink_connection'],
            self.speed_factor,
            self.verbose,
            self.build_dir)
        self.active_runners.append(mavsdk_tests_runner)

        abort = False
        for runner in self.active_runners:
            runner.set_log_filename(
                self.determine_logfile_path(log_dir, runner.name))

            if not self.try_to_run_several_times(runner):
                abort = True
                break

        if abort:
            print("Could not start runner: {}".format(runner.name))
            self.collect_runner_output()
            self.stop_combined_log()
            self.stop_runners()
            sys.exit(1)

    def try_to_run_several_times(self, runner: ph.Runner) -> bool:
        for _ in range(3):
            runner.start()

            if runner.has_started_ok():
                return True

            else:
                runner.stop()
                time.sleep(1)

        return False

    def stop_runners(self) -> None:
        for runner in self.active_runners:
            runner.stop()

    def collect_runner_output(self) -> None:
        for runner in self.active_runners:
            while True:
                line = runner.get_output_line()
                if not line:
                    break

                self.add_to_combined_log(line)
                if self.verbose:
                    print(line, end="")

    def parse_for_ulog_file(self, log: str) -> Optional[str]:

        match = "[logger] Opened full log file: ./"
        for line in log.splitlines():
            found = line.find(match)
            if found != -1:
                return os.path.join(os.getcwd(), self.build_dir,
                                    "tmp_mavsdk_tests/rootfs",
                                    line[found+len(match):])
        return None

    def upload_log(self, ulog_path: Optional[str],
                   model: str, case: str, success: bool) -> None:
        if not ulog_path:
            print("    Could not find ulog log file to upload")
            return

        if not os.getenv('GITHUB_RUN_ID'):
            print("    Upload only implemented for GitHub Actions CI")
            return

        print("    Uploading logfile '{}' ...".format(ulog_path))

        server = "https://logs.px4.io"

        result_str = "passing" if success else "failing"

        payload = {
            "type": "flightreport",
            "description": "SITL integration test - {}: '{}' -> {}"
            .format(model, case, result_str),
            "feedback":
                "{}/{}/actions/runs/{}"
                .format(os.getenv("GITHUB_SERVER_URL"),
                        os.getenv("GITHUB_REPOSITORY"),
                        os.getenv("GITHUB_RUN_ID")),
            "email": "",
            "source": "CI",
            "videoUrl": "",
            "rating": "notset",
            "windSpeed": -1,
            "public": "true"
        }

        with open(ulog_path, 'rb') as f:
            r = requests.post(server + "/upload",
                              data=payload,
                              files={'filearg': f},
                              allow_redirects=False)
            if r.status_code == 302:  # redirect
                if 'Location' in r.headers:
                    plot_url = r.headers['Location']
                    if len(plot_url) > 0 and plot_url[0] == '/':
                        plot_url = server + plot_url
                    print("    Uploaded to: " + plot_url)
            else:
                print("    Upload failed with status_code: {}"
                      .format(r.status_code))

    def start_combined_log(self, filename: str) -> None:
        self.log_fd = open(filename, 'w')

    def stop_combined_log(self) -> None:
        if self.log_fd:
            self.log_fd.close()

    def add_to_combined_log(self, output: str) -> None:
        self.log_fd.write(output)

    def get_combined_log(self, filename: str) -> str:
        with open(filename, 'r') as f:
            return f.read()

    @staticmethod
    def determine_logfile_path(log_dir: str, desc: str) -> str:
        return os.path.join(log_dir, "log-{}.log".format(desc))

    def show_detailed_results(self) -> None:
        print()
        print(colorize("Results:", color.BOLD))

        for test in self.tests:
            if not test['selected']:
                print(colorize(colorize(
                    "  - {} (not selected)".
                    format(test['model']), color.BOLD), color.GRAY))
                continue
            else:
                print(colorize("  - {}:".format(test['model']), color.BOLD))

            for name, case in test['cases'].items():
                if not case['selected']:
                    continue
                print("     - '{}': ".format(name), end="")

                n_succeeded = [result['success']
                               for result in case['results']].count(True)
                n_failed = [result['success']
                            for result in case['results']].count(False)
                n_cancelled = self.iterations - len(case['results'])

                notes = [
                    self.note_if_any(
                        colorize("{}succeeded", color.GREEN),
                        n_succeeded, self.iterations),
                    self.note_if_any(
                        colorize("{}failed", color.RED),
                        n_failed, self.iterations),
                    self.note_if_any(
                        colorize("{}cancelled", color.GRAY),
                        n_cancelled, self.iterations)]
                notes_without_none = list(filter(None, notes))
                print(", ".join(notes_without_none))

    def was_overall_pass(self) -> bool:
        for test in self.tests:
            if not test['selected']:
                continue

            for case in test['cases'].values():
                if not case['selected']:
                    continue
                for result in case['results']:
                    if result['success'] is not True:
                        return False
        return True

    def show_overall_result(self) -> None:
        print(colorize(
            "Overall result: {}".format(
             colorize("PASS", color.GREEN)
             if self.was_overall_pass()
             else colorize("FAIL", color.RED)), color.BOLD))

    @staticmethod
    def note_if_any(text_to_format: str, n: int, total: int) -> Optional[str]:
        assert not n < 0
        if n == 0:
            return None
        elif n == 1 and total == 1:
            return text_to_format.format("")
        else:
            return text_to_format.format(str(n) + " ")

    def sigint_handler(self, sig: int, frame: Optional[FrameType]) \
            -> NoReturn:
        print("Received SIGINT")
        print("Stopping all processes ...")
        self.stop_runners()
        self.stop_combined_log()
        print("Stopping all processes done.")
        self.show_detailed_results()
        sys.exit(-sig)


if __name__ == '__main__':
    main()