1#!/usr/bin/env python3 2# Copyright 2020 The Pigweed Authors 3# 4# Licensed under the Apache License, Version 2.0 (the "License"); you may not 5# use this file except in compliance with the License. You may obtain a copy of 6# the License at 7# 8# https://www.apache.org/licenses/LICENSE-2.0 9# 10# Unless required by applicable law or agreed to in writing, software 11# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT 12# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the 13# License for the specific language governing permissions and limitations under 14# the License. 15"""Tests the trace module.""" 16 17import json 18import struct 19import unittest 20 21from pw_trace import trace 22 23test_events = [ 24 trace.TraceEvent(trace.TraceType.INSTANTANEOUS, "m1", "L1", 1), 25 trace.TraceEvent(trace.TraceType.INSTANTANEOUS_GROUP, "m2", "L2", 2, "G2"), 26 trace.TraceEvent(trace.TraceType.ASYNC_STEP, "m3", "L3", 3, "G3", 103), 27 trace.TraceEvent(trace.TraceType.DURATION_START, "m4", "L4", 4), 28 trace.TraceEvent(trace.TraceType.DURATION_GROUP_START, "m5", "L5", 5, 29 "G5"), 30 trace.TraceEvent(trace.TraceType.ASYNC_START, "m6", "L6", 6, "G6", 106), 31 trace.TraceEvent(trace.TraceType.DURATION_END, "m7", "L7", 7), 32 trace.TraceEvent(trace.TraceType.DURATION_GROUP_END, "m8", "L8", 8, "G8"), 33 trace.TraceEvent(trace.TraceType.ASYNC_END, "m9", "L9", 9, "G9", 109) 34] 35 36test_json = [ 37 {"ph": "I", "pid": "m1", "name": "L1", "ts": 1, "s": "p"}, 38 {"ph": "I", "pid": "m2", "tid": "G2", "name": "L2", "ts": 2, "s": "t"}, 39 {"ph": "n", "pid": "m3", "tid": "G3", "name": "L3", "ts": 3, \ 40 "scope": "G3", "cat": "m3", "id": 103, "args": {"id": 103}}, 41 {"ph": "B", "pid": "m4", "tid": "L4", "name": "L4", "ts": 4}, 42 {"ph": "B", "pid": "m5", "tid": "G5", "name": "L5", "ts": 5}, 43 {"ph": "b", "pid": "m6", "tid": "G6", "name": "L6", "ts": 6, \ 44 "scope": "G6", "cat": "m6", "id": 106, "args": {"id": 106}}, 45 {"ph": "E", "pid": "m7", "tid": "L7", "name": "L7", "ts": 7}, 46 {"ph": "E", "pid": "m8", "tid": "G8", "name": "L8", "ts": 8}, 47 {"ph": "e", "pid": "m9", "tid": "G9", "name": "L9", "ts": 9, \ 48 "scope": "G9", "cat": "m9", "id": 109, "args": {"id": 109}}, 49] 50 51 52class TestTraceGenerateJson(unittest.TestCase): 53 """Tests generate json with various events.""" 54 def test_generate_single_json_event(self): 55 event = trace.TraceEvent(event_type=trace.TraceType.INSTANTANEOUS, 56 module="module", 57 label="label", 58 timestamp_us=10) 59 json_lines = trace.generate_trace_json([event]) 60 self.assertEqual(1, len(json_lines)) 61 self.assertEqual(json.loads(json_lines[0]), { 62 "ph": "I", 63 "pid": "module", 64 "name": "label", 65 "ts": 10, 66 "s": "p" 67 }) 68 69 def test_generate_multiple_json_events(self): 70 json_lines = trace.generate_trace_json(test_events) 71 self.assertEqual(len(test_json), len(json_lines)) 72 for actual, expected in zip(json_lines, test_json): 73 self.assertEqual(expected, json.loads(actual)) 74 75 def test_generate_json_data_arg_label(self): 76 event = trace.TraceEvent( 77 event_type=trace.TraceType.INSTANTANEOUS, 78 module="module", 79 label="", # Is replaced by data string 80 timestamp_us=10, 81 has_data=True, 82 data_fmt="@pw_arg_label", 83 data=bytes("arg", "utf-8")) 84 json_lines = trace.generate_trace_json([event]) 85 self.assertEqual(1, len(json_lines)) 86 self.assertEqual(json.loads(json_lines[0]), { 87 "ph": "I", 88 "pid": "module", 89 "name": "arg", 90 "ts": 10, 91 "s": "p" 92 }) 93 94 def test_generate_json_data_arg_group(self): 95 event = trace.TraceEvent( 96 event_type=trace.TraceType.INSTANTANEOUS_GROUP, 97 module="module", 98 label="label", 99 timestamp_us=10, 100 has_data=True, 101 data_fmt="@pw_arg_group", 102 data=bytes("arg", "utf-8")) 103 json_lines = trace.generate_trace_json([event]) 104 self.assertEqual(1, len(json_lines)) 105 self.assertEqual( 106 json.loads(json_lines[0]), { 107 "ph": "I", 108 "pid": "module", 109 "name": "label", 110 "tid": "arg", 111 "ts": 10, 112 "s": "t" 113 }) 114 115 def test_generate_json_data_counter(self): 116 event = trace.TraceEvent(event_type=trace.TraceType.INSTANTANEOUS, 117 module="module", 118 label="counter", 119 timestamp_us=10, 120 has_data=True, 121 data_fmt="@pw_arg_counter", 122 data=(5).to_bytes(4, byteorder="little")) 123 json_lines = trace.generate_trace_json([event]) 124 self.assertEqual(1, len(json_lines)) 125 self.assertEqual( 126 json.loads(json_lines[0]), { 127 "ph": "C", 128 "pid": "module", 129 "name": "counter", 130 "ts": 10, 131 "s": "p", 132 "args": { 133 "counter": 5 134 } 135 }) 136 137 def test_generate_json_data_struct_fmt_single(self): 138 event = trace.TraceEvent(event_type=trace.TraceType.INSTANTANEOUS, 139 module="module", 140 label="counter", 141 timestamp_us=10, 142 has_data=True, 143 data_fmt="@pw_py_struct_fmt:H", 144 data=(5).to_bytes(2, byteorder="little")) 145 json_lines = trace.generate_trace_json([event]) 146 self.assertEqual(1, len(json_lines)) 147 self.assertEqual( 148 json.loads(json_lines[0]), { 149 "ph": "I", 150 "pid": "module", 151 "name": "counter", 152 "ts": 10, 153 "s": "p", 154 "args": { 155 "data_0": 5 156 } 157 }) 158 159 def test_generate_json_data_struct_fmt_multi(self): 160 event = trace.TraceEvent(event_type=trace.TraceType.INSTANTANEOUS, 161 module="module", 162 label="counter", 163 timestamp_us=10, 164 has_data=True, 165 data_fmt="@pw_py_struct_fmt:Hl", 166 data=struct.pack("Hl", 5, 2)) 167 json_lines = trace.generate_trace_json([event]) 168 self.assertEqual(1, len(json_lines)) 169 self.assertEqual( 170 json.loads(json_lines[0]), { 171 "ph": "I", 172 "pid": "module", 173 "name": "counter", 174 "ts": 10, 175 "s": "p", 176 "args": { 177 "data_0": 5, 178 "data_1": 2 179 } 180 }) 181 182 183if __name__ == '__main__': 184 unittest.main() 185