1 # SPDX-License-Identifier: GPL-2.0-only
3 # Copyright (C) 2019 EfficiOS Inc.
9 from bt2
import value
as bt2_value
10 from bt2
import event_class
as bt2_event_class
11 from bt2
import field_class
as bt2_field_class
12 from bt2
import stream_class
as bt2_stream_class
13 from utils
import TestOutputPortMessageIterator
, get_default_trace_class
16 def _create_const_event_class(tc
, stream_class
):
17 fc1
= tc
.create_structure_field_class()
18 fc2
= tc
.create_structure_field_class()
19 event_class
= stream_class
.create_event_class(
20 payload_field_class
=fc1
, specific_context_field_class
=fc2
23 class MyIter(bt2
._UserMessageIterator
):
24 def __init__(self
, config
, self_port_output
):
26 stream
= trace
.create_stream(stream_class
)
28 self
._create
_stream
_beginning
_message
(stream
),
29 self
._create
_event
_message
(event_class
, stream
),
33 if len(self
._msgs
) == 0:
36 return self
._msgs
.pop(0)
38 class MySrc(bt2
._UserSourceComponent
, message_iterator_class
=MyIter
):
39 def __init__(self
, config
, params
, obj
):
40 self
._add
_output
_port
("out", params
)
43 src_comp
= graph
.add_component(MySrc
, "my_source", None)
44 msg_iter
= TestOutputPortMessageIterator(graph
, src_comp
.output_ports
["out"])
46 # Ignore first message, stream beginning
49 event_msg
= next(msg_iter
)
51 return event_msg
.event
.cls
54 class EventClassTestCase(unittest
.TestCase
):
56 self
._tc
= get_default_trace_class()
58 self
._context
_fc
= self
._tc
.create_structure_field_class()
59 self
._context
_fc
.append_member("allo", self
._tc
.create_string_field_class())
60 self
._context
_fc
.append_member(
61 "zola", self
._tc
.create_signed_integer_field_class(18)
64 self
._payload
_fc
= self
._tc
.create_structure_field_class()
65 self
._payload
_fc
.append_member("zoom", self
._tc
.create_string_field_class())
67 self
._stream
_class
= self
._tc
.create_stream_class(
68 assigns_automatic_event_class_id
=True
71 def test_create_default(self
):
72 ec
= self
._stream
_class
.create_event_class()
74 self
.assertIs(type(ec
), bt2_event_class
._EventClass
)
75 self
.assertIsNone(ec
.name
, "my_event")
76 self
.assertTrue(type(ec
.id), int)
77 self
.assertIsNone(ec
.specific_context_field_class
)
78 self
.assertIsNone(ec
.payload_field_class
)
79 self
.assertIsNone(ec
.emf_uri
)
80 self
.assertIsNone(ec
.log_level
)
81 self
.assertEqual(len(ec
.user_attributes
), 0)
83 def test_create_invalid_id(self
):
84 sc
= self
._tc
.create_stream_class(assigns_automatic_event_class_id
=False)
85 with self
.assertRaises(TypeError):
86 sc
.create_event_class(id="lel")
88 self
.assertEqual(len(sc
), 0)
90 def test_create_specific_context_field_class(self
):
91 fc
= self
._tc
.create_structure_field_class()
92 ec
= self
._stream
_class
.create_event_class(specific_context_field_class
=fc
)
93 self
.assertEqual(ec
.specific_context_field_class
.addr
, fc
.addr
)
95 type(ec
.specific_context_field_class
), bt2_field_class
._StructureFieldClass
98 def test_const_create_specific_context_field_class(self
):
99 ec_const
= _create_const_event_class(self
._tc
, self
._stream
_class
)
101 type(ec_const
.specific_context_field_class
),
102 bt2_field_class
._StructureFieldClassConst
,
105 def test_create_invalid_specific_context_field_class(self
):
106 with self
.assertRaises(TypeError):
107 self
._stream
_class
.create_event_class(specific_context_field_class
="lel")
109 self
.assertEqual(len(self
._stream
_class
), 0)
111 def test_create_payload_field_class(self
):
112 fc
= self
._tc
.create_structure_field_class()
113 ec
= self
._stream
_class
.create_event_class(payload_field_class
=fc
)
114 self
.assertEqual(ec
.payload_field_class
.addr
, fc
.addr
)
116 type(ec
.payload_field_class
), bt2_field_class
._StructureFieldClass
119 def test_const_create_payload_field_class(self
):
120 ec_const
= _create_const_event_class(self
._tc
, self
._stream
_class
)
122 type(ec_const
.payload_field_class
),
123 bt2_field_class
._StructureFieldClassConst
,
126 def test_create_invalid_payload_field_class(self
):
127 with self
.assertRaises(TypeError):
128 self
._stream
_class
.create_event_class(payload_field_class
="lel")
130 self
.assertEqual(len(self
._stream
_class
), 0)
132 def test_create_name(self
):
133 ec
= self
._stream
_class
.create_event_class(name
="viande à chien")
134 self
.assertEqual(ec
.name
, "viande à chien")
136 def test_create_invalid_name(self
):
137 with self
.assertRaises(TypeError):
138 self
._stream
_class
.create_event_class(name
=2)
140 self
.assertEqual(len(self
._stream
_class
), 0)
142 def test_emf_uri(self
):
143 ec
= self
._stream
_class
.create_event_class(emf_uri
="salut")
144 self
.assertEqual(ec
.emf_uri
, "salut")
146 def test_create_invalid_emf_uri(self
):
147 with self
.assertRaises(TypeError):
148 self
._stream
_class
.create_event_class(emf_uri
=23)
150 self
.assertEqual(len(self
._stream
_class
), 0)
152 def test_create_log_level(self
):
153 ec
= self
._stream
_class
.create_event_class(
154 log_level
=bt2
.EventClassLogLevel
.EMERGENCY
156 self
.assertEqual(ec
.log_level
, bt2
.EventClassLogLevel
.EMERGENCY
)
158 def test_create_invalid_log_level(self
):
159 with self
.assertRaises(ValueError):
160 self
._stream
_class
.create_event_class(log_level
="zoom")
162 self
.assertEqual(len(self
._stream
_class
), 0)
164 def test_create_user_attributes(self
):
165 ec
= self
._stream
_class
.create_event_class(user_attributes
={"salut": 23})
166 self
.assertEqual(ec
.user_attributes
, {"salut": 23})
167 self
.assertIs(type(ec
.user_attributes
), bt2_value
.MapValue
)
169 def test_const_create_user_attributes(self
):
170 ec_const
= _create_const_event_class(self
._tc
, self
._stream
_class
)
171 self
.assertIs(type(ec_const
.user_attributes
), bt2_value
._MapValueConst
)
173 def test_create_invalid_user_attributes(self
):
174 with self
.assertRaises(TypeError):
175 self
._stream
_class
.create_event_class(user_attributes
=object())
177 self
.assertEqual(len(self
._stream
_class
), 0)
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 self
.assertEqual(len(self
._stream
_class
), 0)
185 def test_stream_class(self
):
186 ec
= self
._stream
_class
.create_event_class()
187 self
.assertEqual(ec
.stream_class
.addr
, self
._stream
_class
.addr
)
188 self
.assertIs(type(ec
.stream_class
), bt2_stream_class
._StreamClass
)
190 def test_const_stream_class(self
):
191 ec_const
= _create_const_event_class(self
._tc
, self
._stream
_class
)
192 self
.assertIs(type(ec_const
.stream_class
), bt2_stream_class
._StreamClassConst
)
195 if __name__
== "__main__":