Rebase with master

mbed_host_tests/__init__.py

@@ -27,10 +27,8 @@
 
 
 import os
-import sys
 import imp
 import inspect
-from time import time
 from os import listdir
 from os.path import isfile, join, abspath
 from optparse import OptionParser

mbed_host_tests/host_tests_conn_proxy/conn_proxy.py

@@ -22,6 +22,7 @@
 from Queue import Empty as QueueEmpty   # Queue here refers to the module, not a class
 from serial import Serial, SerialException
 from mbed_host_tests import host_tests_plugins
+from mbed_host_tests.host_tests_plugins.host_test_plugins import HostTestPluginBase
 from conn_proxy_logger import HtrunLogger
 
 
@@ -34,10 +35,25 @@ def __init__(self, port, baudrate, prn_lock, config):
         self.config = config
         self.logger = HtrunLogger(prn_lock, 'SERI')
         self.LAST_ERROR = None
+        self.target_id = self.config.get('target_id', None)
 
+        # Values used to call serial port listener...
         self.logger.prn_inf("serial(port=%s, baudrate=%d, timeout=%s)"% (self.port, self.baudrate, self.timeout))
+
+        # Check if serial port for given target_id changed
+        # If it does we will use new port to open connections and make sure reset plugin
+        # later can reuse opened already serial port
+        #
+        # Note: This listener opens serial port and keeps connection so reset plugin uses
+        # serial port object not serial port name!
+        _, serial_port = HostTestPluginBase().check_serial_port_ready(self.port, target_id=self.target_id)
+        if serial_port != self.port:
+            # Serial port changed for given targetID
+            self.logger.prn_inf("serial port changed from '%s to '%s')"% (self.port, serial_port))
+            self.port = serial_port
+
         try:
-            self.serial = Serial(port, baudrate=baudrate, timeout=self.timeout)
+            self.serial = Serial(self.port, baudrate=self.baudrate, timeout=self.timeout)
         except SerialException as e:
             self.serial = None
             self.LAST_ERROR = "connection lost, serial.Serial(%s. %d, %d): %s"% (self.port,
@@ -59,15 +75,16 @@ def reset_dev_via_serial(self, delay=1):
         result = host_tests_plugins.call_plugin('ResetMethod',
             reset_type,
             serial=self.serial,
-            disk=disk)
+            disk=disk,
+            target_id=self.target_id)
         # Post-reset sleep
         self.logger.prn_inf("wait for it...")
         sleep(delay)
         return result
 
     def read(self, count):
         """! Read data from serial port RX buffer """
-        c = ''
+        c = str()
         try:
             if self.serial:
                 c = self.serial.read(count)
@@ -115,7 +132,7 @@ class KiViBufferWalker():
     """! Simple auxiliary class used to walk through a buffer and search for KV tokens """
     def __init__(self):
         self.KIVI_REGEX = r"\{\{([\w\d_-]+);([^\}]+)\}\}"
-        self.buff = ''
+        self.buff = str()
         self.buff_idx = 0
         self.re_kv = re.compile(self.KIVI_REGEX)
 
@@ -157,7 +174,7 @@ def conn_process(event_queue, dut_event_queue, prn_lock, config):
     sync_uuid_discovered = False
 
     # Some RXD data buffering so we can show more text per log line
-    print_data = ''
+    print_data = str()
 
     # Handshake, we will send {{sync;UUID}} preamble and wait for mirrored reply
     logger.prn_inf("sending preamble '%s'..."% sync_uuid)
@@ -197,7 +214,7 @@ def conn_process(event_queue, dut_event_queue, prn_lock, config):
                         if line:
                             logger.prn_rxd(line)
                             event_queue.put(('__rxd_line', line, time()))
-                    print_data = ''
+                    print_data = str()
                 else:
                     for line in print_data_lines[:-1]:
                         if line:

mbed_host_tests/host_tests_plugins/host_test_plugins.py

@@ -20,6 +20,7 @@
 import os
 import sys
 import platform
+import mbed_lstools
 
 from os import access, F_OK
 from sys import stdout
@@ -107,16 +108,43 @@ def print_plugin_char(self, char):
         stdout.flush()
         return True
 
-    def check_mount_point_ready(self, destination_disk, init_delay=0.2, loop_delay=0.25):
+    def check_mount_point_ready(self, destination_disk, init_delay=0.2, loop_delay=0.25, target_id=None):
         """! Waits until destination_disk is ready and can be accessed by e.g. copy commands
-
         @return True if mount point was ready in given time, False otherwise
-
         @param destination_disk Mount point (disk) which will be checked for readiness
         @param init_delay - Initial delay time before first access check
         @param loop_delay - polling delay for access check
         """
+
+        if target_id:
+            # Wait for mount point to appear with mbed-ls
+            # and if it does check if mount point for target_id changed
+            # If mount point changed, use new mount point and check if its ready (os.access)
+            new_destination_disk = destination_disk
+
+            # Sometimes OSes take a long time to mount devices (up to one minute).
+            # Current pooling time: 120x 500ms = 1 minute
+            self.print_plugin_info("Waiting for '%s' mount point (current is '%s')..."% (target_id, destination_disk))
+            for i in range(120):
+                # mbed_lstools.create() should be done inside the loop.
+                # Otherwise it will loop on same data.
+                mbeds = mbed_lstools.create()
+                mbeds_by_tid = mbeds.list_mbeds_by_targetid()   # key: target_id, value mbedls_dict()
+                if target_id in mbeds_by_tid:
+                    if 'mount_point' in mbeds_by_tid[target_id]:
+                        if mbeds_by_tid[target_id]['mount_point']:
+                            # Only assign if mount point is known (not None)
+                            new_destination_disk = mbeds_by_tid[target_id]['mount_point']
+                            break
+                sleep(0.5)
+
+            if new_destination_disk != destination_disk:
+                # Mount point changed, update to new mount point from mbed-ls
+                self.print_plugin_info("Mount point for '%s' changed from '%s' to '%s'..."% (target_id, destination_disk, new_destination_disk))
+                destination_disk = new_destination_disk
+
         result = False
+        # Check if mount point we've promoted to be valid one (by optional target_id check above)
         # Let's wait for 30 * loop_delay + init_delay max
         if not access(destination_disk, F_OK):
             self.print_plugin_info("Waiting for mount point '%s' to be ready..."% destination_disk, NL=False)
@@ -127,16 +155,45 @@ def check_mount_point_ready(self, destination_disk, init_delay=0.2, loop_delay=0
                     break
                 sleep(loop_delay)
                 self.print_plugin_char('.')
-        return result
+        return (result, destination_disk)
+
+    def check_serial_port_ready(self, serial_port, target_id=None):
+        """! Function checks (using mbed-ls) and updates serial port name information for DUT with specified target_id.
+        If no target_id is specified function returns old serial port name.
+        @param serial_port Current serial port name
+        @param target_id Target ID of a device under test which serial port will be checked and updated if needed
+        @return Tuple with result (always True) and serial port read from mbed-ls
+        """
+        result = True
+
+        if target_id:
+            # If serial port changed (check using mbed-ls), use new serial port
+            new_serial_port = serial_port
+            for i in range(25): # 25x 200ms = 5sec
+                # mbed_lstools.create() should be done inside the loop. Otherwise it will loop on same data.
+                mbeds = mbed_lstools.create()
+                mbeds_by_tid = mbeds.list_mbeds_by_targetid()   # key: target_id, value mbedls_dict()
+                if target_id in mbeds_by_tid:
+                    if 'serial_port' in mbeds_by_tid[target_id]:
+                        if mbeds_by_tid[target_id]['serial_port']:
+                            # Only assign if serial port is known (not None)
+                            new_serial_port = mbeds_by_tid[target_id]['serial_port']
+                            break
+                sleep(0.2)
+
+            if new_serial_port != serial_port:
+                # Serial port changed, update to new serial port from mbed-ls
+                self.print_plugin_info("Serial port for tid='%s' changed from '%s' to '%s'..."% (target_id, serial_port, new_serial_port))
+                serial_port = new_serial_port
+
+        return (result, serial_port)
 
     def check_parameters(self, capability, *args, **kwargs):
         """! This function should be ran each time we call execute() to check if none of the required parameters is missing
-
-        @return Returns True if all parameters are passed to plugin, else return False
-
         @param capability Capability name
         @param args Additional parameters
         @param kwargs Additional parameters
+        @return Returns True if all parameters are passed to plugin, else return False
         """
         missing_parameters = []
         for parameter in self.required_parameters:
@@ -149,12 +206,9 @@ def check_parameters(self, capability, *args, **kwargs):
 
     def run_command(self, cmd, shell=True):
         """! Runs command from command line.
-
         @param cmd Command to execute
         @param shell True if shell command should be executed (eg. ls, ps)
-
         @details Function prints 'cmd' return code if execution failed
-
         @return True if command successfully executed
         """
         result = True
@@ -171,7 +225,6 @@ def run_command(self, cmd, shell=True):
 
     def mbed_os_info(self):
         """! Returns information about host OS
-
         @return Returns tuple with information about OS and host platform
         """
         result = (os.name,

mbed_host_tests/host_tests_plugins/module_copy_mbed.py

@@ -72,6 +72,9 @@ def execute(self, capability, *args, **kwargs):
             self.print_plugin_error("Error: destination disk not specified")
             return False
 
+        # This optional parameter can be used if TargetID is provided (-t switch)
+        target_id = kwargs.get('target_id', None)
+
         result = False
         if self.check_parameters(capability, *args, **kwargs):
             # Capability 'default' is a dummy capability
@@ -80,7 +83,9 @@ def execute(self, capability, *args, **kwargs):
                     image_path = os.path.normpath(kwargs['image_path'])
                     destination_disk = os.path.normpath(kwargs['destination_disk'])
                     # Wait for mount point to be ready
-                    self.check_mount_point_ready(destination_disk)  # Blocking
+                    # if mount point changed according to target_id use new mount point
+                    # available in result (_, destination_disk) of check_mount_point_ready
+                    mount_res, destination_disk = self.check_mount_point_ready(destination_disk, target_id=self.target_id)  # Blocking
                     result = self.generic_mbed_copy(image_path, destination_disk)
         return result
 

mbed_host_tests/host_tests_plugins/module_copy_shell.py

@@ -53,13 +53,18 @@ def execute(self, capability, *args, **kwargs):
             self.print_plugin_error("Error: destination disk not specified")
             return False
 
+        # This optional parameter can be used if TargetID is provided (-t switch)
+        target_id = kwargs.get('target_id', None)
+
         result = False
         if self.check_parameters(capability, *args, **kwargs):
             if kwargs['image_path'] and kwargs['destination_disk']:
                 image_path = os.path.normpath(kwargs['image_path'])
                 destination_disk = os.path.normpath(kwargs['destination_disk'])
                 # Wait for mount point to be ready
-                self.check_mount_point_ready(destination_disk)  # Blocking
+                # if mount point changed according to target_id use new mount point
+                # available in result (_, destination_disk) of check_mount_point_ready
+                mount_res, destination_disk = self.check_mount_point_ready(destination_disk, target_id=target_id)  # Blocking
                 # Prepare correct command line parameter values
                 image_base_name = basename(image_path)
                 destination_path = join(destination_disk, image_base_name)

mbed_host_tests/host_tests_runner/host_test.py

@@ -17,6 +17,7 @@
 Author: Przemyslaw Wirkus <[email protected]>
 """
 
+import pkg_resources  # part of setuptools
 from sys import stdout
 from mbed_host_tests.host_tests_runner.mbed_base import Mbed
 
@@ -121,6 +122,13 @@ def finish(self):
         """
         pass
 
+    def get_hello_string(self):
+        """ Hello string used as first print
+        """
+        pkg = 'mbed-host-tests'
+        version = pkg_resources.require(pkg)[0].version
+        return "host test executor ver. " + version
+
 
 class DefaultTestSelectorBase(Test):
     """! Test class with serial port initialization

mbed_host_tests/host_tests_runner/host_test_default.py

@@ -109,7 +109,8 @@ def start_conn_process():
                 "port" : self.mbed.port,
                 "baudrate" : self.mbed.serial_baud,
                 "program_cycle_s" : self.options.program_cycle_s,
-                "reset_type" : self.options.forced_reset_type
+                "reset_type" : self.options.forced_reset_type,
+                "target_id" : self.options.target_id
             }
             # DUT-host communication process
             args = (event_queue, dut_event_queue, self.prn_lock, config)
@@ -310,6 +311,9 @@ def execute(self):
         """
         result = self.RESULT_UNDEF
 
+        # hello sting with htrun version, for debug purposes
+        self.logger.prn_inf(self.get_hello_string())
+
         try:
             # Copy image to device
             if self.options.skip_flashing:

mbed_host_tests/host_tests_runner/mbed_base.py

@@ -110,7 +110,8 @@ def copy_image_raw(self, image_path=None, disk=None, copy_method=None, port=None
                                         copy_method,
                                         image_path=image_path,
                                         serial=port,
-                                        destination_disk=disk)
+                                        destination_disk=disk,
+                                        target_id=self.target_id)
         return result
 
     def hw_reset(self):

mbed_host_tests/host_tests_toolbox/host_functional.py

@@ -17,6 +17,8 @@
 Author: Przemyslaw Wirkus <[email protected]>
 """
 
+import sys
+import json
 from time import sleep
 from serial import Serial, SerialException
 from mbed_host_tests import host_tests_plugins

mbed_host_tests/mbedhtrun.py

@@ -17,6 +17,7 @@
 Author: Przemyslaw Wirkus <[email protected]>
 """
 
+from multiprocessing import freeze_support
 from mbed_host_tests import init_host_test_cli_params
 from mbed_host_tests.host_tests_runner.host_test_default import DefaultTestSelector
 
@@ -26,6 +27,7 @@ def main():
     @details 1. Create DefaultTestSelector object and pass command line parameters
              2. Call default test execution function run() to start test instrumentation
     """
+    freeze_support()
     result = -2
     test_selector = DefaultTestSelector(init_host_test_cli_params())
     try:

setup.py

@@ -36,7 +36,7 @@ def read(fname):
     return open(os.path.join(os.path.dirname(__file__), fname)).read()
 
 setup(name='mbed-host-tests',
-      version='0.2.4',
+      version='0.2.8',
       description=DESCRIPTION,
       long_description=read('README.md'),
       author=OWNER_NAMES,
@@ -54,4 +54,5 @@ def read(fname):
       },
       install_requires=["PySerial>=3.0",
                         "PrettyTable>=0.7.2",
-                        "requests"])
+                        "requests",
+                        "mbed-ls"])