ARMmbed
/
mbed-os
mirror of https://github.com/ARMmbed/mbed-os.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Python2+3: tests import correctly

pull/5848/head
Jimmy Brisson 2018-01-12 15:36:14 -06:00
parent 7abeec9744
commit 45bdd98a85
22 changed files with 219 additions and 159 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
tools/build_travis.py
View File

@ -414,13 +414,6 @@ def run_test_linking(dry_run, vendor):
if os.system(cmdline) != 0:
sys.exit(1)
def run_test_testsuite(dry_run, vendor):
cmdline = "python tools/singletest.py --version"
print "Executing: " + cmdline
if not dry_run:
if os.system(cmdline) != 0:
sys.exit(1)
if __name__ == "__main__":
parser = ArgumentParser()
@ -434,4 +427,3 @@ if __name__ == "__main__":
run_builds("-s" in sys.argv, options.vendor)
run_test_linking("-s" in sys.argv, options.vendor)
run_test_testsuite("-s" in sys.argv, options.vendor)

23
tools/detect_targets.py
View File

@ -15,6 +15,7 @@ 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.
"""
from __future__ import print_function
import sys
import os
import re
@ -67,8 +68,8 @@ def main():
# Only prints matrix of supported toolchains
if options.supported_toolchains:
print mcu_toolchain_matrix(
platform_filter=options.general_filter_regex)
print(mcu_toolchain_matrix(
platform_filter=options.general_filter_regex))
exit(0)
# If auto_detect attribute is present, we assume other auto-detection
@ -81,23 +82,23 @@ def main():
for mut in muts.values():
if re.match(mcu_filter, mut['mcu']):
interface_version = get_interface_version(mut['disk'])
print ""
print "[mbed] Detected %s, port %s, mounted %s, interface version %s:" % \
(mut['mcu'], mut['port'], mut['disk'], interface_version)
print "[mbed] Supported toolchains for %s" % mut['mcu']
print mcu_toolchain_matrix(platform_filter=mut['mcu'])
print("")
print("[mbed] Detected %s, port %s, mounted %s, interface "
"version %s:" %
(mut['mcu'], mut['port'], mut['disk'], interface_version))
print("[mbed] Supported toolchains for %s" % mut['mcu'])
print(mcu_toolchain_matrix(platform_filter=mut['mcu']))
count += 1
if count == 0:
print "[mbed] No mbed targets were detected on your system."
print("[mbed] No mbed targets were detected on your system.")
except KeyboardInterrupt:
print "\n[CTRL+c] exit"
print("\n[CTRL+c] exit")
except Exception as exc:
import traceback
traceback.print_exc(file=sys.stdout)
print "[ERROR] %s" % str(exc)
print("[ERROR] %s" % str(exc))
sys.exit(1)
def get_interface_version(mount_point):

34
tools/host_tests/__init__.py
View File

@ -15,25 +15,25 @@ See the License for the specific language governing permissions and
limitations under the License.
"""
from host_registry import HostRegistry
from .host_registry import HostRegistry
# Host test supervisors
from echo import EchoTest
from rtc_auto import RTCTest
from stdio_auto import StdioTest
from hello_auto import HelloTest
from detect_auto import DetectPlatformTest
from default_auto import DefaultAuto
from dev_null_auto import DevNullTest
from wait_us_auto import WaitusTest
from tcpecho_server_auto import TCPEchoServerTest
from udpecho_server_auto import UDPEchoServerTest
from tcpecho_client_auto import TCPEchoClientTest
from udpecho_client_auto import UDPEchoClientTest
from wfi_auto import WFITest
from serial_nc_rx_auto import SerialNCRXTest
from serial_nc_tx_auto import SerialNCTXTest
from serial_complete_auto import SerialCompleteTest
from .echo import EchoTest
from .rtc_auto import RTCTest
from .stdio_auto import StdioTest
from .hello_auto import HelloTest
from .detect_auto import DetectPlatformTest
from .default_auto import DefaultAuto
from .dev_null_auto import DevNullTest
from .wait_us_auto import WaitusTest
from .tcpecho_server_auto import TCPEchoServerTest
from .udpecho_server_auto import UDPEchoServerTest
from .tcpecho_client_auto import TCPEchoClientTest
from .udpecho_client_auto import UDPEchoClientTest
from .wfi_auto import WFITest
from .serial_nc_rx_auto import SerialNCRXTest
from .serial_nc_tx_auto import SerialNCTXTest
from .serial_complete_auto import SerialCompleteTest
# Populate registry with supervising objects
HOSTREGISTRY = HostRegistry()

3
tools/host_tests/default_auto.py
View File

@ -14,6 +14,7 @@ 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.
"""
from __future__ import print_function
from sys import stdout
@ -30,7 +31,7 @@ class DefaultAuto():
return selftest.RESULT_IO_SERIAL
stdout.write(c)
stdout.flush()
except KeyboardInterrupt, _:
except KeyboardInterrupt:
selftest.notify("\r\n[CTRL+C] exit")
result = selftest.RESULT_ERROR
return result

21
tools/host_tests/host_tests_plugins/__init__.py
View File

@ -14,26 +14,27 @@ 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.
"""
from __future__ import print_function
import host_test_registry
from . import host_test_registry
# This plugins provide 'flashing' methods to host test scripts
import module_copy_mbed
import module_copy_shell
import module_copy_silabs
from . import module_copy_mbed
from . import module_copy_shell
from . import module_copy_silabs
try:
import module_copy_smart
from . import module_copy_smart
except:
pass
#import module_copy_firefox
import module_copy_mps2
from . import module_copy_mps2
# Plugins used to reset certain platform
import module_reset_mbed
import module_reset_silabs
import module_reset_mps2
from . import module_reset_mbed
from . import module_reset_silabs
from . import module_reset_mps2
# Plugin registry instance
@ -77,4 +78,4 @@ def get_plugin_caps(type):
def print_plugin_info():
""" Prints plugins' information in user friendly way
"""
print HOST_TEST_PLUGIN_REGISTRY
print(HOST_TEST_PLUGIN_REGISTRY)

8
tools/host_tests/host_tests_plugins/host_test_plugins.py
View File

@ -14,6 +14,7 @@ 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.
"""
from __future__ import print_function
from os import access, F_OK
from sys import stdout
@ -58,16 +59,13 @@ class HostTestPluginBase:
def print_plugin_error(self, text):
""" Function prints error in console and exits always with False
"""
print "Plugin error: %s::%s: %s"% (self.name, self.type, text)
print("Plugin error: %s::%s: %s" % (self.name, self.type, text))
return False
def print_plugin_info(self, text, NL=True):
""" Function prints notification in console and exits always with True
"""
if NL:
print "Plugin info: %s::%s: %s"% (self.name, self.type, text)
else:
print "Plugin info: %s::%s: %s"% (self.name, self.type, text),
print("Plugin info: %s::%s: %s"% (self.name, self.type, text))
return True
def print_plugin_char(self, char):

3
tools/host_tests/host_tests_plugins/host_test_registry.py
View File

@ -14,6 +14,7 @@ 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.
"""
from __future__ import print_function
class HostTestRegistry:
""" Simple class used to register and store
@ -23,7 +24,7 @@ class HostTestRegistry:
PLUGINS = {} # 'Plugin Name' : Plugin Object
def print_error(self, text):
print "Plugin load failed. Reason: %s"% text
print("Plugin load failed. Reason: %s" % text)
def register_plugin(self, plugin):
""" Registers and stores plugin inside registry for further use.

5
tools/host_tests/host_tests_plugins/module_copy_mbed.py
View File

@ -14,9 +14,10 @@ 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.
"""
from __future__ import print_function
from shutil import copy
from host_test_plugins import HostTestPluginBase
from .host_test_plugins import HostTestPluginBase
from time import sleep
@ -32,7 +33,7 @@ class HostTestPluginCopyMethod_Mbed(HostTestPluginBase):
destination_disk += '/'
try:
copy(image_path, destination_disk)
except Exception, e:
except Exception as e:
self.print_plugin_error("shutil.copy('%s', '%s')"% (image_path, destination_disk))
self.print_plugin_error("Error: %s"% str(e))
result = False

3
tools/host_tests/host_tests_plugins/module_copy_mps2.py
View File

@ -14,12 +14,13 @@ 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.
"""
from __future__ import print_function
import re
import os, shutil
from os.path import join
from host_test_plugins import HostTestPluginBase
from time import sleep
from .host_test_plugins import HostTestPluginBase
class HostTestPluginCopyMethod_MPS2(HostTestPluginBase):

3
tools/host_tests/host_tests_plugins/module_copy_shell.py
View File

@ -14,12 +14,13 @@ 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.
"""
from __future__ import print_function
import os
from os.path import join, basename
from host_test_plugins import HostTestPluginBase
from time import sleep
from .host_test_plugins import HostTestPluginBase
class HostTestPluginCopyMethod_Shell(HostTestPluginBase):

3
tools/host_tests/host_tests_plugins/module_copy_silabs.py
View File

@ -14,9 +14,10 @@ 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.
"""
from __future__ import print_function
from host_test_plugins import HostTestPluginBase
from time import sleep
from .host_test_plugins import HostTestPluginBase
class HostTestPluginCopyMethod_Silabs(HostTestPluginBase):

3
tools/host_tests/host_tests_plugins/module_reset_mbed.py
View File

@ -14,8 +14,9 @@ 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.
"""
from __future__ import print_function
from host_test_plugins import HostTestPluginBase
from .host_test_plugins import HostTestPluginBase
class HostTestPluginResetMethod_Mbed(HostTestPluginBase):

3
tools/host_tests/host_tests_plugins/module_reset_mps2.py
View File

@ -14,10 +14,11 @@ 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.
"""
from __future__ import print_function
import os
from host_test_plugins import HostTestPluginBase
from time import sleep
from .host_test_plugins import HostTestPluginBase
# Note: This plugin is not fully functional, needs improvements

3
tools/host_tests/host_tests_plugins/module_reset_silabs.py
View File

@ -14,8 +14,9 @@ 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.
"""
from __future__ import print_function
from host_test_plugins import HostTestPluginBase
from .host_test_plugins import HostTestPluginBase
class HostTestPluginResetMethod_SiLabs(HostTestPluginBase):

13
tools/host_tests/tcpecho_client_auto.py
View File

@ -14,17 +14,21 @@ 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.
"""
from __future__ import print_function
import sys
import socket
from sys import stdout
from SocketServer import BaseRequestHandler, TCPServer
try:
from SocketServer import BaseRequestHandler, TCPServer
except ImportError:
from socketserver import BaseRequestHandler, TCPServer
class TCPEchoClient_Handler(BaseRequestHandler):
def handle(self):
""" One handle per connection
"""
print "HOST: Connection received...",
print("HOST: Connection received...")
count = 1;
while True:
data = self.request.recv(1024)
@ -32,7 +36,7 @@ class TCPEchoClient_Handler(BaseRequestHandler):
self.request.sendall(data)
if '{{end}}' in str(data):
print
print str(data)
print(str(data))
else:
if not count % 10:
sys.stdout.write('.')
@ -82,6 +86,7 @@ class TCPEchoClientTest():
# Returning none will suppress host test from printing success code
server = TCPServer((SERVER_IP, SERVER_PORT), TCPEchoClient_Handler)
print "HOST: Listening for TCP connections: " + SERVER_IP + ":" + str(SERVER_PORT)
print("HOST: Listening for TCP connections: %s:%s" %
(SERVER_IP, str(SERVER_PORT)))
self.send_server_ip_port(selftest, SERVER_IP, SERVER_PORT)
server.serve_forever()

15
tools/host_tests/tcpecho_server_auto.py
View File

@ -14,6 +14,7 @@ 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.
"""
from __future__ import print_function
import re
import sys
@ -47,18 +48,18 @@ class TCPEchoServerTest():
try:
self.s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
self.s.connect((self.ECHO_SERVER_ADDRESS, self.ECHO_PORT))
except Exception, e:
except Exception as e:
self.s = None
selftest.notify("HOST: Socket error: %s"% e)
return selftest.RESULT_ERROR
print 'HOST: Sending %d echo strings...'% self.ECHO_LOOPs,
print('HOST: Sending %d echo strings...'% self.ECHO_LOOPs,)
for i in range(0, self.ECHO_LOOPs):
TEST_STRING = str(uuid.uuid4())
try:
self.s.sendall(TEST_STRING)
data = self.s.recv(128)
except Exception, e:
except Exception as e:
self.s = None
selftest.notify("HOST: Socket error: %s"% e)
return selftest.RESULT_ERROR
@ -69,10 +70,10 @@ class TCPEchoServerTest():
stdout.flush()
result = True
else:
print "Expected: "
print "'%s'"% TEST_STRING
print "received: "
print "'%s'"% received_str
print("Expected: ")
print("'%s'"% TEST_STRING)
print("received: ")
print("'%s'"% received_str)
result = False
break

11
tools/host_tests/udpecho_client_auto.py
View File

@ -14,11 +14,15 @@ 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.
"""
from __future__ import print_function
import sys
import socket
from sys import stdout
from SocketServer import BaseRequestHandler, UDPServer
try:
from SocketServer import BaseRequestHandler, UDPServer
except ImportError:
from socketserver import BaseRequestHandler, UDPServer
class UDPEchoClient_Handler(BaseRequestHandler):
def handle(self):
@ -27,8 +31,7 @@ class UDPEchoClient_Handler(BaseRequestHandler):
data, socket = self.request
socket.sendto(data, self.client_address)
if '{{end}}' in data:
print
print data
print("\n%s" % data)
else:
sys.stdout.write('.')
stdout.flush()
@ -72,6 +75,6 @@ class UDPEchoClientTest():
# Returning none will suppress host test from printing success code
server = UDPServer((SERVER_IP, SERVER_PORT), UDPEchoClient_Handler)
print "HOST: Listening for UDP connections..."
print("HOST: Listening for UDP connections...")
self.send_server_ip_port(selftest, SERVER_IP, SERVER_PORT)
server.serve_forever()

3
tools/host_tests/udpecho_server_auto.py
View File

@ -14,6 +14,7 @@ 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.
"""
from __future__ import print_function
import re
import sys
@ -45,7 +46,7 @@ class UDPEchoServerTest():
# We assume this test fails so can't send 'error' message to server
try:
self.s = socket(AF_INET, SOCK_DGRAM)
except Exception, e:
except Exception as e:
self.s = None
selftest.notify("HOST: Socket error: %s"% e)
return selftest.RESULT_ERROR

2
tools/memap.py
View File

@ -16,7 +16,7 @@ from copy import deepcopy
from prettytable import PrettyTable
from tools.arm_pack_manager import Cache
from utils import argparse_filestring_type, \
from tools.utils import argparse_filestring_type, \
argparse_lowercase_hyphen_type, argparse_uppercase_type

197
tools/test_api.py
View File

@ -16,6 +16,7 @@ limitations under the License.
Author: Przemyslaw Wirkus <Przemyslaw.wirkus@arm.com>
"""
from __future__ import print_function
import os
import re
@ -28,14 +29,15 @@ import argparse
import datetime
import threading
import ctypes
import functools
from types import ListType
from colorama import Fore, Back, Style
from prettytable import PrettyTable
from copy import copy
from time import sleep, time
from Queue import Queue, Empty
try:
from Queue import Queue, Empty
except ImportError:
from queue import Queue, Empty
from os.path import join, exists, basename, relpath
from threading import Thread, Lock
from multiprocessing import Pool, cpu_count
@ -100,7 +102,7 @@ class ProcessObserver(Thread):
self.active = False
try:
self.proc.terminate()
except Exception, _:
except Exception:
pass
@ -120,12 +122,14 @@ class SingleTestExecutor(threading.Thread):
# Human readable summary
if not self.single_test.opts_suppress_summary:
# prints well-formed summary with results (SQL table like)
print self.single_test.generate_test_summary(test_summary, shuffle_seed)
print(self.single_test.generate_test_summary(test_summary,
shuffle_seed))
if self.single_test.opts_test_x_toolchain_summary:
# prints well-formed summary with results (SQL table like)
# table shows text x toolchain test result matrix
print self.single_test.generate_test_summary_by_target(test_summary, shuffle_seed)
print "Completed in %.2f sec"% (elapsed_time)
print(self.single_test.generate_test_summary_by_target(
test_summary, shuffle_seed))
print("Completed in %.2f sec"% (elapsed_time))
class SingleTestRunner(object):
@ -360,31 +364,40 @@ class SingleTestRunner(object):
# print '=== %s::%s ===' % (target, toolchain)
# Let's build our test
if target not in TARGET_MAP:
print self.logger.log_line(self.logger.LogType.NOTIF, 'Skipped tests for %s target. Target platform not found'% (target))
print(self.logger.log_line(
self.logger.LogType.NOTIF,
'Skipped tests for %s target. Target platform not found' %
(target)))
continue
clean_mbed_libs_options = True if self.opts_goanna_for_mbed_sdk or clean or self.opts_clean else None
clean_mbed_libs_options = (self.opts_goanna_for_mbed_sdk or
self.opts_clean or clean)
profile = extract_profile(self.opts_parser, self.opts, toolchain)
stats_depth = self.opts.stats_depth or 2
try:
build_mbed_libs_result = build_mbed_libs(T,
toolchain,
clean=clean_mbed_libs_options,
verbose=self.opts_verbose,
jobs=self.opts_jobs,
report=build_report,
properties=build_properties,
build_profile=profile)
build_mbed_libs_result = build_mbed_libs(
T, toolchain,
clean=clean_mbed_libs_options,
verbose=self.opts_verbose,
jobs=self.opts_jobs,
report=build_report,
properties=build_properties,
build_profile=profile)
if not build_mbed_libs_result:
print self.logger.log_line(self.logger.LogType.NOTIF, 'Skipped tests for %s target. Toolchain %s is not yet supported for this target'% (T.name, toolchain))
print(self.logger.log_line(
self.logger.LogType.NOTIF,
'Skipped tests for %s target. Toolchain %s is not '
'supported for this target'% (T.name, toolchain)))
continue
except ToolException:
print self.logger.log_line(self.logger.LogType.ERROR, 'There were errors while building MBED libs for %s using %s'% (target, toolchain))
print(self.logger.log_line(
self.logger.LogType.ERROR,
'There were errors while building MBED libs for %s using %s'
% (target, toolchain)))
continue
build_dir = join(BUILD_DIR, "test", target, toolchain)
@ -402,16 +415,22 @@ class SingleTestRunner(object):
if self.db_logger:
self.db_logger.reconnect();
if self.db_logger.is_connected():
self.db_logger.update_build_id_info(self.db_logger_build_id, _shuffle_seed=self.shuffle_random_func())
self.db_logger.update_build_id_info(
self.db_logger_build_id,
_shuffle_seed=self.shuffle_random_func())
self.db_logger.disconnect();
if self.db_logger:
self.db_logger.reconnect();
if self.db_logger.is_connected():
# Update MUTs and Test Specification in database
self.db_logger.update_build_id_info(self.db_logger_build_id, _muts=self.muts, _test_spec=self.test_spec)
self.db_logger.update_build_id_info(
self.db_logger_build_id,
_muts=self.muts, _test_spec=self.test_spec)
# Update Extra information in database (some options passed to test suite)
self.db_logger.update_build_id_info(self.db_logger_build_id, _extra=json.dumps(self.dump_options()))
self.db_logger.update_build_id_info(
self.db_logger_build_id,
_extra=json.dumps(self.dump_options()))
self.db_logger.disconnect();
valid_test_map_keys = self.get_valid_tests(test_map_keys, target, toolchain, test_ids, self.opts_include_non_automated)
@ -449,7 +468,9 @@ class SingleTestRunner(object):
build_profile=profile)
except ToolException:
print self.logger.log_line(self.logger.LogType.ERROR, 'There were errors while building library %s'% (lib_id))
print(self.logger.log_line(
self.logger.LogType.ERROR,
'There were errors while building library %s' % lib_id))
continue
@ -491,23 +512,29 @@ class SingleTestRunner(object):
project_description=test.get_description(),
build_profile=profile, stats_depth=stats_depth)
except Exception, e:
except Exception as e:
project_name_str = project_name if project_name is not None else test_id
test_result = self.TEST_RESULT_FAIL
if isinstance(e, ToolException):
print self.logger.log_line(self.logger.LogType.ERROR, 'There were errors while building project %s'% (project_name_str))
print(self.logger.log_line(
self.logger.LogType.ERROR,
'There were errors while building project %s' %
project_name_str))
test_result = self.TEST_RESULT_BUILD_FAILED
elif isinstance(e, NotSupportedException):
print self.logger.log_line(self.logger.LogType.INFO, 'The project %s is not supported'% (project_name_str))
print(elf.logger.log_line(
self.logger.LogType.INFO,
'Project %s is not supported' % project_name_str))
test_result = self.TEST_RESULT_NOT_SUPPORTED
# Append test results to global test summary
self.test_summary.append(
(test_result, target, toolchain, test_id, test.get_description(), 0, 0, '-')
(test_result, target, toolchain, test_id,
test.get_description(), 0, 0, '-')
)
# Add detailed test result to test summary structure
@ -642,23 +669,33 @@ class SingleTestRunner(object):
if self.opts_test_only_peripheral and not test.peripherals:
if self.opts_verbose_skipped_tests:
print self.logger.log_line(self.logger.LogType.INFO, 'Common test skipped for target %s'% (target))
print(self.logger.log_line(
self.logger.LogType.INFO,
'Common test skipped for target %s' % target))
continue
if self.opts_peripheral_by_names and test.peripherals and not len([i for i in test.peripherals if i in self.opts_peripheral_by_names]):
if (self.opts_peripheral_by_names and test.peripherals and
not any((i in self.opts_peripheral_by_names)
for i in test.peripherals)):
# We will skip tests not forced with -p option
if self.opts_verbose_skipped_tests:
print self.logger.log_line(self.logger.LogType.INFO, 'Common test skipped for target %s'% (target))
print(self.logger.log_line(
self.logger.LogType.INFO,
'Common test skipped for target %s' % target))
continue
if self.opts_test_only_common and test.peripherals:
if self.opts_verbose_skipped_tests:
print self.logger.log_line(self.logger.LogType.INFO, 'Peripheral test skipped for target %s'% (target))
print(self.logger.log_line(
self.logger.LogType.INFO,
'Peripheral test skipped for target %s' % target))
continue
if not include_non_automated and not test.automated:
if self.opts_verbose_skipped_tests:
print self.logger.log_line(self.logger.LogType.INFO, 'Non automated test skipped for target %s'% (target))
print(self.logger.log_line(
self.logger.LogType.INFO,
'Non automated test skipped for target %s' % target))
continue
if test.is_supported(target, toolchain):
@ -673,9 +710,15 @@ class SingleTestRunner(object):
elif not self.is_peripherals_available(target, test.peripherals):
if self.opts_verbose_skipped_tests:
if test.peripherals:
print self.logger.log_line(self.logger.LogType.INFO, 'Peripheral %s test skipped for target %s'% (",".join(test.peripherals), target))
print(self.logger.log_line(
self.logger.LogType.INFO,
'Peripheral %s test skipped for target %s' %
(",".join(test.peripherals), target)))
else:
print self.logger.log_line(self.logger.LogType.INFO, 'Test %s skipped for target %s'% (test_id, target))
print(self.logger.log_line(
self.logger.LogType.INFO,
'Test %s skipped for target %s' %
(test_id, target)))
continue
# The test has made it through all the filters, so add it to the valid tests list
@ -812,7 +855,7 @@ class SingleTestRunner(object):
resutl_msg = ""
try:
os.remove(file_path)
except Exception, e:
except Exception as e:
resutl_msg = e
result = False
return result, resutl_msg
@ -828,7 +871,7 @@ class SingleTestRunner(object):
duration = data.get("duration", 10)
if mut is None:
print "Error: No Mbed available: MUT[%s]" % data['mcu']
print("Error: No Mbed available: MUT[%s]" % data['mcu'])
return None
mcu = mut['mcu']
@ -864,7 +907,7 @@ class SingleTestRunner(object):
break
if not found:
print "Error: mbed not found with MBEDLS: %s" % data['mcu']
print("Error: mbed not found with MBEDLS: %s" % data['mcu'])
return None
else:
mut = muts_list[1]
@ -895,7 +938,7 @@ class SingleTestRunner(object):
single_test_result = self.TEST_RESULT_NO_IMAGE
elapsed_time = 0
single_test_output = self.logger.log_line(self.logger.LogType.ERROR, 'Image file does not exist: %s'% image_path)
print single_test_output
print(single_test_output)
else:
# Host test execution
start_host_exec_time = time()
@ -930,8 +973,9 @@ class SingleTestRunner(object):
'copy_method' : _copy_method,
}
print self.print_test_result(single_test_result, target_name_unique, toolchain_name,
test_id, test_description, elapsed_time, single_timeout)
print(self.print_test_result(
single_test_result, target_name_unique, toolchain_name, test_id,
test_description, elapsed_time, single_timeout))
# Update database entries for ongoing test
if self.db_logger and self.db_logger.is_connected():
@ -1027,7 +1071,7 @@ class SingleTestRunner(object):
"""
try:
c = obs.queue.get(block=True, timeout=0.5)
except Empty, _:
except Empty:
c = None
return c
@ -1060,7 +1104,6 @@ class SingleTestRunner(object):
result = property.groups()[0]
return result
# print "{%s} port:%s disk:%s" % (name, port, disk),
cmd = ["python",
'%s.py'% name,
'-d', disk,
@ -1083,8 +1126,8 @@ class SingleTestRunner(object):
cmd += ["-R", str(reset_tout)]
if verbose:
print Fore.MAGENTA + "Executing '" + " ".join(cmd) + "'" + Fore.RESET
print "Test::Output::Start"
print(Fore.MAGENTA + "Executing '" + " ".join(cmd) + "'" + Fore.RESET)
print("Test::Output::Start")
proc = Popen(cmd, stdout=PIPE, cwd=HOST_TESTS)
obs = ProcessObserver(proc)
@ -1138,7 +1181,7 @@ class SingleTestRunner(object):
output.append(c)
if verbose:
print "Test::Output::Finish"
print("Test::Output::Finish")
# Stop test process
obs.stop()
@ -1205,9 +1248,10 @@ def show_json_file_format_error(json_spec_filename, line, column):
line_no = 1
for json_line in data_file:
if line_no + 5 >= line: # Print last few lines before error
print 'Line %d:\t'%line_no + json_line, # Prints line
print('Line %d:\t'%line_no + json_line)
if line_no == line:
print ' ' * len('Line %d:'%line_no) + '\t', '-' * (column-1) + '^'
print('%s\t%s^' (' ' * len('Line %d:' % line_no),
'-' * (column - 1)))
break
line_no += 1
@ -1244,18 +1288,19 @@ def get_json_data_from_file(json_spec_filename, verbose=False):
result = json.load(data_file)
except ValueError as json_error_msg:
result = None
print 'JSON file %s parsing failed. Reason: %s' % (json_spec_filename, json_error_msg)
print('JSON file %s parsing failed. Reason: %s' %
(json_spec_filename, json_error_msg))
# We can print where error occurred inside JSON file if we can parse exception msg
json_format_defect_pos = json_format_error_defect_pos(str(json_error_msg))
if json_format_defect_pos is not None:
line = json_format_defect_pos[0]
column = json_format_defect_pos[1]
print
print()
show_json_file_format_error(json_spec_filename, line, column)
except IOError as fileopen_error_msg:
print 'JSON file %s not opened. Reason: %s'% (json_spec_filename, fileopen_error_msg)
print
print('JSON file %s not opened. Reason: %s\n'%
(json_spec_filename, fileopen_error_msg))
if verbose and result:
pp = pprint.PrettyPrinter(indent=4)
pp.pprint(result)
@ -1290,7 +1335,7 @@ def print_muts_configuration_from_json(json_data, join_delim=", ", platform_filt
if add_row:
for col in muts_info_cols:
cell_val = mut_info[col] if col in mut_info else None
if type(cell_val) == ListType:
if isinstance(cell_val, list):
cell_val = join_delim.join(cell_val)
row.append(cell_val)
pt.add_row(row)
@ -1423,7 +1468,7 @@ def get_avail_tests_summary_table(cols=None, result_summary=True, join_delim=','
for col in test_properties:
col_value = test[col]
if type(test[col]) == ListType:
if isinstance(test[col], list):
col_value = join_delim.join(test[col])
elif test[col] == None:
col_value = "-"
@ -1502,13 +1547,14 @@ def singletest_in_cli_mode(single_test):
# Human readable summary
if not single_test.opts_suppress_summary:
# prints well-formed summary with results (SQL table like)
print single_test.generate_test_summary(test_summary, shuffle_seed)
print(single_test.generate_test_summary(test_summary, shuffle_seed))
if single_test.opts_test_x_toolchain_summary:
# prints well-formed summary with results (SQL table like)
# table shows text x toolchain test result matrix
print single_test.generate_test_summary_by_target(test_summary, shuffle_seed)
print(single_test.generate_test_summary_by_target(test_summary,
shuffle_seed))
print "Completed in %.2f sec"% (elapsed_time)
print("Completed in %.2f sec" % elapsed_time)
print
# Write summary of the builds
@ -1628,19 +1674,19 @@ def detect_database_verbose(db_url):
# Let's try to connect
db_ = factory_db_logger(db_url)
if db_ is not None:
print "Connecting to database '%s'..."% db_url,
print("Connecting to database '%s'..." % db_url)
db_.connect(host, username, password, db_name)
if db_.is_connected():
print "ok"
print "Detecting database..."
print db_.detect_database(verbose=True)
print "Disconnecting...",
print("ok")
print("Detecting database...")
print(db_.detect_database(verbose=True))
print("Disconnecting...")
db_.disconnect()
print "done"
print("done")
else:
print "Database type '%s' unknown"% db_type
print("Database type '%s' unknown" % db_type)
else:
print "Parse error: '%s' - DB Url error"% (db_url)
print("Parse error: '%s' - DB Url error" % db_url)
def get_module_avail(module_name):
@ -2089,13 +2135,14 @@ def print_tests(tests, format="list", sort=True):
if format == "list":
for test_name in sorted(tests.keys()):
test_path = tests[test_name][0]
print "Test Case:"
print " Name: %s" % test_name
print " Path: %s" % test_path
print("Test Case:")
print(" Name: %s" % test_name)
print(" Path: %s" % test_path)
elif format == "json":
print json.dumps({test_name: test_path[0] for test_name, test_paths in tests}, indent=2)
print(json.dumps({test_name: test_path[0] for test_name, test_paths
in tests}, indent=2))
else:
print "Unknown format '%s'" % format
print("Unknown format '%s'" % format)
sys.exit(1)
def norm_relative_path(path, start):
@ -2141,11 +2188,11 @@ def build_test_worker(*args, **kwargs):
ret['bin_file'] = bin_file
ret['kwargs'] = kwargs
except NotSupportedException, e:
except NotSupportedException as e:
ret['reason'] = e
except ToolException, e:
except ToolException as e:
ret['reason'] = e
except KeyboardInterrupt, e:
except KeyboardInterrupt as e:
ret['reason'] = e
except:
# Print unhandled exceptions here
@ -2265,8 +2312,8 @@ def build_tests(tests, base_source_paths, build_path, target, toolchain_name,
test_key = worker_result['kwargs']['project_id'].upper()
if report:
print report[target_name][toolchain_name][test_key][0][0]['output'].rstrip()
print 'Image: %s\n' % bin_file
print(report[target_name][toolchain_name][test_key][0][0]['output'].rstrip())
print('Image: %s\n' % bin_file)
except:
if p._taskqueue.queue:

12
tools/test_exporters.py
View File

@ -304,11 +304,13 @@ class ReportExporter():
def exporter_print_helper(self, array, print_log=False):
for item in array:
print " * %s::%s::%s" % (item["target_name"], item["toolchain_name"], item["id"])
print(" * %s::%s::%s" % (item["target_name"],
item["toolchain_name"],
item["id"]))
if print_log:
log_lines = item["output"].split("\n")
for log_line in log_lines:
print " %s" % log_line
print(" %s" % log_line)
def exporter_print(self, test_result_ext, print_log_for_failures=False):
""" Export test results in print format.
@ -343,15 +345,15 @@ class ReportExporter():
raise Exception("'test_run' did not have a 'result' value")
if successes:
print "\n\nBuild successes:"
print("\n\nBuild successes:")
self.exporter_print_helper(successes)
if skips:
print "\n\nBuild skips:"
print("\n\nBuild skips:")
self.exporter_print_helper(skips)
if failures:
print "\n\nBuild failures:"
print("\n\nBuild failures:")
self.exporter_print_helper(failures, print_log=print_log_for_failures)
return False
else:

2
tools/tests.py
View File

@ -17,7 +17,7 @@ limitations under the License.
from tools.paths import *
from tools.data.support import DEFAULT_SUPPORT, CORTEX_ARM_SUPPORT
from argparse import ArgumentTypeError
from utils import columnate
from tools.utils import columnate
TEST_CMSIS_LIB = join(TEST_DIR, "cmsis", "lib")
TEST_MBED_LIB = join(TEST_DIR, "mbed", "env")