tool/mbed/mbed-sdk/workspace_tools/test_webapi.py

   1 """
   2 mbed SDK
   3 Copyright (c) 2011-2014 ARM Limited
   4
   5 Licensed under the Apache License, Version 2.0 (the "License");
   6 you may not use this file except in compliance with the License.
   7 You may obtain a copy of the License at
   8
   9     http://www.apache.org/licenses/LICENSE-2.0
  10
  11 Unless required by applicable law or agreed to in writing, software
  12 distributed under the License is distributed on an "AS IS" BASIS,
  13 WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  14 See the License for the specific language governing permissions and
  15 limitations under the License.
  16
  17 Author: Przemyslaw Wirkus <Przemyslaw.wirkus@arm.com>
  18 """
  19
  20 import sys
  21 import json
  22 import optparse
  23 from flask import Flask
  24 from os.path import join, abspath, dirname
  25
  26 # Be sure that the tools directory is in the search path
  27 ROOT = abspath(join(dirname(__file__), ".."))
  28 sys.path.insert(0, ROOT)
  29
  30 # Imports related to mbed build api
  31 from workspace_tools.utils import construct_enum
  32 from workspace_tools.build_api import mcu_toolchain_matrix
  33
  34 # Imports from TEST API
  35 from test_api import SingleTestRunner
  36 from test_api import SingleTestExecutor
  37 from test_api import get_json_data_from_file
  38 from test_api import print_muts_configuration_from_json
  39 from test_api import print_test_configuration_from_json
  40 from test_api import get_avail_tests_summary_table
  41 from test_api import get_default_test_options_parser
  42
  43
  44 class SingleTestRunnerWebService(SingleTestRunner):
  45     def __init__(self):
  46         super(SingleTestRunnerWebService, self).__init__()
  47
  48         # With this lock we should control access to certain resources inside this class
  49         self.resource_lock = thread.allocate_lock()
  50
  51         self.RestRequest = construct_enum(REST_MUTS='muts',
  52                                           REST_TEST_SPEC='test_spec',
  53                                           REST_TEST_RESULTS='test_results')
  54
  55     def get_rest_result_template(self, result, command, success_code):
  56         """ Returns common part of every web service request
  57         """
  58         result = {"result" : result,
  59                   "command" : command,
  60                   "success_code": success_code} # 0 - OK, >0 - Error number
  61         return result
  62
  63     # REST API handlers for Flask framework
  64     def rest_api_status(self):
  65         """ Returns current test execution status. E.g. running / finished etc.
  66         """
  67         with self.resource_lock:
  68             pass
  69
  70     def rest_api_config(self):
  71         """ Returns configuration passed to SingleTest executor
  72         """
  73         with self.resource_lock:
  74             pass
  75
  76     def rest_api_log(self):
  77         """ Returns current test log """
  78         with self.resource_lock:
  79             pass
  80
  81     def rest_api_request_handler(self, request_type):
  82         """ Returns various data structures. Both static and mutable during test
  83         """
  84         result = {}
  85         success_code = 0
  86         with self.resource_lock:
  87             if request_type == self.RestRequest.REST_MUTS:
  88                 result = self.muts # Returns MUTs
  89             elif request_type == self.RestRequest.REST_TEST_SPEC:
  90                 result = self.test_spec # Returns Test Specification
  91             elif request_type == self.RestRequest.REST_TEST_RESULTS:
  92                 pass # Returns test results
  93             else:
  94                 success_code = -1
  95         return json.dumps(self.get_rest_result_template(result, 'request/' + request_type, success_code), indent=4)
  96
  97
  98 def singletest_in_webservice_mode():
  99     # TODO Implement this web service functionality
 100     pass
 101
 102
 103 def get_default_test_webservice_options_parser():
 104     """ Get test script web service options used by CLI, webservices etc.
 105     """
 106     parser = get_default_test_options_parser()
 107
 108     # Things related to web services offered by test suite scripts
 109     parser.add_option('', '--rest-api',
 110                       dest='rest_api_enabled',
 111                       default=False,
 112                       action="store_true",
 113                       help='Enables REST API.')
 114
 115     parser.add_option('', '--rest-api-port',
 116                       dest='rest_api_port_no',
 117                       help='Sets port for REST API interface')
 118
 119     return parser
 120
 121 '''
 122 if __name__ == '__main__':
 123     # Command line options
 124     parser = get_default_test_options_parser()
 125
 126     parser.description = """This script allows you to run mbed defined test cases for particular MCU(s) and corresponding toolchain(s)."""
 127     parser.epilog = """Example: singletest.py -i test_spec.json -M muts_all.json"""
 128
 129     (opts, args) = parser.parse_args()
 130
 131     # Print summary / information about automation test status
 132     if opts.test_automation_report:
 133         print get_avail_tests_summary_table()
 134         exit(0)
 135
 136     # Print summary / information about automation test status
 137     if opts.test_case_report:
 138         test_case_report_cols = ['id', 'automated', 'description', 'peripherals', 'host_test', 'duration', 'source_dir']
 139         print get_avail_tests_summary_table(cols=test_case_report_cols, result_summary=False, join_delim='\n')
 140         exit(0)
 141
 142     # Only prints matrix of supported toolchains
 143     if opts.supported_toolchains:
 144         print mcu_toolchain_matrix(platform_filter=opts.general_filter_regex)
 145         exit(0)
 146
 147     # Open file with test specification
 148     # test_spec_filename tells script which targets and their toolchain(s)
 149     # should be covered by the test scenario
 150     test_spec = get_json_data_from_file(opts.test_spec_filename) if opts.test_spec_filename else None
 151     if test_spec is None:
 152         if not opts.test_spec_filename:
 153             parser.print_help()
 154         exit(-1)
 155
 156     # Get extra MUTs if applicable
 157     MUTs = get_json_data_from_file(opts.muts_spec_filename) if opts.muts_spec_filename else None
 158
 159     if MUTs is None:
 160         if not opts.muts_spec_filename:
 161             parser.print_help()
 162         exit(-1)
 163
 164     # Only prints read MUTs configuration
 165     if MUTs and opts.verbose_test_configuration_only:
 166         print "MUTs configuration in %s:"% opts.muts_spec_filename
 167         print print_muts_configuration_from_json(MUTs)
 168         print
 169         print "Test specification in %s:"% opts.test_spec_filename
 170         print print_test_configuration_from_json(test_spec)
 171         exit(0)
 172
 173     # Verbose test specification and MUTs configuration
 174     if MUTs and opts.verbose:
 175         print print_muts_configuration_from_json(MUTs)
 176     if test_spec and opts.verbose:
 177         print print_test_configuration_from_json(test_spec)
 178
 179     if opts.only_build_tests:
 180         # We are skipping testing phase, and suppress summary
 181         opts.suppress_summary = True
 182
 183     single_test = SingleTestRunner(_global_loops_count=opts.test_global_loops_value,
 184                                    _test_loops_list=opts.test_loops_list,
 185                                    _muts=MUTs,
 186                                    _test_spec=test_spec,
 187                                    _opts_goanna_for_mbed_sdk=opts.goanna_for_mbed_sdk,
 188                                    _opts_goanna_for_tests=opts.goanna_for_tests,
 189                                    _opts_shuffle_test_order=opts.shuffle_test_order,
 190                                    _opts_shuffle_test_seed=opts.shuffle_test_seed,
 191                                    _opts_test_by_names=opts.test_by_names,
 192                                    _opts_test_only_peripheral=opts.test_only_peripheral,
 193                                    _opts_test_only_common=opts.test_only_common,
 194                                    _opts_verbose_skipped_tests=opts.verbose_skipped_tests,
 195                                    _opts_verbose_test_result_only=opts.verbose_test_result_only,
 196                                    _opts_verbose=opts.verbose,
 197                                    _opts_firmware_global_name=opts.firmware_global_name,
 198                                    _opts_only_build_tests=opts.only_build_tests,
 199                                    _opts_suppress_summary=opts.suppress_summary,
 200                                    _opts_test_x_toolchain_summary=opts.test_x_toolchain_summary,
 201                                    _opts_copy_method=opts.copy_method
 202                                    )
 203
 204     try:
 205         st_exec_thread = SingleTestExecutor(single_test)
 206     except KeyboardInterrupt, e:
 207         print "\n[CTRL+c] exit"
 208     st_exec_thread.start()
 209
 210     if opts.rest_api_enabled:
 211         # Enable REST API
 212
 213         app = Flask(__name__)
 214
 215         @app.route('/')
 216         def hello_world():
 217             return 'Hello World!'
 218
 219         @app.route('/status')
 220         def rest_api_status():
 221             return single_test.rest_api_status() # TODO
 222
 223         @app.route('/config')
 224         def rest_api_config():
 225             return single_test.rest_api_config() # TODO
 226
 227         @app.route('/log')
 228         def rest_api_log():
 229             return single_test.rest_api_log() # TODO
 230
 231         @app.route('/request/<request_type>') # 'muts', 'test_spec', 'test_results'
 232         def rest_api_request_handler(request_type):
 233             result = single_test.rest_api_request_handler(request_type) # TODO
 234             return result
 235
 236         rest_api_port = int(opts.rest_api_port_no) if opts.rest_api_port_no else 5555
 237         app.debug = False
 238         app.run(port=rest_api_port) # Blocking Flask REST API web service
 239     else:
 240         st_exec_thread.join()
 241
 242 '''