| 1 | # The MIT License (MIT) |
| 2 | # |
| 3 | # Copyright (c) 2017 Jérémie Galarneau <jeremie.galarneau@efficios.com> |
| 4 | # Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com> |
| 5 | # |
| 6 | # Permission is hereby granted, free of charge, to any person obtaining a copy |
| 7 | # of this software and associated documentation files (the "Software"), to deal |
| 8 | # in the Software without restriction, including without limitation the rights |
| 9 | # to use, copy, modify, merge, publish, distribute, sublicense, and/or sell |
| 10 | # copies of the Software, and to permit persons to whom the Software is |
| 11 | # furnished to do so, subject to the following conditions: |
| 12 | # |
| 13 | # The above copyright notice and this permission notice shall be included in |
| 14 | # all copies or substantial portions of the Software. |
| 15 | # |
| 16 | # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR |
| 17 | # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, |
| 18 | # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE |
| 19 | # AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER |
| 20 | # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, |
| 21 | # OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN |
| 22 | # THE SOFTWARE. |
| 23 | |
| 24 | import collections |
| 25 | import unittest |
| 26 | import bt2 |
| 27 | import babeltrace |
| 28 | import datetime |
| 29 | |
| 30 | |
| 31 | class EventTestCase(unittest.TestCase): |
| 32 | def setUp(self): |
| 33 | self._values = { |
| 34 | 'ph_field_1' : 42, |
| 35 | 'ph_field_2' : 'bla bla', |
| 36 | 'spc_field' : 'some string', |
| 37 | 'seh_field' : 'another string', |
| 38 | 'sec_field' : 68752, |
| 39 | 'ec_field' : 89, |
| 40 | 'ef_field' : 8476, |
| 41 | } |
| 42 | |
| 43 | self._int_ft = bt2.IntegerFieldType(32) |
| 44 | self._str_ft = bt2.StringFieldType() |
| 45 | |
| 46 | self._trace = bt2.Trace() |
| 47 | self._trace.packet_header_field_type = bt2.StructureFieldType() |
| 48 | self._trace.packet_header_field_type += collections.OrderedDict([ |
| 49 | ('ph_field_1', self._int_ft), |
| 50 | ('ph_field_2', self._str_ft), |
| 51 | ]) |
| 52 | |
| 53 | self._sc = bt2.StreamClass() |
| 54 | self._sc.packet_context_field_type = bt2.StructureFieldType() |
| 55 | self._sc.packet_context_field_type += collections.OrderedDict([ |
| 56 | ('spc_field', self._str_ft), |
| 57 | ]) |
| 58 | |
| 59 | self._sc.event_header_field_type = bt2.StructureFieldType() |
| 60 | self._sc.event_header_field_type += collections.OrderedDict([ |
| 61 | ('seh_field', self._str_ft), |
| 62 | ]) |
| 63 | |
| 64 | self._sc.event_context_field_type = bt2.StructureFieldType() |
| 65 | self._sc.event_context_field_type += collections.OrderedDict([ |
| 66 | ('sec_field', self._int_ft), |
| 67 | ]) |
| 68 | |
| 69 | self._clock_class = bt2.ClockClass('allo', 1000) |
| 70 | self._trace.add_clock_class(self._clock_class) |
| 71 | |
| 72 | self._ec = bt2.EventClass('event_class_name') |
| 73 | self._ec.context_field_type = bt2.StructureFieldType() |
| 74 | self._ec.context_field_type += collections.OrderedDict([ |
| 75 | ('ec_field', self._int_ft), |
| 76 | ]) |
| 77 | self._ec.payload_field_type = bt2.StructureFieldType() |
| 78 | self._ec.payload_field_type += collections.OrderedDict([ |
| 79 | ('ef_field', self._int_ft), |
| 80 | ]) |
| 81 | |
| 82 | self._sc.add_event_class(self._ec) |
| 83 | |
| 84 | self._trace.add_stream_class(self._sc) |
| 85 | self._cc_prio_map = bt2.ClockClassPriorityMap() |
| 86 | self._cc_prio_map[self._clock_class] = 231 |
| 87 | self._stream = self._sc() |
| 88 | self._packet = self._stream.create_packet() |
| 89 | self._packet.header_field['ph_field_1'] = self._values['ph_field_1'] |
| 90 | self._packet.header_field['ph_field_2'] = self._values['ph_field_2'] |
| 91 | self._packet.context_field['spc_field'] = self._values['spc_field'] |
| 92 | |
| 93 | self._event = self._ec() |
| 94 | self._event.clock_values.add(self._clock_class(1772)) |
| 95 | self._event.header_field['seh_field'] = self._values['seh_field'] |
| 96 | self._event.stream_event_context_field['sec_field'] = self._values[ |
| 97 | 'sec_field'] |
| 98 | self._event.context_field['ec_field'] = self._values['ec_field'] |
| 99 | self._event.payload_field['ef_field'] = self._values['ef_field'] |
| 100 | self._event.packet = self._packet |
| 101 | |
| 102 | def tearDown(self): |
| 103 | del self._trace |
| 104 | del self._sc |
| 105 | del self._ec |
| 106 | del self._int_ft |
| 107 | del self._str_ft |
| 108 | del self._clock_class |
| 109 | del self._cc_prio_map |
| 110 | del self._stream |
| 111 | del self._packet |
| 112 | del self._event |
| 113 | |
| 114 | def _get_event(self): |
| 115 | notif = bt2.EventNotification(self._event, self._cc_prio_map) |
| 116 | return babeltrace.reader_event._create_event(notif) |
| 117 | |
| 118 | def test_attr_name(self): |
| 119 | event = self._get_event() |
| 120 | self.assertEqual(event.name, 'event_class_name') |
| 121 | |
| 122 | def test_attr_cycles(self): |
| 123 | event = self._get_event() |
| 124 | self.assertEqual(event.cycles, 1772) |
| 125 | |
| 126 | def test_attr_timestamp(self): |
| 127 | event = self._get_event() |
| 128 | clock_class = self._cc_prio_map.highest_priority_clock_class |
| 129 | self.assertEqual(event.timestamp, 1772 * (1E9 / clock_class.frequency)) |
| 130 | |
| 131 | def test_attr_datetime(self): |
| 132 | event = self._get_event() |
| 133 | clock_class = self._cc_prio_map.highest_priority_clock_class |
| 134 | ns = self._event.clock_values[clock_class].ns_from_epoch |
| 135 | self.assertEqual(datetime.date.fromtimestamp(ns / 1E9), event.datetime) |
| 136 | |
| 137 | def test_getitem(self): |
| 138 | event = self._get_event() |
| 139 | for name, value in self._values.items(): |
| 140 | self.assertEqual(event[name], value) |
| 141 | |
| 142 | with self.assertRaises(KeyError): |
| 143 | field = event['non-existant-key'] |
| 144 | |
| 145 | def test_field_list_with_scope(self): |
| 146 | event = self._get_event() |
| 147 | self.assertEqual( |
| 148 | set(event.field_list_with_scope( |
| 149 | babeltrace.CTFScope.TRACE_PACKET_HEADER)), |
| 150 | set(['ph_field_1', 'ph_field_2'])) |
| 151 | |
| 152 | self.assertEqual( |
| 153 | set(event.field_list_with_scope( |
| 154 | babeltrace.CTFScope.STREAM_PACKET_CONTEXT)), |
| 155 | set(['spc_field'])) |
| 156 | |
| 157 | self.assertEqual( |
| 158 | set(event.field_list_with_scope( |
| 159 | babeltrace.CTFScope.STREAM_EVENT_HEADER)), |
| 160 | set(['seh_field'])) |
| 161 | |
| 162 | self.assertEqual( |
| 163 | set(event.field_list_with_scope( |
| 164 | babeltrace.CTFScope.STREAM_EVENT_CONTEXT)), |
| 165 | set(['sec_field'])) |
| 166 | |
| 167 | self.assertEqual( |
| 168 | set(event.field_list_with_scope( |
| 169 | babeltrace.CTFScope.EVENT_CONTEXT)), |
| 170 | set(['ec_field'])) |
| 171 | |
| 172 | self.assertEqual( |
| 173 | set(event.field_list_with_scope( |
| 174 | babeltrace.CTFScope.EVENT_FIELDS)), |
| 175 | set(['ef_field'])) |
| 176 | |
| 177 | def test_field_with_scope(self): |
| 178 | event = self._get_event() |
| 179 | self.assertEqual(event.field_with_scope( |
| 180 | 'seh_field', babeltrace.CTFScope.STREAM_EVENT_HEADER), |
| 181 | self._values['seh_field']) |
| 182 | |
| 183 | def test_get(self): |
| 184 | event = self._get_event() |
| 185 | self.assertEqual(event.get('spc_field'), self._values['spc_field']) |
| 186 | self.assertEqual(event.get('non-existant field', 'No field'), |
| 187 | 'No field') |
| 188 | |
| 189 | def test_keys(self): |
| 190 | event = self._get_event() |
| 191 | self.assertEqual(set(self._values.keys()), set(event.keys())) |
| 192 | |
| 193 | def test_len(self): |
| 194 | event = self._get_event() |
| 195 | self.assertEqual(len(self._values), len(event)) |