tool/mbed/mbed-sdk/workspace_tools/host_tests/host_test.py

   1 """
   2 mbed SDK
   3 Copyright (c) 2011-2013 ARM Limited
   4
   5 Licensed under the Apache License, Version 2.0 (the "License");
   6 you may not use this file except in compliance with the License.
   7 You may obtain a copy of the License at
   8
   9     http://www.apache.org/licenses/LICENSE-2.0
  10
  11 Unless required by applicable law or agreed to in writing, software
  12 distributed under the License is distributed on an "AS IS" BASIS,
  13 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14 See the License for the specific language governing permissions and
  15 limitations under the License.
  16 """
  17
  18 # Check if 'serial' module is installed
  19 try:
  20     from serial import Serial
  21 except ImportError, e:
  22     print "Error: Can't import 'serial' module: %s"% e
  23     exit(-1)
  24
  25 import os
  26 import re
  27 import types
  28 from sys import stdout
  29 from time import sleep, time
  30 from optparse import OptionParser
  31
  32 import host_tests_plugins
  33
  34 # This is a little tricky. We need to add upper directory to path so
  35 # we can find packages we want from the same level as other files do
  36 import sys
  37 sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../..')))
  38
  39
  40 class Mbed:
  41     """ Base class for a host driven test
  42     """
  43     def __init__(self):
  44         parser = OptionParser()
  45
  46         parser.add_option("-m", "--micro",
  47                           dest="micro",
  48                           help="The target microcontroller",
  49                           metavar="MICRO")
  50
  51         parser.add_option("-p", "--port",
  52                           dest="port",
  53                           help="The serial port of the target mbed",
  54                           metavar="PORT")
  55
  56         parser.add_option("-d", "--disk",
  57                           dest="disk",
  58                           help="The target disk path",
  59                           metavar="DISK_PATH")
  60
  61         parser.add_option("-f", "--image-path",
  62                           dest="image_path",
  63                           help="Path with target's image",
  64                           metavar="IMAGE_PATH")
  65
  66         parser.add_option("-c", "--copy",
  67                           dest="copy_method",
  68                           help="Copy method selector",
  69                           metavar="COPY_METHOD")
  70
  71         parser.add_option("-C", "--program_cycle_s",
  72                           dest="program_cycle_s",
  73                           help="Program cycle sleep. Define how many seconds you want wait after copying bianry onto target",
  74                           type="float",
  75                           metavar="COPY_METHOD")
  76
  77         parser.add_option("-t", "--timeout",
  78                           dest="timeout",
  79                           help="Timeout",
  80                           metavar="TIMEOUT")
  81
  82         parser.add_option("-r", "--reset",
  83                           dest="forced_reset_type",
  84                           help="Forces different type of reset")
  85
  86         parser.add_option("-R", "--reset-timeout",
  87                           dest="forced_reset_timeout",
  88                           metavar="NUMBER",
  89                           type="int",
  90                           help="When forcing a reset using option -r you can set up after reset timeout in seconds")
  91
  92         (self.options, _) = parser.parse_args()
  93
  94         self.DEFAULT_RESET_TOUT = 0
  95         self.DEFAULT_TOUT = 10
  96
  97         if self.options.port is None:
  98             raise Exception("The serial port of the target mbed have to be provided as command line arguments")
  99
 100         # Options related to copy / reset mbed device
 101         self.port = self.options.port
 102         self.disk = self.options.disk
 103         self.image_path = self.options.image_path.strip('"')
 104         self.copy_method = self.options.copy_method
 105         self.program_cycle_s = float(self.options.program_cycle_s)
 106
 107         self.serial = None
 108         self.serial_baud = 9600
 109         self.serial_timeout = 1
 110
 111         self.timeout = self.DEFAULT_TOUT if self.options.timeout is None else self.options.timeout
 112         print 'MBED: Instrumentation: "%s" and disk: "%s"' % (self.port, self.disk)
 113
 114     def init_serial_params(self, serial_baud=9600, serial_timeout=1):
 115         """ Initialize port parameters.
 116             This parameters will be used by self.init_serial() function to open serial port
 117         """
 118         self.serial_baud = serial_baud
 119         self.serial_timeout = serial_timeout
 120
 121     def init_serial(self, serial_baud=None, serial_timeout=None):
 122         """ Initialize serial port.
 123             Function will return error is port can't be opened or initialized
 124         """
 125         # Overload serial port configuration from default to parameters' values if they are specified
 126         serial_baud = serial_baud if serial_baud is not None else self.serial_baud
 127         serial_timeout = serial_timeout if serial_timeout is not None else self.serial_timeout
 128
 129         # Clear serial port
 130         if self.serial:
 131             self.serial.close()
 132             self.serial = None
 133
 134         # We will pool for serial to be re-mounted if it was unmounted after device reset
 135         result = self.pool_for_serial_init(serial_baud, serial_timeout) # Blocking
 136
 137         # Port can be opened
 138         if result:
 139             self.flush()
 140         return result
 141
 142     def pool_for_serial_init(self, serial_baud, serial_timeout, pooling_loops=40, init_delay=0.5, loop_delay=0.25):
 143         """ Functions pools for serial port readiness
 144         """
 145         result = True
 146         last_error = None
 147         # This loop is used to check for serial port availability due to
 148         # some delays and remounting when devices are being flashed with new software.
 149         for i in range(pooling_loops):
 150             sleep(loop_delay if i else init_delay)
 151             try:
 152                 self.serial = Serial(self.port, baudrate=serial_baud, timeout=serial_timeout)
 153             except Exception as e:
 154                 result = False
 155                 last_error = "MBED: %s"% str(e)
 156                 stdout.write('.')
 157                 stdout.flush()
 158             else:
 159                 print "...port ready!"
 160                 result = True
 161                 break
 162         if not result and last_error:
 163             print last_error
 164         return result
 165
 166     def set_serial_timeout(self, timeout):
 167         """ Wraps self.mbed.serial object timeout property
 168         """
 169         result = None
 170         if self.serial:
 171             self.serial.timeout = timeout
 172             result = True
 173         return result
 174
 175     def serial_read(self, count=1):
 176         """ Wraps self.mbed.serial object read method
 177         """
 178         result = None
 179         if self.serial:
 180             try:
 181                 result = self.serial.read(count)
 182             except:
 183                 result = None
 184         return result
 185
 186     def serial_readline(self, timeout=5):
 187         """ Wraps self.mbed.serial object read method to read one line from serial port
 188         """
 189         result = ''
 190         start = time()
 191         while (time() - start) < timeout:
 192             if self.serial:
 193                 try:
 194                     c = self.serial.read(1)
 195                     result += c
 196                 except Exception as e:
 197                     print "MBED: %s"% str(e)
 198                     result = None
 199                     break
 200                 if c == '\n':
 201                     break
 202         return result
 203
 204     def serial_write(self, write_buffer):
 205         """ Wraps self.mbed.serial object write method
 206         """
 207         result = None
 208         if self.serial:
 209             try:
 210                 result = self.serial.write(write_buffer)
 211             except:
 212                result = None
 213         return result
 214
 215     def reset_timeout(self, timeout):
 216         """ Timeout executed just after reset command is issued
 217         """
 218         for n in range(0, timeout):
 219             sleep(1)
 220
 221     def reset(self):
 222         """ Calls proper reset plugin to do the job.
 223             Please refer to host_test_plugins functionality
 224         """
 225         # Flush serials to get only input after reset
 226         self.flush()
 227         if self.options.forced_reset_type:
 228             result = host_tests_plugins.call_plugin('ResetMethod', self.options.forced_reset_type, disk=self.disk)
 229         else:
 230             result = host_tests_plugins.call_plugin('ResetMethod', 'default', serial=self.serial)
 231         # Give time to wait for the image loading
 232         reset_tout_s = self.options.forced_reset_timeout if self.options.forced_reset_timeout is not None else self.DEFAULT_RESET_TOUT
 233         self.reset_timeout(reset_tout_s)
 234         return result
 235
 236     def copy_image(self, image_path=None, disk=None, copy_method=None):
 237         """ Closure for copy_image_raw() method.
 238             Method which is actually copying image to mbed
 239         """
 240         # Set closure environment
 241         image_path = image_path if image_path is not None else self.image_path
 242         disk = disk if disk is not None else self.disk
 243         copy_method = copy_method if copy_method is not None else self.copy_method
 244         # Call proper copy method
 245         result = self.copy_image_raw(image_path, disk, copy_method)
 246         sleep(self.program_cycle_s)
 247         return result
 248
 249     def copy_image_raw(self, image_path=None, disk=None, copy_method=None):
 250         """ Copy file depending on method you want to use. Handles exception
 251             and return code from shell copy commands.
 252         """
 253         if copy_method is not None:
 254             # image_path - Where is binary with target's firmware
 255             result = host_tests_plugins.call_plugin('CopyMethod', copy_method, image_path=image_path, destination_disk=disk)
 256         else:
 257             copy_method = 'default'
 258             result = host_tests_plugins.call_plugin('CopyMethod', copy_method, image_path=image_path, destination_disk=disk)
 259         return result;
 260
 261     def flush(self):
 262         """ Flush serial ports
 263         """
 264         result = False
 265         if self.serial:
 266             self.serial.flushInput()
 267             self.serial.flushOutput()
 268             result = True
 269         return result
 270
 271
 272 class HostTestResults:
 273     """ Test results set by host tests
 274     """
 275     def __init__(self):
 276         self.RESULT_SUCCESS = 'success'
 277         self.RESULT_FAILURE = 'failure'
 278         self.RESULT_ERROR = 'error'
 279         self.RESULT_IO_SERIAL = 'ioerr_serial'
 280         self.RESULT_NO_IMAGE = 'no_image'
 281         self.RESULT_IOERR_COPY = "ioerr_copy"
 282         self.RESULT_PASSIVE = "passive"
 283         self.RESULT_NOT_DETECTED = "not_detected"
 284         self.RESULT_MBED_ASSERT = "mbed_assert"
 285
 286
 287 import workspace_tools.host_tests as host_tests
 288
 289
 290 class Test(HostTestResults):
 291     """ Base class for host test's test runner
 292     """
 293     # Select default host_test supervision (replaced after autodetection)
 294     test_supervisor = host_tests.get_host_test("default")
 295
 296     def __init__(self):
 297         self.mbed = Mbed()
 298
 299     def detect_test_config(self, verbose=False):
 300         """ Detects test case configuration
 301         """
 302         result = {}
 303         while True:
 304             line = self.mbed.serial_readline()
 305             if "{start}" in line:
 306                 self.notify("HOST: Start test...")
 307                 break
 308             else:
 309                 # Detect if this is property from TEST_ENV print
 310                 m = re.search('{([\w_]+);([\w\d\+ ]+)}}', line[:-1])
 311                 if m and len(m.groups()) == 2:
 312                     # This is most likely auto-detection property
 313                     result[m.group(1)] = m.group(2)
 314                     if verbose:
 315                         self.notify("HOST: Property '%s' = '%s'"% (m.group(1), m.group(2)))
 316                 else:
 317                     # We can check if this is TArget Id in mbed specific format
 318                     m2 = re.search('^([\$]+)([a-fA-F0-9]+)', line[:-1])
 319                     if m2 and len(m2.groups()) == 2:
 320                         if verbose:
 321                             target_id = m2.group(1) + m2.group(2)
 322                             self.notify("HOST: TargetID '%s'"% target_id)
 323                             self.notify(line[len(target_id):-1])
 324                     else:
 325                         self.notify("HOST: Unknown property: %s"% line.strip())
 326         return result
 327
 328     def run(self):
 329         """ Test runner for host test. This function will start executing
 330             test and forward test result via serial port to test suite
 331         """
 332         # Copy image to device
 333         self.notify("HOST: Copy image onto target...")
 334         result = self.mbed.copy_image()
 335         if not result:
 336             self.print_result(self.RESULT_IOERR_COPY)
 337
 338         # Initialize and open target's serial port (console)
 339         self.notify("HOST: Initialize serial port...")
 340         result = self.mbed.init_serial()
 341         if not result:
 342             self.print_result(self.RESULT_IO_SERIAL)
 343
 344         # Reset device
 345         self.notify("HOST: Reset target...")
 346         result = self.mbed.reset()
 347         if not result:
 348             self.print_result(self.RESULT_IO_SERIAL)
 349
 350         # Run test
 351         try:
 352             CONFIG = self.detect_test_config(verbose=True) # print CONFIG
 353
 354             if "host_test_name" in CONFIG:
 355                 if host_tests.is_host_test(CONFIG["host_test_name"]):
 356                     self.test_supervisor = host_tests.get_host_test(CONFIG["host_test_name"])
 357             result = self.test_supervisor.test(self)    #result = self.test()
 358
 359             if result is not None:
 360                 self.print_result(result)
 361             else:
 362                 self.notify("HOST: Passive mode...")
 363         except Exception, e:
 364             print str(e)
 365             self.print_result(self.RESULT_ERROR)
 366
 367     def setup(self):
 368         """ Setup and check if configuration for test is
 369             correct. E.g. if serial port can be opened.
 370         """
 371         result = True
 372         if not self.mbed.serial:
 373             result = False
 374             self.print_result(self.RESULT_IO_SERIAL)
 375         return result
 376
 377     def notify(self, message):
 378         """ On screen notification function
 379         """
 380         print message
 381         stdout.flush()
 382
 383     def print_result(self, result):
 384         """ Test result unified printing function
 385         """
 386         self.notify("\r\n{{%s}}\r\n{{end}}" % result)
 387
 388
 389 class DefaultTestSelector(Test):
 390     """ Test class with serial port initialization
 391     """
 392     def __init__(self):
 393         HostTestResults.__init__(self)
 394         Test.__init__(self)
 395
 396 if __name__ == '__main__':
 397     DefaultTestSelector().run()