tests/bindings/python/babeltrace/test_reader_event_declaration.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 babeltrace.reader_event_declaration as event_declaration
  29
  30 class EventDeclarationTestCase(unittest.TestCase):
  31     def setUp(self):
  32         self._values = {
  33             'ph_field_1' : 42,
  34             'ph_field_2' : 'bla bla',
  35             'spc_field' : 'some string',
  36             'seh_field' : 'another string',
  37             'sec_field' : 68752,
  38             'ec_field' : 89,
  39             'ef_field' : 8476,
  40         }
  41
  42         self._int_ft = bt2.IntegerFieldType(32)
  43         self._str_ft = bt2.StringFieldType()
  44
  45         self._trace = bt2.Trace()
  46         self._trace.packet_header_field_type = bt2.StructureFieldType()
  47         self._trace.packet_header_field_type += collections.OrderedDict([
  48             ('ph_field_1', self._int_ft),
  49             ('ph_field_2', self._str_ft),
  50         ])
  51
  52         self._sc = bt2.StreamClass()
  53         self._sc.packet_context_field_type = bt2.StructureFieldType()
  54         self._sc.packet_context_field_type += collections.OrderedDict([
  55             ('spc_field', self._str_ft),
  56         ])
  57
  58         self._sc.event_header_field_type = bt2.StructureFieldType()
  59         self._sc.event_header_field_type += collections.OrderedDict([
  60             ('seh_field', self._str_ft),
  61         ])
  62
  63         self._sc.event_context_field_type = bt2.StructureFieldType()
  64         self._sc.event_context_field_type += collections.OrderedDict([
  65             ('sec_field', self._int_ft),
  66         ])
  67
  68         self._clock_class = bt2.ClockClass('allo', 1000)
  69         self._trace.add_clock_class(self._clock_class)
  70
  71         self._ec = bt2.EventClass('event_class_name')
  72         self._ec.id = 42
  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     def tearDown(self):
  94         del self._trace
  95         del self._sc
  96         del self._ec
  97         del self._int_ft
  98         del self._str_ft
  99         del self._clock_class
 100         del self._cc_prio_map
 101         del self._stream
 102         del self._packet
 103
 104     def _get_event_declaration(self):
 105         return event_declaration._create_event_declaration(self._ec)
 106
 107     def test_name(self):
 108         declaration = self._get_event_declaration()
 109         self.assertEqual(declaration.name, 'event_class_name')
 110
 111     def test_id(self):
 112         declaration = self._get_event_declaration()
 113         self.assertEqual(declaration.id, 42)
 114
 115     def test_fields(self):
 116         declaration = self._get_event_declaration()
 117         fields = declaration.fields
 118         self.assertEqual(len(list(fields)), len(self._values))
 119
 120     def test_fields_scope(self):
 121         declaration = self._get_event_declaration()
 122         event_fields = list(
 123             declaration.fields_scope(babeltrace.CTFScope.EVENT_FIELDS))
 124         self.assertEqual(len(event_fields), 1)
 125         self.assertEqual(event_fields[0].name, 'ef_field')
 126         self.assertEqual(event_fields[0].scope,
 127                          babeltrace.CTFScope.EVENT_FIELDS)
 128
 129         event_ctx_fields = list(
 130             declaration.fields_scope(babeltrace.CTFScope.EVENT_CONTEXT))
 131         self.assertEqual(len(event_ctx_fields), 1)
 132         self.assertEqual(event_ctx_fields[0].name, 'ec_field')
 133         self.assertEqual(event_ctx_fields[0].scope,
 134                          babeltrace.CTFScope.EVENT_CONTEXT)
 135
 136         stream_ectx_fields = list(
 137             declaration.fields_scope(babeltrace.CTFScope.STREAM_EVENT_CONTEXT))
 138         self.assertEqual(len(stream_ectx_fields), 1)
 139         self.assertEqual(stream_ectx_fields[0].name, 'sec_field')
 140         self.assertEqual(stream_ectx_fields[0].scope,
 141                          babeltrace.CTFScope.STREAM_EVENT_CONTEXT)
 142
 143         stream_eh_fields = list(
 144             declaration.fields_scope(babeltrace.CTFScope.STREAM_EVENT_HEADER))
 145         self.assertEqual(len(stream_eh_fields), 1)
 146         self.assertEqual(stream_eh_fields[0].name, 'seh_field')
 147         self.assertEqual(stream_eh_fields[0].scope,
 148                          babeltrace.CTFScope.STREAM_EVENT_HEADER)
 149
 150         stream_pctx_fields = list(
 151             declaration.fields_scope(babeltrace.CTFScope.STREAM_PACKET_CONTEXT))
 152         self.assertEqual(len(stream_pctx_fields), 1)
 153         self.assertEqual(stream_pctx_fields[0].name, 'spc_field')
 154         self.assertEqual(stream_pctx_fields[0].scope,
 155                          babeltrace.CTFScope.STREAM_PACKET_CONTEXT)
 156
 157         stream_ph_fields = list(
 158             declaration.fields_scope(babeltrace.CTFScope.TRACE_PACKET_HEADER))
 159         self.assertEqual(len(stream_ph_fields), 2)
 160         self.assertEqual(stream_ph_fields[0].name, 'ph_field_1')
 161         self.assertEqual(stream_ph_fields[0].scope,
 162                          babeltrace.CTFScope.TRACE_PACKET_HEADER)
 163         self.assertEqual(stream_ph_fields[1].name, 'ph_field_2')
 164         self.assertEqual(stream_ph_fields[1].scope,
 165                          babeltrace.CTFScope.TRACE_PACKET_HEADER)