Added serial port wrappers for read/write instead of raw serial port access to access MUT serial port. This will allow to catch more issues with communication with MUT during tests. This will also 'hide' tracebacks from test suite execution.

Author: PrzemekWirkus

workspace_tools/host_tests/dev_null_auto.py

@@ -20,7 +20,11 @@
 
 class DevNullTest(DefaultTest):
     def run(self):
-        c = self.mbed.serial.read(512)
+        c = self.mbed.serial_read(512)
+        if c is None:
+            self.print_result("ioerr_serial")
+            return
+        # Data from serial received correctly
         print "Received %d bytes" % len(c)
         if "{failure}" not in c:
             self.print_result('success')

workspace_tools/host_tests/hello_auto.py

@@ -22,7 +22,10 @@ class HelloTest(DefaultTest):
     HELLO_WORLD = "Hello World\n"
 
     def run(self):
-        c = self.mbed.serial.read(len(self.HELLO_WORLD))
+        c = self.mbed.serial_read(len(self.HELLO_WORLD))
+        if c is None:
+            self.print_result("ioerr_serial")
+            return
         stdout.write(c)
         if c == self.HELLO_WORLD: # Hello World received
             self.print_result('success')

workspace_tools/host_tests/rtc_auto.py

@@ -28,7 +28,10 @@ def run(self):
         test_result = True
         c = self.mbed.serial.timeout = None
         for i in range(0, 5):
-            c = self.mbed.serial.read(38)   # 38 len("[1256729742] [2009-10-28 11:35:42 AM]\n"
+            c = self.mbed.serial_read(38)   # 38 len("[1256729742] [2009-10-28 11:35:42 AM]\n"
+            if c is None:
+                self.print_result("ioerr_serial")
+                return
             stdout.flush()
             m = self.re_detect_rtc_value.search(c)
             if m and len(m.groups()):

workspace_tools/host_tests/stdio_auto.py

@@ -37,7 +37,10 @@ def run(self):
             ip_msg_timeout = self.mbed.options.timeout
             start_serial_pool = time();
             while (time() - start_serial_pool) < ip_msg_timeout:
-                c = self.mbed.serial.read(512)
+                c = self.mbed.serial_read(512)
+                if c is None:
+                    self.print_result("ioerr_serial")
+                    return
                 stdout.write(c)
                 stdout.flush()
                 serial_stdio_msg += c

workspace_tools/host_tests/tcpecho_server_auto.py

@@ -34,7 +34,10 @@ def run(self):
         serial_ip_msg = ""
         start_serial_pool = time();
         while (time() - start_serial_pool) < ip_msg_timeout:
-            c = self.mbed.serial.read(512)
+            c = self.mbed.serial_read(512)
+            if c is None:
+                self.print_result("ioerr_serial")
+                return
             stdout.write(c)
             stdout.flush()
             serial_ip_msg += c
@@ -77,7 +80,10 @@ def run(self):
         # Receiving
         try:
             while True:
-                c = self.mbed.serial.read(512)
+                c = self.mbed.serial_read(512)
+                if c is None:
+                    self.print_result("ioerr_serial")
+                    break
                 stdout.write(c)
                 stdout.flush()
         except KeyboardInterrupt, _:

workspace_tools/host_tests/udp_link_layer_auto.py

@@ -79,7 +79,10 @@ def run(self):
         serial_ip_msg = ""
         start_serial_pool = time()
         while (time() - start_serial_pool) < ip_msg_timeout:
-            c = self.mbed.serial.read(512)
+            c = self.mbed.serial_read(512)
+            if c is None:
+                self.print_result("ioerr_serial")
+                return
             stdout.write(c)
             stdout.flush()
             serial_ip_msg += c
@@ -141,7 +144,10 @@ def run(self):
         print "Remaining mbed serial port data:"
         try:
             while True:
-                c = self.mbed.serial.read(512)
+                c = self.mbed.serial_read(512)
+                if c is None:
+                    self.print_result("ioerr_serial")
+                    break
                 stdout.write(c)
                 stdout.flush()
         except KeyboardInterrupt, _:

workspace_tools/host_tests/udpecho_server_auto.py

@@ -34,7 +34,10 @@ def run(self):
         serial_ip_msg = ""
         start_serial_pool = time();
         while (time() - start_serial_pool) < ip_msg_timeout:
-            c = self.mbed.serial.read(512)
+            c = self.mbed.serial_read(512)
+            if c is None:
+                self.print_result("ioerr_serial")
+                return
             stdout.write(c)
             stdout.flush()
             serial_ip_msg += c
@@ -76,7 +79,10 @@ def run(self):
         # Receiving
         try:
             while True:
-                c = self.mbed.serial.read(512)
+                c = self.mbed.serial_read(512)
+                if c is None:
+                    self.print_result("ioerr_serial")
+                    break
                 stdout.write(c)
                 stdout.flush()
         except KeyboardInterrupt, _:

workspace_tools/host_tests/wait_us_auto.py

@@ -24,16 +24,26 @@ def run(self):
         test_result = True
         # First character to start test (to know after reset when test starts)
         self.mbed.serial.timeout = None
-        c = self.mbed.serial.read(1)
+        c = self.mbed.serial_read(1)
+        if c is None:
+            self.print_result("ioerr_serial")
+            return
         if c == '$': # target will printout TargetID e.g.: $$$$1040e649d5c09a09a3f6bc568adef61375c6
             #Read additional 39 bytes of TargetID
-            self.mbed.serial.read(39)
-            c = self.mbed.serial.read(1) # Re-read first 'tick'
-
+            if not self.mbed.serial_read(39):
+                self.print_result("ioerr_serial")
+                return
+            c = self.mbed.serial_read(1) # Re-read first 'tick'
+            if c is None:
+                self.print_result("ioerr_serial")
+                return
         print "Test started"
         start_serial_pool = start = time();
         for i in range(0, 10):
-            c = self.mbed.serial.read(1)
+            c = self.mbed.serial_read(1)
+            if c is None:
+                self.print_result("ioerr_serial")
+                return
             if i > 2: # we will ignore first few measurements
                 delta = time() - start
                 deviation = abs(delta - 1)