resources/libraries/python/model/parse.py

   1 # Copyright (c) 2024 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 test results are sorted by test_id,
  58     as the filesystem order is not deterministic enough.
  59
  60     The result is also cached as results.json file.
  61
  62     :param dirpath: Path to the directory tree to examine.
  63     :param fail_value: Fake value to use for test cases that failed.
  64     :type dirpath: str
  65     :type fail_falue: float
  66     :returns: Mapping from test IDs to list of measured values.
  67     :rtype: Dict[str, List[float]]
  68     :raises RuntimeError: On duplicate test ID or unknown test type.
  69     """
  70     if not pathlib.Path(dirpath).is_dir():
  71         # This happens when per-patch runs out of iterations.
  72         return {}
  73     resultpath = pathlib.Path(f"{dirpath}/results.json")
  74     if resultpath.is_file():
  75         with open(resultpath, "rt", encoding="utf8") as file_in:
  76             return json.load(file_in)
  77     results = {}
  78     for root, _, files in os.walk(dirpath):
  79         for filename in files:
  80             if not filename.endswith(".json"):
  81                 continue
  82             filepath = os.path.join(root, filename)
  83             with open(filepath, "rt", encoding="utf8") as file_in:
  84                 data = json.load(file_in)
  85             if "test_id" not in data:
  86                 continue
  87             name = data["test_id"]
  88             if name in results:
  89                 raise RuntimeError(f"Duplicate: {name}")
  90             if not data["passed"]:
  91                 results[name] = [fake_value] * 4
  92                 continue
  93             result_object = data["result"]
  94             result_type = result_object["type"]
  95             if result_type == "mrr":
  96                 results[name] = result_object["receive_rate"]["rate"]["values"]
  97             elif result_type == "ndrpdr":
  98                 results[name] = [result_object["pdr"]["lower"]["rate"]["value"]]
  99             elif result_type == "soak":
 100                 results[name] = [
 101                     result_object["critical_rate"]["lower"]["rate"]["value"]
 102                 ]
 103             elif result_type == "reconf":
 104                 results[name] = [result_object["loss"]["time"]["value"]]
 105             elif result_type == "hoststack":
 106                 results[name] = [result_object["bandwidth"]["value"]]
 107             else:
 108                 raise RuntimeError(f"Unknown result type: {result_type}")
 109     results = {test_id: results[test_id] for test_id in sorted(results)}
 110     with open(resultpath, "wt", encoding="utf8") as file_out:
 111         json.dump(results, file_out, indent=1, separators=(", ", ": "))
 112     return results