resources/libraries/python/model/parse.py

   1 # Copyright (c) 2023 Cisco and/or its affiliates.
   2 # Licensed under the Apache License, Version 2.0 (the "License");
   3 # you may not use this file except in compliance with the License.
   4 # You may obtain a copy of the License at:
   5 #
   6 #     http://www.apache.org/licenses/LICENSE-2.0
   7 #
   8 # Unless required by applicable law or agreed to in writing, software
   9 # distributed under the License is distributed on an "AS IS" BASIS,
  10 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  11 # See the License for the specific language governing permissions and
  12 # limitations under the License.
  13
  14 """Library for parsing results from JSON back to python objects.
  15
  16 This is useful for vpp-csit jobs like per-patch performance verify.
  17 Such jobs invoke robot multiple times, each time on a different build.
  18 Each robot invocation may execute several test cases.
  19 How exactly are the results compared depends on the job type,
  20 but extracting just the main results from jsons (file trees) is a common task,
  21 so it is placed into this library.
  22
  23 As such, the code in this file does not directly interact
  24 with the code in other files in this directory
  25 (result comparison is done outside robot invocation),
  26 but all files share common assumptions about json structure.
  27
  28 The function here expects a particular tree created on a filesystem by
  29 a bootstrap script, including test results
  30 exported as json files according to a current model schema.
  31 This script extracts the results (according to result type)
  32 and joins them mapping from test IDs to lists of floats.
  33 Also, the result is cached into a results.json file,
  34 so each tree is parsed only once.
  35
  36 The cached result does not depend on tree placement,
  37 so the bootstrap script may move and copy trees around
  38 before or after parsing.
  39 """
  40
  41 import json
  42 import os
  43 import pathlib
  44
  45 from typing import Dict, List
  46
  47
  48 def parse(dirpath: str, fake_value: float = 1.0) -> Dict[str, List[float]]:
  49     """Look for test jsons, extract scalar results.
  50
  51     Files other than .json are skipped, jsons without test_id are skipped.
  52     If the test failed, four fake values are used as a fake result.
  53
  54     Units are ignored, as both parent and current are tested
  55     with the same CSIT code so the unit should be identical.
  56
  57     The result is also cached as results.json file.
  58
  59     :param dirpath: Path to the directory tree to examine.
  60     :param fail_value: Fake value to use for test cases that failed.
  61     :type dirpath: str
  62     :type fail_falue: float
  63     :returns: Mapping from test IDs to list of measured values.
  64     :rtype: Dict[str, List[float]]
  65     :raises RuntimeError: On duplicate test ID or unknown test type.
  66     """
  67     if not pathlib.Path(dirpath).is_dir():
  68         # This happens when per-patch runs out of iterations.
  69         return {}
  70     resultpath = pathlib.Path(f"{dirpath}/results.json")
  71     if resultpath.is_file():
  72         with open(resultpath, "rt", encoding="utf8") as file_in:
  73             return json.load(file_in)
  74     results = {}
  75     for root, _, files in os.walk(dirpath):
  76         for filename in files:
  77             if not filename.endswith(".json"):
  78                 continue
  79             filepath = os.path.join(root, filename)
  80             with open(filepath, "rt", encoding="utf8") as file_in:
  81                 data = json.load(file_in)
  82             if "test_id" not in data:
  83                 continue
  84             name = data["test_id"]
  85             if name in results:
  86                 raise RuntimeError(f"Duplicate: {name}")
  87             if not data["passed"]:
  88                 results[name] = [fake_value] * 4
  89                 continue
  90             result_object = data["result"]
  91             result_type = result_object["type"]
  92             if result_type == "mrr":
  93                 results[name] = result_object["receive_rate"]["rate"]["values"]
  94             elif result_type == "ndrpdr":
  95                 results[name] = [result_object["pdr"]["lower"]["rate"]["value"]]
  96             elif result_type == "soak":
  97                 results[name] = [
  98                     result_object["critical_rate"]["lower"]["rate"]["value"]
  99                 ]
 100             elif result_type == "reconf":
 101                 results[name] = [result_object["loss"]["time"]["value"]]
 102             elif result_type == "hoststack":
 103                 results[name] = [result_object["bandwidth"]["value"]]
 104             else:
 105                 raise RuntimeError(f"Unknown result type: {result_type}")
 106     with open(resultpath, "wt", encoding="utf8") as file_out:
 107         json.dump(results, file_out, indent=1, separators=(", ", ": "))
 108     return results