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_argument('', '--rest-api',
00110                         dest='rest_api_enabled',
00111                         default=False,
00112                         action="store_true",
00113                         help='Enables REST API.')
00114 
00115     parser.add_argument('', '--rest-api-port',
00116                         dest='rest_api_port_no',
00117                         type=int,
00118                         help='Sets port for REST API interface')
00119 
00120     return parser
00121 
00122 '''
00123 if __name__ == '__main__':
00124     # Command line options
00125     parser = get_default_test_options_parser()
00126 
00127     parser.description = """This script allows you to run mbed defined test cases for particular MCU(s) and corresponding toolchain(s)."""
00128     parser.epilog = """Example: singletest.py -i test_spec.json -M muts_all.json"""
00129 
00130     (opts, args) = parser.parse_args()
00131 
00132     # Print summary / information about automation test status
00133     if opts.test_automation_report:
00134         print get_avail_tests_summary_table()
00135         exit(0)
00136 
00137     # Print summary / information about automation test status
00138     if opts.test_case_report:
00139         test_case_report_cols = ['id', 'automated', 'description', 'peripherals', 'host_test', 'duration', 'source_dir']
00140         print get_avail_tests_summary_table(cols=test_case_report_cols, result_summary=False, join_delim='\n')
00141         exit(0)
00142 
00143     # Only prints matrix of supported toolchains
00144     if opts.supported_toolchains:
00145         print mcu_toolchain_matrix(platform_filter=opts.general_filter_regex)
00146         exit(0)
00147 
00148     # Open file with test specification
00149     # test_spec_filename tells script which targets and their toolchain(s)
00150     # should be covered by the test scenario
00151     test_spec = get_json_data_from_file(opts.test_spec_filename) if opts.test_spec_filename else None
00152     if test_spec is None:
00153         if not opts.test_spec_filename:
00154             parser.print_help()
00155         exit(-1)
00156 
00157     # Get extra MUTs if applicable
00158     MUTs = get_json_data_from_file(opts.muts_spec_filename) if opts.muts_spec_filename else None
00159 
00160     if MUTs is None:
00161         if not opts.muts_spec_filename:
00162             parser.print_help()
00163         exit(-1)
00164 
00165     # Only prints read MUTs configuration
00166     if MUTs and opts.verbose_test_configuration_only:
00167         print "MUTs configuration in %s:"% opts.muts_spec_filename
00168         print print_muts_configuration_from_json(MUTs)
00169         print
00170         print "Test specification in %s:"% opts.test_spec_filename
00171         print print_test_configuration_from_json(test_spec)
00172         exit(0)
00173 
00174     # Verbose test specification and MUTs configuration
00175     if MUTs and opts.verbose:
00176         print print_muts_configuration_from_json(MUTs)
00177     if test_spec and opts.verbose:
00178         print print_test_configuration_from_json(test_spec)
00179 
00180     if opts.only_build_tests:
00181         # We are skipping testing phase, and suppress summary
00182         opts.suppress_summary = True
00183 
00184     single_test = SingleTestRunner(_global_loops_count=opts.test_global_loops_value,
00185                                    _test_loops_list=opts.test_loops_list,
00186                                    _muts=MUTs,
00187                                    _test_spec=test_spec,
00188                                    _opts_goanna_for_mbed_sdk=opts.goanna_for_mbed_sdk,
00189                                    _opts_goanna_for_tests=opts.goanna_for_tests,
00190                                    _opts_shuffle_test_order=opts.shuffle_test_order,
00191                                    _opts_shuffle_test_seed=opts.shuffle_test_seed,
00192                                    _opts_test_by_names=opts.test_by_names,
00193                                    _opts_test_only_peripheral=opts.test_only_peripheral,
00194                                    _opts_test_only_common=opts.test_only_common,
00195                                    _opts_verbose_skipped_tests=opts.verbose_skipped_tests,
00196                                    _opts_verbose_test_result_only=opts.verbose_test_result_only,
00197                                    _opts_verbose=opts.verbose,
00198                                    _opts_firmware_global_name=opts.firmware_global_name,
00199                                    _opts_only_build_tests=opts.only_build_tests,
00200                                    _opts_suppress_summary=opts.suppress_summary,
00201                                    _opts_test_x_toolchain_summary=opts.test_x_toolchain_summary,
00202                                    _opts_copy_method=opts.copy_method
00203                                    )
00204 
00205     try:
00206         st_exec_thread = SingleTestExecutor(single_test)
00207     except KeyboardInterrupt, e:
00208         print "\n[CTRL+c] exit"
00209     st_exec_thread.start()
00210 
00211     if opts.rest_api_enabled:
00212         # Enable REST API
00213 
00214         app = Flask(__name__)
00215 
00216         @app.route('/')
00217         def hello_world():
00218             return 'Hello World!'
00219 
00220         @app.route('/status')
00221         def rest_api_status():
00222             return single_test.rest_api_status() # TODO
00223 
00224         @app.route('/config')
00225         def rest_api_config():
00226             return single_test.rest_api_config() # TODO
00227 
00228         @app.route('/log')
00229         def rest_api_log():
00230             return single_test.rest_api_log() # TODO
00231 
00232         @app.route('/request/<request_type>') # 'muts', 'test_spec', 'test_results'
00233         def rest_api_request_handler(request_type):
00234             result = single_test.rest_api_request_handler(request_type) # TODO
00235             return result
00236 
00237         rest_api_port = int(opts.rest_api_port_no) if opts.rest_api_port_no else 5555
00238         app.debug = False
00239         app.run(port=rest_api_port) # Blocking Flask REST API web service
00240     else:
00241         st_exec_thread.join()
00242 
00243 '''