Mercurial > repos > shellac > guppy_basecaller
comparison env/lib/python3.7/site-packages/planemo/test/results.py @ 0:26e78fe6e8c4 draft
"planemo upload commit c699937486c35866861690329de38ec1a5d9f783"
| author | shellac |
|---|---|
| date | Sat, 02 May 2020 07:14:21 -0400 |
| parents | |
| children |
comparison
equal
deleted
inserted
replaced
| -1:000000000000 | 0:26e78fe6e8c4 |
|---|---|
| 1 """Describes results. | |
| 2 | |
| 3 Is a JSON. | |
| 4 """ | |
| 5 import json | |
| 6 import os | |
| 7 | |
| 8 from planemo.io import error | |
| 9 | |
| 10 | |
| 11 class StructuredData(object): | |
| 12 """Abstraction around a simple data structure describing test results.""" | |
| 13 | |
| 14 def __init__(self, json_path=None, data=None): | |
| 15 """Create a :class:`StructuredData` from a JSON file.""" | |
| 16 def data_error(): | |
| 17 error("An invalid JSON for structured test result data - " | |
| 18 "summary information and planemo reports will be " | |
| 19 "incorrect.") | |
| 20 | |
| 21 self.json_path = json_path | |
| 22 structured_data = {} | |
| 23 structured_data_tests = {} | |
| 24 if json_path and os.path.exists(json_path) and data is None: | |
| 25 try: | |
| 26 with open(json_path, "r") as output_json_f: | |
| 27 data = json.load(output_json_f) | |
| 28 except Exception: | |
| 29 data_error() | |
| 30 try: | |
| 31 structured_data = data | |
| 32 structured_data_tests = structured_data["tests"] | |
| 33 except Exception: | |
| 34 data_error() | |
| 35 | |
| 36 self.structured_data = structured_data | |
| 37 self.structured_data_tests = structured_data_tests | |
| 38 structured_data_by_id = {} | |
| 39 for test in self.structured_data_tests: | |
| 40 structured_data_by_id[test["id"]] = test["data"] | |
| 41 self.structured_data_by_id = structured_data_by_id | |
| 42 self.has_details = "summary" in structured_data | |
| 43 if self.has_details: | |
| 44 self.read_summary() | |
| 45 | |
| 46 def update(self): | |
| 47 """Write out an updated version of this data structure to supplied json path.""" | |
| 48 with open(self.json_path, "w") as out_f: | |
| 49 json.dump(self.structured_data, out_f) | |
| 50 | |
| 51 def set_exit_code(self, exit_code): | |
| 52 """Set the exit_code for the this test.""" | |
| 53 self.structured_data["exit_code"] = exit_code | |
| 54 | |
| 55 def calculate_summary_data_if_needed(self): | |
| 56 if "summary" not in self.structured_data: | |
| 57 self.calculate_summary_data() | |
| 58 | |
| 59 def calculate_summary_data(self): | |
| 60 """Use full details on individual test data to update structured data with summary info.""" | |
| 61 num_tests = 0 | |
| 62 num_failures = 0 | |
| 63 num_skips = 0 | |
| 64 num_errors = 0 | |
| 65 | |
| 66 for test in self.structured_data_tests: | |
| 67 test_data = get_dict_value("data", test) | |
| 68 status = get_dict_value("status", test_data) | |
| 69 num_tests += 1 | |
| 70 if status == "skip": | |
| 71 num_skips += 1 | |
| 72 elif status == "failure": | |
| 73 num_failures += 1 | |
| 74 elif status == "error": | |
| 75 num_errors += 1 | |
| 76 elif status != "success": | |
| 77 raise Exception("Unknown test status encountered [%s]" % status) | |
| 78 | |
| 79 summary = {} | |
| 80 summary["num_tests"] = num_tests | |
| 81 summary["num_failures"] = num_failures | |
| 82 summary["num_skips"] = num_skips | |
| 83 summary["num_errors"] = num_errors | |
| 84 self.structured_data["summary"] = summary | |
| 85 self.read_summary() | |
| 86 | |
| 87 def read_summary(self): | |
| 88 """Read summary data into properties on this class.""" | |
| 89 summary = self.structured_data["summary"] | |
| 90 num_tests = summary["num_tests"] | |
| 91 num_failures = summary["num_failures"] | |
| 92 num_skips = summary["num_skips"] | |
| 93 num_errors = summary["num_errors"] | |
| 94 | |
| 95 self.num_tests = num_tests | |
| 96 self.num_problems = num_skips + num_errors + num_failures | |
| 97 | |
| 98 self.exit_code = self.structured_data.get("exit_code", None) | |
| 99 | |
| 100 @property | |
| 101 def failed_ids(self): | |
| 102 """Find set of IDs for failed tests.""" | |
| 103 ids = set([]) | |
| 104 for test_data in self.structured_data_tests: | |
| 105 if test_data["data"]["status"] == "success": | |
| 106 continue | |
| 107 test_case = test_data["id"].replace(".test_toolbox.", ".test_toolbox:") | |
| 108 ids.add(test_case) | |
| 109 return ids | |
| 110 | |
| 111 | |
| 112 def get_dict_value(key, data): | |
| 113 """Return data[key] with improved KeyError.""" | |
| 114 try: | |
| 115 return data[key] | |
| 116 except (KeyError, TypeError): | |
| 117 raise KeyError("No key [%s] in [%s]" % (key, data)) | |
| 118 | |
| 119 | |
| 120 __all__ = ( | |
| 121 "StructuredData", | |
| 122 "get_dict_value", | |
| 123 ) |
