test_webapi.py

00001 """
00002 mbed SDK
00003 Copyright (c) 2011-2014 ARM Limited
00004 
00005 Licensed under the Apache License, Version 2.0 (the "License");
00006 you may not use this file except in compliance with the License.
00007 You may obtain a copy of the License at
00008 
00009     http://www.apache.org/licenses/LICENSE-2.0
00010 
00011 Unless required by applicable law or agreed to in writing, software
00012 distributed under the License is distributed on an "AS IS" BASIS,
00013 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
00014 See the License for the specific language governing permissions and
00015 limitations under the License.
00016 
00017 Author: Przemyslaw Wirkus <Przemyslaw.wirkus@arm.com>
00018 """
00019 
00020 import sys
00021 import json
00022 import optparse
00023 from flask import Flask
00024 from os.path import join, abspath, dirname
00025 
00026 # Be sure that the tools directory is in the search path
00027 ROOT = abspath(join(dirname(__file__), ".."))
00028 sys.path.insert(0, ROOT)
00029 
00030 # Imports related to mbed build api
00031 from tools.utils import construct_enum
00032 from tools.build_api import mcu_toolchain_matrix
00033 
00034 # Imports from TEST API
00035 from test_api import SingleTestRunner
00036 from test_api import SingleTestExecutor
00037 from test_api import get_json_data_from_file
00038 from test_api import print_muts_configuration_from_json
00039 from test_api import print_test_configuration_from_json
00040 from test_api import get_avail_tests_summary_table
00041 from test_api import get_default_test_options_parser
00042 
00043 
00044 class SingleTestRunnerWebService(SingleTestRunner):
00045     def __init__(self):
00046         super(SingleTestRunnerWebService, self).__init__()
00047 
00048         # With this lock we should control access to certain resources inside this class
00049         self.resource_lock = thread.allocate_lock()
00050 
00051         self.RestRequest = construct_enum(REST_MUTS='muts',
00052                                           REST_TEST_SPEC='test_spec',
00053                                           REST_TEST_RESULTS='test_results')
00054 
00055     def get_rest_result_template(self, result, command, success_code):
00056         """ Returns common part of every web service request
00057         """
00058         result = {"result" : result,
00059                   "command" : command,
00060                   "success_code": success_code} # 0 - OK, >0 - Error number
00061         return result
00062 
00063     # REST API handlers for Flask framework
00064     def rest_api_status(self):
00065         """ Returns current test execution status. E.g. running / finished etc.
00066         """
00067         with self.resource_lock:
00068             pass
00069 
00070     def rest_api_config(self):
00071         """ Returns configuration passed to SingleTest executor
00072         """
00073         with self.resource_lock:
00074             pass
00075 
00076     def rest_api_log(self):
00077         """ Returns current test log """
00078         with self.resource_lock:
00079             pass
00080 
00081     def rest_api_request_handler(self, request_type):
00082         """ Returns various data structures. Both static and mutable during test
00083         """
00084         result = {}
00085         success_code = 0
00086         with self.resource_lock:
00087             if request_type == self.RestRequest.REST_MUTS:
00088                 result = self.muts # Returns MUTs
00089             elif request_type == self.RestRequest.REST_TEST_SPEC:
00090                 result = self.test_spec # Returns Test Specification
00091             elif request_type == self.RestRequest.REST_TEST_RESULTS:
00092                 pass # Returns test results
00093             else:
00094                 success_code = -1
00095         return json.dumps(self.get_rest_result_template(result, 'request/' + request_type, success_code), indent=4)
00096 
00097 
00098 def singletest_in_webservice_mode():
00099     # TODO Implement this web service functionality
00100     pass
00101 
00102 
00103 def get_default_test_webservice_options_parser ():
00104     """ Get test script web service options used by CLI, webservices etc.
00105     """
00106     parser = get_default_test_options_parser()
00107 
00108     # Things related to web services offered by test suite scripts
00109     parser.add_option('', '--rest-api',
00110                       dest='rest_api_enabled',
00111                       default=False,
00112                       action="store_true",
00113                       help='Enables REST API.')
00114 
00115     parser.add_option('', '--rest-api-port',
00116                       dest='rest_api_port_no',
00117                       help='Sets port for REST API interface')
00118 
00119     return parser
00120 
00121 '''
00122 if __name__ == '__main__':
00123     # Command line options
00124     parser = get_default_test_options_parser()
00125 
00126     parser.description = """This script allows you to run mbed defined test cases for particular MCU(s) and corresponding toolchain(s)."""
00127     parser.epilog = """Example: singletest.py -i test_spec.json -M muts_all.json"""
00128 
00129     (opts, args) = parser.parse_args()
00130 
00131     # Print summary / information about automation test status
00132     if opts.test_automation_report:
00133         print get_avail_tests_summary_table()
00134         exit(0)
00135 
00136     # Print summary / information about automation test status
00137     if opts.test_case_report:
00138         test_case_report_cols = ['id', 'automated', 'description', 'peripherals', 'host_test', 'duration', 'source_dir']
00139         print get_avail_tests_summary_table(cols=test_case_report_cols, result_summary=False, join_delim='\n')
00140         exit(0)
00141 
00142     # Only prints matrix of supported toolchains
00143     if opts.supported_toolchains:
00144         print mcu_toolchain_matrix(platform_filter=opts.general_filter_regex)
00145         exit(0)
00146 
00147     # Open file with test specification
00148     # test_spec_filename tells script which targets and their toolchain(s)
00149     # should be covered by the test scenario
00150     test_spec = get_json_data_from_file(opts.test_spec_filename) if opts.test_spec_filename else None
00151     if test_spec is None:
00152         if not opts.test_spec_filename:
00153             parser.print_help()
00154         exit(-1)
00155 
00156     # Get extra MUTs if applicable
00157     MUTs = get_json_data_from_file(opts.muts_spec_filename) if opts.muts_spec_filename else None
00158 
00159     if MUTs is None:
00160         if not opts.muts_spec_filename:
00161             parser.print_help()
00162         exit(-1)
00163 
00164     # Only prints read MUTs configuration
00165     if MUTs and opts.verbose_test_configuration_only:
00166         print "MUTs configuration in %s:"% opts.muts_spec_filename
00167         print print_muts_configuration_from_json(MUTs)
00168         print
00169         print "Test specification in %s:"% opts.test_spec_filename
00170         print print_test_configuration_from_json(test_spec)
00171         exit(0)
00172 
00173     # Verbose test specification and MUTs configuration
00174     if MUTs and opts.verbose:
00175         print print_muts_configuration_from_json(MUTs)
00176     if test_spec and opts.verbose:
00177         print print_test_configuration_from_json(test_spec)
00178 
00179     if opts.only_build_tests:
00180         # We are skipping testing phase, and suppress summary
00181         opts.suppress_summary = True
00182 
00183     single_test = SingleTestRunner(_global_loops_count=opts.test_global_loops_value,
00184                                    _test_loops_list=opts.test_loops_list,
00185                                    _muts=MUTs,
00186                                    _test_spec=test_spec,
00187                                    _opts_goanna_for_mbed_sdk=opts.goanna_for_mbed_sdk,
00188                                    _opts_goanna_for_tests=opts.goanna_for_tests,
00189                                    _opts_shuffle_test_order=opts.shuffle_test_order,
00190                                    _opts_shuffle_test_seed=opts.shuffle_test_seed,
00191                                    _opts_test_by_names=opts.test_by_names,
00192                                    _opts_test_only_peripheral=opts.test_only_peripheral,
00193                                    _opts_test_only_common=opts.test_only_common,
00194                                    _opts_verbose_skipped_tests=opts.verbose_skipped_tests,
00195                                    _opts_verbose_test_result_only=opts.verbose_test_result_only,
00196                                    _opts_verbose=opts.verbose,
00197                                    _opts_firmware_global_name=opts.firmware_global_name,
00198                                    _opts_only_build_tests=opts.only_build_tests,
00199                                    _opts_suppress_summary=opts.suppress_summary,
00200                                    _opts_test_x_toolchain_summary=opts.test_x_toolchain_summary,
00201                                    _opts_copy_method=opts.copy_method
00202                                    )
00203 
00204     try:
00205         st_exec_thread = SingleTestExecutor(single_test)
00206     except KeyboardInterrupt, e:
00207         print "\n[CTRL+c] exit"
00208     st_exec_thread.start()
00209 
00210     if opts.rest_api_enabled:
00211         # Enable REST API
00212 
00213         app = Flask(__name__)
00214 
00215         @app.route('/')
00216         def hello_world():
00217             return 'Hello World!'
00218 
00219         @app.route('/status')
00220         def rest_api_status():
00221             return single_test.rest_api_status() # TODO
00222 
00223         @app.route('/config')
00224         def rest_api_config():
00225             return single_test.rest_api_config() # TODO
00226 
00227         @app.route('/log')
00228         def rest_api_log():
00229             return single_test.rest_api_log() # TODO
00230 
00231         @app.route('/request/<request_type>') # 'muts', 'test_spec', 'test_results'
00232         def rest_api_request_handler(request_type):
00233             result = single_test.rest_api_request_handler(request_type) # TODO
00234             return result
00235 
00236         rest_api_port = int(opts.rest_api_port_no) if opts.rest_api_port_no else 5555
00237         app.debug = False
00238         app.run(port=rest_api_port) # Blocking Flask REST API web service
00239     else:
00240         st_exec_thread.join()
00241 
00242 '''