<commit_before>"""
mbed SDK
Copyright (c) 2011-2013 ARM Limited

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.

Author: <NAME> <<EMAIL>>

Usage:
1.  Update your private_settings.py with all MUTs you can possibly connect.
    Make sure mcu / port / serial names are concretely inputed.
2.  Update test_spec dictionary in __main__ section.

    Example 1:
    In below example only LPC11U24 will be tested
    and test will be prepared using only uARM toolchain. Note that other
    targets are just commented.
    Uncomment or add your own targets at will.

    test_spec = {
        "targets": {
            # "KL25Z": ["ARM", "GCC_ARM"],
            # "LPC1768": ["ARM", "GCC_ARM", "GCC_CR", "GCC_CS", "IAR"],
            "LPC11U24": ["uARM"]
            # "NRF51822": ["ARM"]
            # "NUCLEO_F103RB": ["ARM"]
        }
    }

"""

import sys
import json
import optparse
import pprint
from prettytable import PrettyTable
from serial import Serial

from os.path import join, abspath, dirname, exists
from shutil import copy
from subprocess import call
from time import sleep, time

ROOT = abspath(join(dirname(__file__), ".."))
sys.path.insert(0, ROOT)
# Imports related to mbed build pi
from workspace_tools.build_api import build_project, build_mbed_libs
from workspace_tools.paths import BUILD_DIR
from workspace_tools.targets import TARGET_MAP
from workspace_tools.tests import TEST_MAP

# Be sure that the tools directory is in the search path
ROOT = abspath(join(dirname(__file__), ".."))
sys.path.insert(0, ROOT)

# Imports related to mbed build pi
from workspace_tools.utils import delete_dir_files
from workspace_tools.settings import MUTs


class SingleTestRunner():
    """ Object wrapper for single test run which may involve multiple MUTs."""
    def __init__(self):
        pass

    def reset(self, mcu_name, serial,
              verbose=False, sleep_before_reset=0, sleep_after_reset=0):
        """
        Functions resets target using various methods (e.g. serial break)
        depending on target type.
        """
        if sleep_before_reset > 0:
            sleep(sleep_before_reset)
        if verbose:
            verbose_msg = "Reset::cmd(sendBreak)"
        # Reset type decision
        if mcu_name.startswith('NRF51822'): # Nordic
            call(["nrfjprog", "-r"])
            verbose_msg = "Reset::cmd(nrfjprog)"
        elif mcu_name.startswith('NUCLEO'): # ST NUCLEO
            call(["ST-LINK_CLI.exe", "-Rst"])
            verbose_msg = "Reset::cmd(ST-LINK_CLI.exe)"
        else:
            serial.sendBreak()

        if sleep_before_reset > 0:
            sleep(sleep_after_reset)
        if verbose:
            print verbose_msg

    def flush_serial(self, serial):
        """ Flushing serial in/out. """
        serial.flushInput()
        serial.flushOutput()

    def is_peripherals_available(self, target, peripherals=None):
        """ Checks if specified target should run specific peripheral test case."""
        if peripherals is not None:
            peripherals = set(peripherals)

        for id, mut in MUTs.iteritems():
            # Target check
            if mut["mcu"] != target:
                continue
            # Peripherals check
            if peripherals is not None:
                if 'peripherals' not in mut:
                    continue
                if not peripherals.issubset(set(mut['peripherals'])):
                    continue
            return True
        return False

    def run_host_test(self, name, target_name, disk, port,
                      duration, extra_serial, verbose=False):
        """
        Functions resets target and grabs by timeouted pooling test log
        via serial port.
        Function assumes target is already flashed with proper 'test' binary.
        """
        output = ""
        # Prepare serial for receiving data from target
        baud = 9600
        serial = Serial(port, timeout=1)
        serial.setBaudrate(baud)
        self.flush_serial(serial)
        # Resetting target and pooling
        self.reset(target_name, serial, verbose=verbose)
        start = time()
        try:
            while (time() - start) < duration:
                test_output = serial.read(512)
                output += test_output
                self.flush_serial(serial)
                if '{end}' in output:
                    break

        except KeyboardInterrupt, _:
            print "CTRL+C break"
        self.flush_serial(serial)
        serial.close()

        # Handle verbose mode
        if verbose:
            print "Test::Output::Start"
            print output
            print "Test::Output::Finish"

        # Parse test 'output' data
        result = "UNDEF"
        for line in output.splitlines():
            if '{success}' in line: result = "OK"
            if '{failure}' in line: result = "FAIL"
            if '{error}' in line: result = "ERROR"
            if '{end}' in line: break
        return result

    def print_test_result(self, test_result, target_name, toolchain_name,
                          test_id, test_description, elapsed_time, duration):
        """ Use specific convention to pront test result and related data."""
        tokens = []
        tokens.append("TargetTest")
        tokens.append(target_name)
        tokens.append(toolchain_name)
        tokens.append(test_id)
        tokens.append(test_description)
        separator = "::"
        time_info = " in %d of %d sec" % (elapsed_time, duration)<commit_msg>Fixed minor pylint warnings and comments<commit_after>"""
mbed SDK
Copyright (c) 2011-2013 ARM Limited

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

    http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.

Author: <NAME> <<EMAIL>>

Usage:
1.  Update your private_settings.py with all MUTs you can possibly connect.
    Make sure mcu / port / serial names are concretely inputed.
2.  Update test_spec dictionary in __main__ section.

    Example 1:
    In below example only LPC11U24 will be tested
    and test will be prepared using only uARM toolchain. Note that other
    targets are just commented.
    Uncomment or add your own targets at will.

    test_spec = {
        "targets": {
            # "KL25Z": ["ARM", "GCC_ARM"],
            # "LPC1768": ["ARM", "GCC_ARM", "GCC_CR", "GCC_CS", "IAR"],
            "LPC11U24": ["uARM"]
            # "NRF51822": ["ARM"]
            # "NUCLEO_F103RB": ["ARM"]
        }
    }

"""

import sys
import json
import optparse
import pprint
from prettytable import PrettyTable
from serial import Serial

from os.path import join, abspath, dirname, exists
from shutil import copy
from subprocess import call
from time import sleep, time

ROOT = abspath(join(dirname(__file__), ".."))
sys.path.insert(0, ROOT)
# Imports related to mbed build pi
from workspace_tools.build_api import build_project, build_mbed_libs
from workspace_tools.paths import BUILD_DIR
from workspace_tools.targets import TARGET_MAP
from workspace_tools.tests import TEST_MAP

# Be sure that the tools directory is in the search path
ROOT = abspath(join(dirname(__file__), ".."))
sys.path.insert(0, ROOT)

# Imports related to mbed build pi
from workspace_tools.utils import delete_dir_files
from workspace_tools.settings import MUTs


class SingleTestRunner():
    """ Object wrapper for single test run which may involve multiple MUTs."""
    def __init__(self):
        pass

    def reset(self, mcu_name, serial,
              verbose=False, sleep_before_reset=0, sleep_after_reset=0):
        """
        Functions resets target using various methods (e.g. serial break)
        depending on target type.
        """
        if sleep_before_reset > 0:
            sleep(sleep_before_reset)
        if verbose:
            verbose_msg = "Reset::cmd(sendBreak)"
        # Reset type decision
        if mcu_name.startswith('NRF51822'): # Nordic
            call(["nrfjprog", "-r"])
            verbose_msg = "Reset::cmd(nrfjprog)"
        elif mcu_name.startswith('NUCLEO'): # ST NUCLEO
            call(["ST-LINK_CLI.exe", "-Rst"])
            verbose_msg = "Reset::cmd(ST-LINK_CLI.exe)"
        else:
            serial.sendBreak()

        if sleep_before_reset > 0:
            sleep(sleep_after_reset)
        if verbose:
            print verbose_msg

    def flush_serial(self, serial):
        """ Flushing serial in/out. """
        serial.flushInput()
        serial.flushOutput()

    def is_peripherals_available(self, target, peripherals=None):
        """ Checks if specified target should run specific peripheral test case."""
        if peripherals is not None:
            peripherals = set(peripherals)

        for id, mut in MUTs.iteritems():
            # Target check
            if mut["mcu"] != target:
                continue
            # Peripherals check
            if peripherals is not None:
                if 'peripherals' not in mut:
                    continue
                if not peripherals.issubset(set(mut['peripherals'])):
                    continue
            return True
        return False

    def run_host_test(self, name, target_name, disk, port,
                      duration, extra_serial, verbose=False):
        """
        Functions resets target and grabs by timeouted pooling test log
        via serial port.
        Function assumes target is already flashed with proper 'test' binary.
        """
        output = ""
        # Prepare serial for receiving data from target
        baud = 9600
        serial = Serial(port, timeout=1)
        serial.setBaudrate(baud)
        self.flush_serial(serial)
        # Resetting target and pooling
        self.reset(target_name, serial, verbose=verbose)
        start = time()
        try:
            while (time() - start) < duration:
                test_output = serial.read(512)
                output += test_output
                self.flush_serial(serial)
                if '{end}' in output:
                    break

        except KeyboardInterrupt, _:
            print "CTRL+C break"
        self.flush_serial(serial)
        serial.close()

        # Handle verbose mode
        if verbose:
            print "Test::Output::Start"
            print output
            print "Test::Output::Finish"

        # Parse test 'output' data
        result = "UNDEF"
        for line in output.splitlines():
            if '{success}' in line:
                result = "OK"
            if '{failure}' in line:
                result = "FAIL"
            if '{error}' in line:
                result = "ERROR"
            if '{end}' in line:
                break
        return result

    def print_test_result(self, test_result, target_name, toolchain_name,
                          test_id, test_description, elapsed_time, duration):
        """ Use specific convention to pront test result and related data."""
        tokens = []
        tokens.append("TargetTest")
        tokens.append(target_name)
        tokens.append(toolchain_name)
        tokens.append(test_id)
        tokens.append(test_description)
        separator = "::"
        time_info = " in %d of %d sec" % (elapsed_time, duration)
{'fraction_non_alphanumeric': 0.08724553385957623, 'fraction_numerical': 0.009306190278354798, 'mean_word_length': 2.272430668841762, 'pattern_counts': {'":': 14, '<': 11, '<?xml version=': 0, '>': 13, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, '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': '2204476', 'n_tokens_mistral': 3430, 'n_tokens_neox': 3239, 'n_words': 1106}