begriffs open source - cmsis-freertos/blob - Test/litani/lib/job_outcome.py

   1 # Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved.
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License").
   4 # You may not use this file except in compliance with the License.
   5 # A copy of the License is located at
   6 #
   7 #     http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 # or in the "license" file accompanying this file. This file is distributed
  10 # on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either
  11 # express or implied. See the License for the specific language governing
  12 # permissions and limitations under the License.
  13
  14
  15 import dataclasses
  16 import json
  17 import logging
  18 import os
  19
  20
  21 ################################################################################
  22 # Decider classes
  23 # ``````````````````````````````````````````````````````````````````````````````
  24 # Each of these classes has get_job_fields method.  The class evaluates what
  25 # the result of a single Litani job is and returns that result in the dict.
  26 ################################################################################
  27
  28
  29
  30 @dataclasses.dataclass
  31 class OutcomeTableDecider:
  32     """Decide what the result of a job is based on an outcome table.
  33
  34     An 'outcome table' is a mapping from 'outcomes'---like return codes,
  35     timeouts, or a wildcard---to whether or not the job is successful. This
  36     class takes a user-specified or default outcome table, and decides what the
  37     result of a single Litani job was by iterating through the table.
  38     """
  39
  40     table: dict
  41     return_code: int
  42     timeout_reached: bool
  43     loaded_from_file: bool
  44
  45
  46     def _get_wildcard_outcome(self):
  47         for outcome in self.table["outcomes"]:
  48             if outcome["type"] == "wildcard":
  49                 return outcome["action"]
  50         raise UserWarning(
  51             "Outcome table contains no wildcard rule: %s" % json.dumps(
  52                 self.table, indent=2))
  53
  54
  55     def _get_timeout_outcome(self):
  56         for outcome in self.table["outcomes"]:
  57             if outcome["type"] == "timeout":
  58                 return outcome["action"]
  59         return None
  60
  61
  62     def _get_return_code_outcome(self, return_code):
  63         for outcome in self.table["outcomes"]:
  64             if outcome["type"] == "return-code" and \
  65                     outcome["value"] == return_code:
  66                 return outcome["action"]
  67         return None
  68
  69
  70     def get_job_fields(self):
  71         return {
  72             "outcome": self.get_outcome(),
  73             "loaded_outcome_dict":
  74                 self.table if self.loaded_from_file else None
  75         }
  76
  77
  78     def get_outcome(self):
  79         timeout_outcome = self._get_timeout_outcome()
  80         if self.timeout_reached:
  81             if timeout_outcome:
  82                 return timeout_outcome
  83             return self._get_wildcard_outcome()
  84
  85         rc_outcome = self._get_return_code_outcome(self.return_code)
  86         if rc_outcome:
  87             return rc_outcome
  88
  89         return self._get_wildcard_outcome()
  90
  91
  92
  93 ################################################################################
  94 # Utilities
  95 ################################################################################
  96
  97
  98 def _get_default_outcome_dict(args):
  99     """Litani's default behavior if the user does not specify an outcome table.
 100
 101     This is not a constant dict as it also depends on whether the user passed in
 102     command-line flags that affect how the result is decided, like
 103     --ignore-returns etc.
 104     """
 105
 106     outcomes = []
 107     if args.timeout_ok:
 108         outcomes.append({
 109             "type": "timeout",
 110             "action": "success",
 111         })
 112     elif args.timeout_ignore:
 113         outcomes.append({
 114             "type": "timeout",
 115             "action": "fail_ignored",
 116         })
 117
 118     if args.ok_returns:
 119         for rc in args.ok_returns:
 120             outcomes.append({
 121                 "type": "return-code",
 122                 "value": int(rc),
 123                 "action": "success",
 124             })
 125
 126     if args.ignore_returns:
 127         for rc in args.ignore_returns:
 128             outcomes.append({
 129                 "type": "return-code",
 130                 "value": int(rc),
 131                 "action": "fail_ignored",
 132             })
 133
 134     outcomes.extend([{
 135         "type": "return-code",
 136         "value": 0,
 137         "action": "success",
 138     }, {
 139         "type": "wildcard",
 140         "action": "fail",
 141     }])
 142
 143     return {"outcomes": outcomes}
 144
 145
 146 def validate_outcome_table(table):
 147     try:
 148         import voluptuous
 149     except ImportError:
 150         logging.debug("Skipping outcome table validation as voluptuous is not installed")
 151         return
 152
 153     actions = voluptuous.Any("success", "fail", "fail_ignored")
 154     schema = voluptuous.Schema({
 155         # A description of the outcome table as a whole.
 156         voluptuous.Optional("comment"): str,
 157
 158         # We use the first item in this list that matches the job
 159         "outcomes": [voluptuous.Any({
 160             "type": "return-code",
 161             "value": int,
 162             "action": actions,
 163             voluptuous.Optional("comment"): str,
 164         }, {
 165             "type": "timeout",
 166             "action": actions,
 167             voluptuous.Optional("comment"): str,
 168         }, {
 169             "type": "wildcard",
 170             "action": actions,
 171             voluptuous.Optional("comment"): str,
 172         })]
 173     }, required=True)
 174     voluptuous.humanize.validate_with_humanized_errors(table, schema)
 175
 176
 177 def _get_outcome_table_job_decider(args, return_code, timeout_reached):
 178     if args.outcome_table:
 179         _, ext = os.path.splitext(args.outcome_table)
 180         with open(args.outcome_table) as handle:
 181             if ext == ".json":
 182                 outcome_table = json.load(handle)
 183             elif ext == ".yaml":
 184                 import yaml
 185                 outcome_table = yaml.safe_load(handle)
 186             else:
 187                 raise UserWarning("Unsupported outcome table format (%s)" % ext)
 188         loaded_from_file = True
 189     else:
 190         loaded_from_file = False
 191         outcome_table = _get_default_outcome_dict(args)
 192
 193     logging.debug("Using outcome table: %s", json.dumps(outcome_table, indent=2))
 194     validate_outcome_table(outcome_table)
 195
 196     return OutcomeTableDecider(
 197         outcome_table, return_code, timeout_reached,
 198         loaded_from_file=loaded_from_file)
 199
 200
 201 ################################################################################
 202 # Entry point
 203 ################################################################################
 204
 205
 206 def fill_in_result(runner, job_data, args):
 207     """Add fields pertaining to job result to job_data dict
 208
 209     The 'result' of a job can be evaluated in several ways. The most simple
 210     mechanism, where a return code of 0 means success and anything else is a
 211     failure, is encoded by the "default outcome table". Users can also supply
 212     their own outcome table as a JSON file, and other mechanisms could be
 213     available in the future.
 214
 215     Depending on how we are to evaluate the result, we construct an instance of
 216     one of the Decider classes in this module, and use the Decider to evaluate
 217     the result of the job. The result is a dict, whose keys and values we add to
 218     the job's dict.
 219     """
 220
 221     job_data["complete"] = True
 222     job_data["timeout_reached"] = runner.reached_timeout()
 223     job_data["command_return_code"] = runner.get_return_code()
 224     job_data["memory_trace"] = runner.get_memory_trace()
 225
 226     # These get set by the deciders
 227     job_data["loaded_outcome_dict"] = None
 228
 229     decider = _get_outcome_table_job_decider(
 230         args, runner.get_return_code(), runner.reached_timeout())
 231
 232     fields = decider.get_job_fields()
 233     for k, v in fields.items():
 234         job_data[k] = v
 235     job_data["wrapper_return_code"] = 1 if job_data["outcome"] == "fail" else 0