<commit_before>## Copyright (c) 2012 Aldebaran Robotics. All rights reserved.
## Use of this source code is governed by a BSD-style license that can be
## found in the COPYING file.

""" Re-implementation of CTest in Python.

Necessary to by-pass some small ctest shortcomings.
"""

import Queue
import datetime
import errno
import os
import re
import shlex
import signal
import sys
import threading
import time

import qisys
from qisys import ui
import qibuild
import qibuild.config

def _str_from_signal(code):
    """ Returns a nice string describing the signal

    """
    if code == signal.SIGSEGV:
        return "Segmentation fault"
    if code == signal.SIGABRT:
        return "Aborted"
    else:
        return "%i" % code


class QueueTimeout(Queue.Queue):
    def join(self):
        while self.unfinished_tasks:
            time.sleep(1)


_LOCK = threading.Lock()
_MULTIPLE_JOBS = False


class TestResult:
    """ Just a small class to store the results for a test

    """
    def __init__(self, test_name, test_number, test_count, **kwargs):
        self.test_name = test_name
        self.time = 0
        self.ok = False
        self.not_run = False
        self.out = ""
        self.message = ""
        self.verbose = kwargs.get('verbose', False)
        self.result_dir = kwargs.get('result_dir', None)

        count = "(%2i/%2i)" % (test_number + 1, test_count)
        self.line = [ui.green, " * ", ui.reset, ui.bold, count, ui.blue,
                     self.test_name.ljust(25)]
        _LOCK.acquire()
        ui.info(*self.line, end='')
        if _MULTIPLE_JOBS and sys.stdout.isatty():
            ui.info(ui.blue, '[start]')
            _LOCK.release()

    def print_result(self):
        if _MULTIPLE_JOBS and sys.stdout.isatty():
            _LOCK.acquire()
            ui.info(*self.line, end='')
        if self.ok:
            ui.info(ui.green, "[OK]")
        else:
            ui.info(ui.red, "[FAIL]", self.message)
            if qisys.command.SIGINT_EVENT.is_set():
                pass
            elif self.verbose:
                if self.out:
                    print self.out
                else:
                    print '\tno output\n'
        sys.stdout.flush()
        if self.result_dir is not None:
            xml_out = os.path.join(self.result_dir, self.test_name + ".xml")
            # Sometimes the XML already exists. For instance if it comes from
            # a GTest created by qi_create_gtest()
            if not os.path.exists(xml_out):
                write_xml(xml_out, self)
        _LOCK.release()


def parse_valgrind(valgrind_log, tst):
    """
    parse valgrind logs and extract interesting errors.
    """
    leak_fd_regex      = re.compile("==\d+== FILE DESCRIPTORS: (\d+)")
    invalid_read_regex = re.compile("==\d+== Invalid read of size (\d+)")
    with open(valgrind_log, "r") as f:
        lines = f.readlines()

    for l in lines:
        tst.out += l
        r = leak_fd_regex.search(l)
        if r:
            fdopen = int(r.group(1))
            # 4: in/out/err + valgrind_log
            if fdopen > 4:
                tst.ok = False
                tst.message += "Error file descriptors leaks:" + str(fdopen - 4) + "\n"
            continue
        r = invalid_read_regex.search(l)
        if r:
            tst.ok = False
            tst.message += "Invalid read " + r.group(1) + "\n"


class Test:
    def __init__(self, build_dir, test_name, cmd, properties, build_env,
                 test_number, test_count, **kwargs):
        self.build_dir, self.test_name, self.cmd = build_dir, test_name, cmd
        self.properties, self.build_env = properties, build_env
        self.test_number, self.test_count = test_number, test_count

        # keyword arguments
        self.verbose = kwargs.get('verbose', False)
        self.valgrind = kwargs.get('valgrind', False)
        self.nightmare = kwargs.get('nightmare', False)

    def run(self, output_queue):
        """ Run a test.

        """
        timeout = self.properties.get("TIMEOUT")
        result_dir = os.path.join(self.build_dir, "test-results")
        res = TestResult(self.test_name, self.test_number, self.test_count,
                         verbose=self.verbose, result_dir=result_dir)
        if timeout:
            timeout = int(timeout)
        # we will merge the build env coming from toc
        # config with the env coming from CMake config,
        # assuming that cmake is always right
        env = self.build_env.copy()
        cmake_env = self.properties.get("ENVIRONMENT")
        if cmake_env:
            cmake_env = cmake_env.split(";")
            for key_value in cmake_env:
                key, value = key_value.split("=")
                env[key] = value
        working_dir = self.properties.get("WORKING_DIRECTORY")
        if working_dir:
            cwd = working_dir
        else:
            cwd = self.build_dir
        ncmd = self.cmd
        if self.valgrind:
            env['VALGRIND'] = '1'
            timeout *= 10
            valgrind_log = os.path.join(self.build_dir, self.test_name + "valgrind_output.log")
            ncmd = [ "valgrind", "--track-fds=yes", "--log-file=%s" % valgrind_log ]
            ncmd.extend(self.cmd)
        process = qisys.command.Process(ncmd,
            name=self.test_name,
            cwd=cwd,
            env=env,
            verbose=self.verbose)
        if self.nightmare:
            ncmd.extend(["--gtest_shuffle", "--gtest_repeat=20"])
        start = datetime.datetime.now()
        process.run(timeout)<commit_msg>test --nightmare : Tests are run 20 times, adapt timeout<commit_after>## Copyright (c) 2012 Aldebaran Robotics. All rights reserved.
## Use of this source code is governed by a BSD-style license that can be
## found in the COPYING file.

""" Re-implementation of CTest in Python.

Necessary to by-pass some small ctest shortcomings.
"""

import Queue
import datetime
import errno
import os
import re
import shlex
import signal
import sys
import threading
import time

import qisys
from qisys import ui
import qibuild
import qibuild.config

def _str_from_signal(code):
    """ Returns a nice string describing the signal

    """
    if code == signal.SIGSEGV:
        return "Segmentation fault"
    if code == signal.SIGABRT:
        return "Aborted"
    else:
        return "%i" % code


class QueueTimeout(Queue.Queue):
    def join(self):
        while self.unfinished_tasks:
            time.sleep(1)


_LOCK = threading.Lock()
_MULTIPLE_JOBS = False


class TestResult:
    """ Just a small class to store the results for a test

    """
    def __init__(self, test_name, test_number, test_count, **kwargs):
        self.test_name = test_name
        self.time = 0
        self.ok = False
        self.not_run = False
        self.out = ""
        self.message = ""
        self.verbose = kwargs.get('verbose', False)
        self.result_dir = kwargs.get('result_dir', None)

        count = "(%2i/%2i)" % (test_number + 1, test_count)
        self.line = [ui.green, " * ", ui.reset, ui.bold, count, ui.blue,
                     self.test_name.ljust(25)]
        _LOCK.acquire()
        ui.info(*self.line, end='')
        if _MULTIPLE_JOBS and sys.stdout.isatty():
            ui.info(ui.blue, '[start]')
            _LOCK.release()

    def print_result(self):
        if _MULTIPLE_JOBS and sys.stdout.isatty():
            _LOCK.acquire()
            ui.info(*self.line, end='')
        if self.ok:
            ui.info(ui.green, "[OK]")
        else:
            ui.info(ui.red, "[FAIL]", self.message)
            if qisys.command.SIGINT_EVENT.is_set():
                pass
            elif self.verbose:
                if self.out:
                    print self.out
                else:
                    print '\tno output\n'
        sys.stdout.flush()
        if self.result_dir is not None:
            xml_out = os.path.join(self.result_dir, self.test_name + ".xml")
            # Sometimes the XML already exists. For instance if it comes from
            # a GTest created by qi_create_gtest()
            if not os.path.exists(xml_out):
                write_xml(xml_out, self)
        _LOCK.release()


def parse_valgrind(valgrind_log, tst):
    """
    parse valgrind logs and extract interesting errors.
    """
    leak_fd_regex      = re.compile("==\d+== FILE DESCRIPTORS: (\d+)")
    invalid_read_regex = re.compile("==\d+== Invalid read of size (\d+)")
    with open(valgrind_log, "r") as f:
        lines = f.readlines()

    for l in lines:
        tst.out += l
        r = leak_fd_regex.search(l)
        if r:
            fdopen = int(r.group(1))
            # 4: in/out/err + valgrind_log
            if fdopen > 4:
                tst.ok = False
                tst.message += "Error file descriptors leaks:" + str(fdopen - 4) + "\n"
            continue
        r = invalid_read_regex.search(l)
        if r:
            tst.ok = False
            tst.message += "Invalid read " + r.group(1) + "\n"


class Test:
    def __init__(self, build_dir, test_name, cmd, properties, build_env,
                 test_number, test_count, **kwargs):
        self.build_dir, self.test_name, self.cmd = build_dir, test_name, cmd
        self.properties, self.build_env = properties, build_env
        self.test_number, self.test_count = test_number, test_count

        # keyword arguments
        self.verbose = kwargs.get('verbose', False)
        self.valgrind = kwargs.get('valgrind', False)
        self.nightmare = kwargs.get('nightmare', False)

    def run(self, output_queue):
        """ Run a test.

        """
        timeout = self.properties.get("TIMEOUT")
        result_dir = os.path.join(self.build_dir, "test-results")
        res = TestResult(self.test_name, self.test_number, self.test_count,
                         verbose=self.verbose, result_dir=result_dir)
        if timeout:
            timeout = int(timeout)
        # we will merge the build env coming from toc
        # config with the env coming from CMake config,
        # assuming that cmake is always right
        env = self.build_env.copy()
        cmake_env = self.properties.get("ENVIRONMENT")
        if cmake_env:
            cmake_env = cmake_env.split(";")
            for key_value in cmake_env:
                key, value = key_value.split("=")
                env[key] = value
        working_dir = self.properties.get("WORKING_DIRECTORY")
        if working_dir:
            cwd = working_dir
        else:
            cwd = self.build_dir
        ncmd = self.cmd
        if self.valgrind:
            env['VALGRIND'] = '1'
            timeout *= 10
            valgrind_log = os.path.join(self.build_dir, self.test_name + "valgrind_output.log")
            ncmd = [ "valgrind", "--track-fds=yes", "--log-file=%s" % valgrind_log ]
            ncmd.extend(self.cmd)
        process = qisys.command.Process(ncmd,
            name=self.test_name,
            cwd=cwd,
            env=env,
            verbose=self.verbose)
        if self.nightmare:
            ncmd.extend(["--gtest_shuffle", "--gtest_repeat=20"])
            timeout = timeout * 20
        start = datetime.datetime.now()
        process.run(timeout)
{'fraction_non_alphanumeric': 0.11206281227694503, 'fraction_numerical': 0.004104211277658815, 'mean_word_length': 1.9661286054511775, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 10}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7114330', 'n_tokens_mistral': 3536, 'n_tokens_neox': 3428, 'n_words': 872}