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:
6 # http://www.apache.org/licenses/LICENSE-2.0
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.
14 """Module for converting in-memory data into JSON output.
16 CSIT and VPP PAPI are using custom data types that are not directly serializable
19 Thus, before writing the output onto disk, the data is recursively converted to
20 equivalent serializable types, in extreme cases replaced by string
23 Validation is outside the scope of this module, as it should use the JSON data
30 from collections.abc import Iterable, Mapping, Set
31 from enum import IntFlag
32 from dateutil.parser import parse
35 def _pre_serialize_recursive(data):
36 """Recursively sort and convert to a more serializable form.
38 VPP PAPI code can give data with its own MACAddres type,
39 or various other enum and flag types.
40 The default json.JSONEncoder method raises TypeError on that.
41 First point of this function is to apply str() or repr()
42 to leaf values that need it.
44 Also, PAPI responses are namedtuples, which confuses
45 the json.JSONEncoder method (so it does not recurse).
46 Dictization (see PapiExecutor) helps somewhat, but it turns namedtuple
47 into a UserDict, which also confuses json.JSONEncoder.
48 Therefore, we recursively convert any Mapping into an ordinary dict.
50 We also convert iterables to list (sorted if the iterable was a set),
51 and prevent numbers from getting converted to strings.
53 As we are doing such low level operations,
54 we also convert mapping keys to strings
55 and sort the mapping items by keys alphabetically,
56 except "data" field moved to the end.
58 :param data: Object to make serializable, dictized when applicable.
60 :returns: Serializable equivalent of the argument.
62 :raises ValueError: If the argument does not support string conversion.
64 # Recursion ends at scalar values, first handle irregular ones.
65 if isinstance(data, IntFlag):
67 if isinstance(data, bytes):
69 # The regular ones are good to go.
70 if isinstance(data, (str, int, float, bool)):
72 # Recurse over, convert and sort mappings.
73 if isinstance(data, Mapping):
74 # Convert and sort alphabetically.
76 str(key): _pre_serialize_recursive(data[key])
77 for key in sorted(data.keys())
79 # If exists, move "data" field to the end.
81 data_value = ret.pop(u"data")
82 ret[u"data"] = data_value
83 # If exists, move "type" field at the start.
85 type_value = ret.pop(u"type")
87 ret = dict(type=type_value)
90 # Recurse over and convert iterables.
91 if isinstance(data, Iterable):
92 list_data = [_pre_serialize_recursive(item) for item in data]
93 # Additionally, sets are exported as sorted.
94 if isinstance(data, Set):
95 list_data = sorted(list_data)
97 # Unknown structure, attempt str().
101 def _pre_serialize_root(data):
102 """Recursively convert to a more serializable form, tweak order.
104 See _pre_serialize_recursive for most of changes this does.
106 The logic here (outside the recursive function) only affects
107 field ordering in the root mapping,
108 to make it more human friendly.
109 We are moving "version" to the top,
110 followed by start time and end time.
111 and various long fields to the bottom.
113 Some edits are done in-place, do not trust the argument value after calling.
115 :param data: Root data to make serializable, dictized when applicable.
117 :returns: Order-tweaked version of the argument.
119 :raises KeyError: If the data does not contain required fields.
120 :raises TypeError: If the argument is not a dict.
121 :raises ValueError: If the argument does not support string conversion.
123 if not isinstance(data, dict):
124 raise RuntimeError(f"Root data object needs to be a dict: {data!r}")
125 data = _pre_serialize_recursive(data)
126 new_data = dict(version=data.pop(u"version"))
127 new_data[u"start_time"] = data.pop(u"start_time")
128 new_data[u"end_time"] = data.pop(u"end_time")
129 new_data.update(data)
133 def _merge_into_suite_info_file(teardown_path):
134 """Move setup and teardown data into a singe file, remove old files.
136 The caller has to confirm the argument is correct, e.g. ending in
137 "/teardown.info.json".
139 :param teardown_path: Local filesystem path to teardown file.
140 :type teardown_path: str
141 :returns: Local filesystem path to newly created suite file.
144 # Manual right replace: https://stackoverflow.com/a/9943875
145 setup_path = u"setup".join(teardown_path.rsplit(u"teardown", 1))
146 with open(teardown_path, u"rt", encoding="utf-8") as file_in:
147 teardown_data = json.load(file_in)
148 # Transforming setup data into suite data.
149 with open(setup_path, u"rt", encoding="utf-8") as file_in:
150 suite_data = json.load(file_in)
152 end_time = teardown_data[u"end_time"]
153 suite_data[u"end_time"] = end_time
154 start_float = parse(suite_data[u"start_time"]).timestamp()
155 end_float = parse(suite_data[u"end_time"]).timestamp()
156 suite_data[u"duration"] = end_float - start_float
157 setup_telemetry = suite_data.pop(u"telemetry")
158 suite_data[u"setup_telemetry"] = setup_telemetry
159 suite_data[u"teardown_telemetry"] = teardown_data[u"telemetry"]
161 suite_path = u"suite".join(teardown_path.rsplit(u"teardown", 1))
162 with open(suite_path, u"wt", encoding="utf-8") as file_out:
163 json.dump(suite_data, file_out, indent=1)
164 # We moved everything useful from temporary setup/teardown info files.
165 os.remove(setup_path)
166 os.remove(teardown_path)
171 def write_output(file_path, data):
172 """Prepare data for serialization and dump into a file.
174 Ancestor directories are created if needed.
176 :param file_path: Local filesystem path, including the file name.
177 :param data: Root data to make serializable, dictized when applicable.
181 data = _pre_serialize_root(data)
183 # Lets move Telemetry to the end.
184 telemetry = data.pop(u"telemetry")
185 data[u"telemetry"] = telemetry
187 os.makedirs(os.path.dirname(file_path), exist_ok=True)
188 with open(file_path, u"wt", encoding="utf-8") as file_out:
189 json.dump(data, file_out, indent=1)
191 if file_path.endswith(u"/teardown.info.json"):
192 file_path = _merge_into_suite_info_file(file_path)