2 # Copyright (C) 2019 EfficiOS Inc.
4 # This program is free software; you can redistribute it and/or
5 # modify it under the terms of the GNU General Public License
6 # as published by the Free Software Foundation; only version 2
9 # This program is distributed in the hope that it will be useful,
10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12 # GNU General Public License for more details.
14 # You should have received a copy of the GNU General Public License
15 # along with this program; if not, write to the Free Software
16 # Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
21 from utils
import get_default_trace_class
22 from bt2
import stream_class
as bt2_stream_class
23 from bt2
import event_class
as bt2_event_class
24 from bt2
import field_class
as bt2_field_class
25 from bt2
import value
as bt2_value
26 from utils
import TestOutputPortMessageIterator
29 def _create_const_event_class(tc
, stream_class
):
30 fc1
= tc
.create_structure_field_class()
31 fc2
= tc
.create_structure_field_class()
32 event_class
= stream_class
.create_event_class(
33 payload_field_class
=fc1
, specific_context_field_class
=fc2
36 class MyIter(bt2
._UserMessageIterator
):
37 def __init__(self
, config
, self_port_output
):
40 stream
= trace
.create_stream(stream_class
)
42 self
._create
_stream
_beginning
_message
(stream
),
43 self
._create
_event
_message
(event_class
, stream
),
47 if len(self
._msgs
) == 0:
50 return self
._msgs
.pop(0)
52 class MySrc(bt2
._UserSourceComponent
, message_iterator_class
=MyIter
):
53 def __init__(self
, config
, params
, obj
):
54 self
._add
_output
_port
('out', params
)
57 src_comp
= graph
.add_component(MySrc
, 'my_source', None)
58 msg_iter
= TestOutputPortMessageIterator(graph
, src_comp
.output_ports
['out'])
60 # Ignore first message, stream beginning
63 event_msg
= next(msg_iter
)
65 return event_msg
.event
.cls
68 class EventClassTestCase(unittest
.TestCase
):
70 self
._tc
= get_default_trace_class()
72 self
._context
_fc
= self
._tc
.create_structure_field_class()
73 self
._context
_fc
.append_member('allo', self
._tc
.create_string_field_class())
74 self
._context
_fc
.append_member(
75 'zola', self
._tc
.create_signed_integer_field_class(18)
78 self
._payload
_fc
= self
._tc
.create_structure_field_class()
79 self
._payload
_fc
.append_member('zoom', self
._tc
.create_string_field_class())
81 self
._stream
_class
= self
._tc
.create_stream_class(
82 assigns_automatic_event_class_id
=True
85 def test_create_default(self
):
86 ec
= self
._stream
_class
.create_event_class()
88 self
.assertIs(type(ec
), bt2_event_class
._EventClass
)
89 self
.assertIsNone(ec
.name
, 'my_event')
90 self
.assertTrue(type(ec
.id), int)
91 self
.assertIsNone(ec
.specific_context_field_class
)
92 self
.assertIsNone(ec
.payload_field_class
)
93 self
.assertIsNone(ec
.emf_uri
)
94 self
.assertIsNone(ec
.log_level
)
95 self
.assertEqual(len(ec
.user_attributes
), 0)
97 def test_create_invalid_id(self
):
98 sc
= self
._tc
.create_stream_class(assigns_automatic_event_class_id
=False)
99 with self
.assertRaises(TypeError):
100 sc
.create_event_class(id='lel')
102 self
.assertEqual(len(sc
), 0)
104 def test_create_specific_context_field_class(self
):
105 fc
= self
._tc
.create_structure_field_class()
106 ec
= self
._stream
_class
.create_event_class(specific_context_field_class
=fc
)
107 self
.assertEqual(ec
.specific_context_field_class
.addr
, fc
.addr
)
109 type(ec
.specific_context_field_class
), bt2_field_class
._StructureFieldClass
112 def test_const_create_specific_context_field_class(self
):
113 ec_const
= _create_const_event_class(self
._tc
, self
._stream
_class
)
115 type(ec_const
.specific_context_field_class
),
116 bt2_field_class
._StructureFieldClassConst
,
119 def test_create_invalid_specific_context_field_class(self
):
120 with self
.assertRaises(TypeError):
121 self
._stream
_class
.create_event_class(specific_context_field_class
='lel')
123 self
.assertEqual(len(self
._stream
_class
), 0)
125 def test_create_payload_field_class(self
):
126 fc
= self
._tc
.create_structure_field_class()
127 ec
= self
._stream
_class
.create_event_class(payload_field_class
=fc
)
128 self
.assertEqual(ec
.payload_field_class
.addr
, fc
.addr
)
130 type(ec
.payload_field_class
), bt2_field_class
._StructureFieldClass
133 def test_const_create_payload_field_class(self
):
134 ec_const
= _create_const_event_class(self
._tc
, self
._stream
_class
)
136 type(ec_const
.payload_field_class
),
137 bt2_field_class
._StructureFieldClassConst
,
140 def test_create_invalid_payload_field_class(self
):
141 with self
.assertRaises(TypeError):
142 self
._stream
_class
.create_event_class(payload_field_class
='lel')
144 self
.assertEqual(len(self
._stream
_class
), 0)
146 def test_create_name(self
):
147 ec
= self
._stream
_class
.create_event_class(name
='viande à chien')
148 self
.assertEqual(ec
.name
, 'viande à chien')
150 def test_create_invalid_name(self
):
151 with self
.assertRaises(TypeError):
152 self
._stream
_class
.create_event_class(name
=2)
154 self
.assertEqual(len(self
._stream
_class
), 0)
156 def test_emf_uri(self
):
157 ec
= self
._stream
_class
.create_event_class(emf_uri
='salut')
158 self
.assertEqual(ec
.emf_uri
, 'salut')
160 def test_create_invalid_emf_uri(self
):
161 with self
.assertRaises(TypeError):
162 self
._stream
_class
.create_event_class(emf_uri
=23)
164 self
.assertEqual(len(self
._stream
_class
), 0)
166 def test_create_log_level(self
):
167 ec
= self
._stream
_class
.create_event_class(
168 log_level
=bt2
.EventClassLogLevel
.EMERGENCY
170 self
.assertEqual(ec
.log_level
, bt2
.EventClassLogLevel
.EMERGENCY
)
172 def test_create_invalid_log_level(self
):
173 with self
.assertRaises(ValueError):
174 self
._stream
_class
.create_event_class(log_level
='zoom')
176 self
.assertEqual(len(self
._stream
_class
), 0)
178 def test_create_user_attributes(self
):
179 ec
= self
._stream
_class
.create_event_class(user_attributes
={'salut': 23})
180 self
.assertEqual(ec
.user_attributes
, {'salut': 23})
181 self
.assertIs(type(ec
.user_attributes
), bt2_value
.MapValue
)
183 def test_const_create_user_attributes(self
):
184 ec_const
= _create_const_event_class(self
._tc
, self
._stream
_class
)
185 self
.assertIs(type(ec_const
.user_attributes
), bt2_value
._MapValueConst
)
187 def test_create_invalid_user_attributes(self
):
188 with self
.assertRaises(TypeError):
189 self
._stream
_class
.create_event_class(user_attributes
=object())
191 self
.assertEqual(len(self
._stream
_class
), 0)
193 def test_create_invalid_user_attributes_value_type(self
):
194 with self
.assertRaises(TypeError):
195 self
._stream
_class
.create_event_class(user_attributes
=23)
197 self
.assertEqual(len(self
._stream
_class
), 0)
199 def test_stream_class(self
):
200 ec
= self
._stream
_class
.create_event_class()
201 self
.assertEqual(ec
.stream_class
.addr
, self
._stream
_class
.addr
)
202 self
.assertIs(type(ec
.stream_class
), bt2_stream_class
._StreamClass
)
204 def test_const_stream_class(self
):
205 ec_const
= _create_const_event_class(self
._tc
, self
._stream
_class
)
206 self
.assertIs(type(ec_const
.stream_class
), bt2_stream_class
._StreamClassConst
)
209 if __name__
== '__main__':