Added 'real-time' test result printing when tests are executed (view test execution progress using option -V)

workspace_tools/host_tests/dev_null_auto.py

@@ -27,6 +27,7 @@ class DevNullTest(DefaultTest):
         # Data from serial received correctly
         print "Received %d bytes:"% len(c)
         print c
+        stdout.flush()
         # Check for expected and unexpected prints in Mbed output
         result = True
         if "re-routing stdout to /null" not in c:
@@ -39,7 +40,6 @@ class DevNullTest(DefaultTest):
             self.print_result('success')
         else:
             self.print_result('failure')
-        stdout.flush()
 
 if __name__ == '__main__':
     DevNullTest().run()

workspace_tools/host_tests/hello_auto.py

@@ -28,6 +28,7 @@ class HelloTest(DefaultTest):
            self.print_result("ioerr_serial")
            return
         print "Read %d bytes"% len(c)
+        stdout.flush()
         result = True
         # Because we can have targetID here let's try to decode
         if len(c) < len(self.HELLO_WORLD):
@@ -38,7 +39,7 @@ class HelloTest(DefaultTest):
             result = res is not None
         else:
             result = (c.startswith(self.HELLO_WORLD))
-            
+
         if result: # Hello World received
             self.print_result('success')
         else:

workspace_tools/host_tests/host_test.py

@@ -77,7 +77,7 @@ class Mbed:
 
         (self.options, _) = parser.parse_args()
 
-        self.DEFAULT_RESET_TOUT = 2
+        self.DEFAULT_RESET_TOUT = 0
         self.DEFAULT_TOUT = 10
 
         if self.options.port is None:

workspace_tools/host_tests/stdio_auto.py

@@ -31,6 +31,7 @@ class StdioTest(DefaultTest):
         for i in range(1, 5):
             random_integer = random.randint(-10000, 10000)
             print "Generated number: " + str(random_integer)
+            stdout.flush()
             self.mbed.serial_write(str(random_integer) + "\n")
             serial_stdio_msg = ""
 
@@ -61,7 +62,6 @@ class StdioTest(DefaultTest):
             self.print_result('success')
         else:
             self.print_result('failure')
-        stdout.flush()
 
 if __name__ == '__main__':
     StdioTest().run()

workspace_tools/host_tests/udp_link_layer_auto.py

@@ -130,7 +130,8 @@ class UDPEchoServerTest(DefaultTest):
             sleep(1)
             summary_datagram_success = (float(len(dict_udp_recv_datagrams)) / float(self.TEST_PACKET_COUNT)) * 100.0
             # print dict_udp_recv_datagrams
-            print "Datagrams recved after +%d sec: %.3f%% (%d / %d), stress=%.3f ms" % (d, summary_datagram_success, len(dict_udp_recv_datagrams), self.TEST_PACKET_COUNT, self.TEST_STRESS_FACTOR)
+            print "Datagrams received after +%d sec: %.3f%% (%d / %d), stress=%.3f ms" % (d, summary_datagram_success, len(dict_udp_recv_datagrams), self.TEST_PACKET_COUNT, self.TEST_STRESS_FACTOR)
+            stdout.flush()
 
         # Getting control data from test
         print
@@ -138,6 +139,7 @@ class UDPEchoServerTest(DefaultTest):
         mbed_stats = self.get_control_data()
         print mbed_stats
         print
+        stdout.flush()
 
         # Receiving serial data from mbed
         print

workspace_tools/host_tests/wait_us_auto.py

@@ -55,6 +55,7 @@ class WaitusTest(DefaultTest):
                 print ". in %.2f sec (%.2f) [%s]" % (delta, deviation, msg)
             else:
                 print ". skipped"
+            stdout.flush()
             start = time();
         measurement_time = time() - start_serial_pool
         print "Completed in %.2f sec" % (measurement_time)
@@ -63,7 +64,6 @@ class WaitusTest(DefaultTest):
             self.print_result('success')
         else:
             self.print_result('failure')
-        stdout.flush()
 
 if __name__ == '__main__':
     WaitusTest().run()

workspace_tools/test_api.py

@@ -19,6 +19,7 @@ Author: Przemyslaw Wirkus <Przemyslaw.wirkus@arm.com>
 
 import os
 import re
+import sys
 import json
 import pprint
 import random
@@ -812,6 +813,7 @@ class SingleTestRunner(object):
 
             if c:
                 output.append(c)
+                sys.stdout.write(c)
                 # Give the mbed under test a way to communicate the end of the test
                 if c in ['\n', '\r']:
                     if '{end}' in line:
@@ -820,6 +822,14 @@ class SingleTestRunner(object):
                 else:
                     line += c
 
+        try:
+            c = obs.queue.get(block=True, timeout=0.5)
+        except Empty, _:
+            c = None
+
+        if c:
+            output.append(c)
+            sys.stdout.write(c)
         # Stop test process
         obs.stop()