1 # The MIT License (MIT)
3 # Copyright (c) 2017 Jérémie Galarneau <jeremie.galarneau@efficios.com>
4 # Copyright (c) 2017 Philippe Proulx <pproulx@efficios.com>
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:
13 # The above copyright notice and this permission notice shall be included in
14 # all copies or substantial portions of the Software.
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
28 import babeltrace
.reader_event_declaration
as event_declaration
30 class EventDeclarationTestCase(unittest
.TestCase
):
34 'ph_field_2' : 'bla bla',
35 'spc_field' : 'some string',
36 'seh_field' : 'another string',
42 self
._int
_ft
= bt2
.IntegerFieldType(32)
43 self
._str
_ft
= bt2
.StringFieldType()
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
),
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
),
58 self
._sc
.event_header_field_type
= bt2
.StructureFieldType()
59 self
._sc
.event_header_field_type
+= collections
.OrderedDict([
60 ('seh_field', self
._str
_ft
),
63 self
._sc
.event_context_field_type
= bt2
.StructureFieldType()
64 self
._sc
.event_context_field_type
+= collections
.OrderedDict([
65 ('sec_field', self
._int
_ft
),
68 self
._clock
_class
= bt2
.ClockClass('allo', 1000)
69 self
._trace
.add_clock_class(self
._clock
_class
)
71 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
),
77 self
._ec
.payload_field_type
= bt2
.StructureFieldType()
78 self
._ec
.payload_field_type
+= collections
.OrderedDict([
79 ('ef_field', self
._int
_ft
),
82 self
._sc
.add_event_class(self
._ec
)
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']
100 del self
._cc
_prio
_map
104 def _get_event_declaration(self
):
105 return event_declaration
._create
_event
_declaration
(self
._ec
)
108 declaration
= self
._get
_event
_declaration
()
109 self
.assertEqual(declaration
.name
, 'event_class_name')
112 declaration
= self
._get
_event
_declaration
()
113 self
.assertEqual(declaration
.id, 42)
115 def test_fields(self
):
116 declaration
= self
._get
_event
_declaration
()
117 fields
= declaration
.fields
118 self
.assertEqual(len(list(fields
)), len(self
._values
))
120 def test_fields_scope(self
):
121 declaration
= self
._get
_event
_declaration
()
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
)
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
)
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
)
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
)
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
)
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
)