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
, 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 def test_create_specific_context_field_class(self
):
103 fc
= self
._tc
.create_structure_field_class()
104 ec
= self
._stream
_class
.create_event_class(specific_context_field_class
=fc
)
105 self
.assertEqual(ec
.specific_context_field_class
.addr
, fc
.addr
)
107 type(ec
.specific_context_field_class
), bt2_field_class
._StructureFieldClass
110 def test_const_create_specific_context_field_class(self
):
111 ec_const
= _create_const_event_class(self
._tc
, self
._stream
_class
)
113 type(ec_const
.specific_context_field_class
),
114 bt2_field_class
._StructureFieldClassConst
,
117 def test_create_invalid_specific_context_field_class(self
):
118 with self
.assertRaises(TypeError):
119 self
._stream
_class
.create_event_class(specific_context_field_class
='lel')
121 def test_create_payload_field_class(self
):
122 fc
= self
._tc
.create_structure_field_class()
123 ec
= self
._stream
_class
.create_event_class(payload_field_class
=fc
)
124 self
.assertEqual(ec
.payload_field_class
.addr
, fc
.addr
)
126 type(ec
.payload_field_class
), bt2_field_class
._StructureFieldClass
129 def test_const_create_payload_field_class(self
):
130 ec_const
= _create_const_event_class(self
._tc
, self
._stream
_class
)
132 type(ec_const
.payload_field_class
),
133 bt2_field_class
._StructureFieldClassConst
,
136 def test_create_invalid_payload_field_class(self
):
137 with self
.assertRaises(TypeError):
138 self
._stream
_class
.create_event_class(payload_field_class
='lel')
140 def test_create_name(self
):
141 ec
= self
._stream
_class
.create_event_class(name
='viande à chien')
142 self
.assertEqual(ec
.name
, 'viande à chien')
144 def test_create_invalid_name(self
):
145 with self
.assertRaises(TypeError):
146 self
._stream
_class
.create_event_class(name
=2)
148 def test_emf_uri(self
):
149 ec
= self
._stream
_class
.create_event_class(emf_uri
='salut')
150 self
.assertEqual(ec
.emf_uri
, 'salut')
152 def test_create_invalid_emf_uri(self
):
153 with self
.assertRaises(TypeError):
154 self
._stream
_class
.create_event_class(emf_uri
=23)
156 def test_create_log_level(self
):
157 ec
= self
._stream
_class
.create_event_class(
158 log_level
=bt2
.EventClassLogLevel
.EMERGENCY
160 self
.assertEqual(ec
.log_level
, bt2
.EventClassLogLevel
.EMERGENCY
)
162 def test_create_invalid_log_level(self
):
163 with self
.assertRaises(ValueError):
164 self
._stream
_class
.create_event_class(log_level
='zoom')
166 def test_create_user_attributes(self
):
167 ec
= self
._stream
_class
.create_event_class(user_attributes
={'salut': 23})
168 self
.assertEqual(ec
.user_attributes
, {'salut': 23})
169 self
.assertIs(type(ec
.user_attributes
), bt2_value
.MapValue
)
171 def test_const_create_user_attributes(self
):
172 ec_const
= _create_const_event_class(self
._tc
, self
._stream
_class
)
173 self
.assertIs(type(ec_const
.user_attributes
), bt2_value
._MapValueConst
)
175 def test_create_invalid_user_attributes(self
):
176 with self
.assertRaises(TypeError):
177 self
._stream
_class
.create_event_class(user_attributes
=object())
179 def test_create_invalid_user_attributes_value_type(self
):
180 with self
.assertRaises(TypeError):
181 self
._stream
_class
.create_event_class(user_attributes
=23)
183 def test_stream_class(self
):
184 ec
= self
._stream
_class
.create_event_class()
185 self
.assertEqual(ec
.stream_class
.addr
, self
._stream
_class
.addr
)
186 self
.assertIs(type(ec
.stream_class
), bt2_stream_class
._StreamClass
)
188 def test_const_stream_class(self
):
189 ec_const
= _create_const_event_class(self
._tc
, self
._stream
_class
)
190 self
.assertIs(type(ec_const
.stream_class
), bt2_stream_class
._StreamClassConst
)
193 if __name__
== '__main__':