tests/bindings/python/babeltrace/test_reader_event.py

   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.add_clock_value(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_value(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))