1 # SPDX-License-Identifier: GPL-2.0-only
3 # Copyright (C) 2019 EfficiOS Inc.
8 from utils
import get_default_trace_class
9 from bt2
import stream_class
as bt2_stream_class
10 from bt2
import event_class
as bt2_event_class
11 from bt2
import field_class
as bt2_field_class
12 from bt2
import value
as bt2_value
13 from utils
import TestOutputPortMessageIterator
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
):
27 stream
= trace
.create_stream(stream_class
)
29 self
._create
_stream
_beginning
_message
(stream
),
30 self
._create
_event
_message
(event_class
, stream
),
34 if len(self
._msgs
) == 0:
37 return self
._msgs
.pop(0)
39 class MySrc(bt2
._UserSourceComponent
, message_iterator_class
=MyIter
):
40 def __init__(self
, config
, params
, obj
):
41 self
._add
_output
_port
("out", params
)
44 src_comp
= graph
.add_component(MySrc
, "my_source", None)
45 msg_iter
= TestOutputPortMessageIterator(graph
, src_comp
.output_ports
["out"])
47 # Ignore first message, stream beginning
50 event_msg
= next(msg_iter
)
52 return event_msg
.event
.cls
55 class EventClassTestCase(unittest
.TestCase
):
57 self
._tc
= get_default_trace_class()
59 self
._context
_fc
= self
._tc
.create_structure_field_class()
60 self
._context
_fc
.append_member("allo", self
._tc
.create_string_field_class())
61 self
._context
_fc
.append_member(
62 "zola", self
._tc
.create_signed_integer_field_class(18)
65 self
._payload
_fc
= self
._tc
.create_structure_field_class()
66 self
._payload
_fc
.append_member("zoom", self
._tc
.create_string_field_class())
68 self
._stream
_class
= self
._tc
.create_stream_class(
69 assigns_automatic_event_class_id
=True
72 def test_create_default(self
):
73 ec
= self
._stream
_class
.create_event_class()
75 self
.assertIs(type(ec
), bt2_event_class
._EventClass
)
76 self
.assertIsNone(ec
.name
, "my_event")
77 self
.assertTrue(type(ec
.id), int)
78 self
.assertIsNone(ec
.specific_context_field_class
)
79 self
.assertIsNone(ec
.payload_field_class
)
80 self
.assertIsNone(ec
.emf_uri
)
81 self
.assertIsNone(ec
.log_level
)
82 self
.assertEqual(len(ec
.user_attributes
), 0)
84 def test_create_invalid_id(self
):
85 sc
= self
._tc
.create_stream_class(assigns_automatic_event_class_id
=False)
86 with self
.assertRaises(TypeError):
87 sc
.create_event_class(id="lel")
89 self
.assertEqual(len(sc
), 0)
91 def test_create_specific_context_field_class(self
):
92 fc
= self
._tc
.create_structure_field_class()
93 ec
= self
._stream
_class
.create_event_class(specific_context_field_class
=fc
)
94 self
.assertEqual(ec
.specific_context_field_class
.addr
, fc
.addr
)
96 type(ec
.specific_context_field_class
), bt2_field_class
._StructureFieldClass
99 def test_const_create_specific_context_field_class(self
):
100 ec_const
= _create_const_event_class(self
._tc
, self
._stream
_class
)
102 type(ec_const
.specific_context_field_class
),
103 bt2_field_class
._StructureFieldClassConst
,
106 def test_create_invalid_specific_context_field_class(self
):
107 with self
.assertRaises(TypeError):
108 self
._stream
_class
.create_event_class(specific_context_field_class
="lel")
110 self
.assertEqual(len(self
._stream
_class
), 0)
112 def test_create_payload_field_class(self
):
113 fc
= self
._tc
.create_structure_field_class()
114 ec
= self
._stream
_class
.create_event_class(payload_field_class
=fc
)
115 self
.assertEqual(ec
.payload_field_class
.addr
, fc
.addr
)
117 type(ec
.payload_field_class
), bt2_field_class
._StructureFieldClass
120 def test_const_create_payload_field_class(self
):
121 ec_const
= _create_const_event_class(self
._tc
, self
._stream
_class
)
123 type(ec_const
.payload_field_class
),
124 bt2_field_class
._StructureFieldClassConst
,
127 def test_create_invalid_payload_field_class(self
):
128 with self
.assertRaises(TypeError):
129 self
._stream
_class
.create_event_class(payload_field_class
="lel")
131 self
.assertEqual(len(self
._stream
_class
), 0)
133 def test_create_name(self
):
134 ec
= self
._stream
_class
.create_event_class(name
="viande à chien")
135 self
.assertEqual(ec
.name
, "viande à chien")
137 def test_create_invalid_name(self
):
138 with self
.assertRaises(TypeError):
139 self
._stream
_class
.create_event_class(name
=2)
141 self
.assertEqual(len(self
._stream
_class
), 0)
143 def test_emf_uri(self
):
144 ec
= self
._stream
_class
.create_event_class(emf_uri
="salut")
145 self
.assertEqual(ec
.emf_uri
, "salut")
147 def test_create_invalid_emf_uri(self
):
148 with self
.assertRaises(TypeError):
149 self
._stream
_class
.create_event_class(emf_uri
=23)
151 self
.assertEqual(len(self
._stream
_class
), 0)
153 def test_create_log_level(self
):
154 ec
= self
._stream
_class
.create_event_class(
155 log_level
=bt2
.EventClassLogLevel
.EMERGENCY
157 self
.assertEqual(ec
.log_level
, bt2
.EventClassLogLevel
.EMERGENCY
)
159 def test_create_invalid_log_level(self
):
160 with self
.assertRaises(ValueError):
161 self
._stream
_class
.create_event_class(log_level
="zoom")
163 self
.assertEqual(len(self
._stream
_class
), 0)
165 def test_create_user_attributes(self
):
166 ec
= self
._stream
_class
.create_event_class(user_attributes
={"salut": 23})
167 self
.assertEqual(ec
.user_attributes
, {"salut": 23})
168 self
.assertIs(type(ec
.user_attributes
), bt2_value
.MapValue
)
170 def test_const_create_user_attributes(self
):
171 ec_const
= _create_const_event_class(self
._tc
, self
._stream
_class
)
172 self
.assertIs(type(ec_const
.user_attributes
), bt2_value
._MapValueConst
)
174 def test_create_invalid_user_attributes(self
):
175 with self
.assertRaises(TypeError):
176 self
._stream
_class
.create_event_class(user_attributes
=object())
178 self
.assertEqual(len(self
._stream
_class
), 0)
180 def test_create_invalid_user_attributes_value_type(self
):
181 with self
.assertRaises(TypeError):
182 self
._stream
_class
.create_event_class(user_attributes
=23)
184 self
.assertEqual(len(self
._stream
_class
), 0)
186 def test_stream_class(self
):
187 ec
= self
._stream
_class
.create_event_class()
188 self
.assertEqual(ec
.stream_class
.addr
, self
._stream
_class
.addr
)
189 self
.assertIs(type(ec
.stream_class
), bt2_stream_class
._StreamClass
)
191 def test_const_stream_class(self
):
192 ec_const
= _create_const_event_class(self
._tc
, self
._stream
_class
)
193 self
.assertIs(type(ec_const
.stream_class
), bt2_stream_class
._StreamClassConst
)
196 if __name__
== "__main__":