Move to kernel style SPDX license identifiers
[babeltrace.git] / tests / bindings / python / bt2 / test_event_class.py
CommitLineData
0235b0db 1# SPDX-License-Identifier: GPL-2.0-only
d2d857a8
MJ
2#
3# Copyright (C) 2019 EfficiOS Inc.
4#
d2d857a8 5
9cf643d1 6import unittest
9cf643d1 7import bt2
65531d55 8from utils import get_default_trace_class
f0a42b33
FD
9from bt2 import stream_class as bt2_stream_class
10from bt2 import event_class as bt2_event_class
11from bt2 import field_class as bt2_field_class
12from bt2 import value as bt2_value
13from utils import TestOutputPortMessageIterator
14
15
16def _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
21 )
22
23 class MyIter(bt2._UserMessageIterator):
8d8b141d 24 def __init__(self, config, self_port_output):
f0a42b33
FD
25
26 trace = tc()
27 stream = trace.create_stream(stream_class)
28 self._msgs = [
29 self._create_stream_beginning_message(stream),
30 self._create_event_message(event_class, stream),
31 ]
32
33 def __next__(self):
34 if len(self._msgs) == 0:
35 raise StopIteration
36
37 return self._msgs.pop(0)
38
39 class MySrc(bt2._UserSourceComponent, message_iterator_class=MyIter):
59225a3e 40 def __init__(self, config, params, obj):
f0a42b33
FD
41 self._add_output_port('out', params)
42
43 graph = bt2.Graph()
44 src_comp = graph.add_component(MySrc, 'my_source', None)
45 msg_iter = TestOutputPortMessageIterator(graph, src_comp.output_ports['out'])
46
47 # Ignore first message, stream beginning
48 _ = next(msg_iter)
49
50 event_msg = next(msg_iter)
51
52 return event_msg.event.cls
9cf643d1
PP
53
54
55class EventClassTestCase(unittest.TestCase):
56 def setUp(self):
65531d55
SM
57 self._tc = get_default_trace_class()
58
59 self._context_fc = self._tc.create_structure_field_class()
d47b87ac 60 self._context_fc.append_member('allo', self._tc.create_string_field_class())
cfbd7cf3
FD
61 self._context_fc.append_member(
62 'zola', self._tc.create_signed_integer_field_class(18)
63 )
65531d55
SM
64
65 self._payload_fc = self._tc.create_structure_field_class()
d47b87ac 66 self._payload_fc.append_member('zoom', self._tc.create_string_field_class())
9cf643d1 67
cfbd7cf3
FD
68 self._stream_class = self._tc.create_stream_class(
69 assigns_automatic_event_class_id=True
70 )
9cf643d1 71
65531d55
SM
72 def test_create_default(self):
73 ec = self._stream_class.create_event_class()
9cf643d1 74
f0a42b33 75 self.assertIs(type(ec), bt2_event_class._EventClass)
65531d55
SM
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)
5783664e 82 self.assertEqual(len(ec.user_attributes), 0)
65531d55
SM
83
84 def test_create_invalid_id(self):
85 sc = self._tc.create_stream_class(assigns_automatic_event_class_id=False)
9cf643d1 86 with self.assertRaises(TypeError):
65531d55
SM
87 sc.create_event_class(id='lel')
88
3321d9b0
SM
89 self.assertEqual(len(sc), 0)
90
65531d55
SM
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)
f0a42b33
FD
95 self.assertIs(
96 type(ec.specific_context_field_class), bt2_field_class._StructureFieldClass
97 )
98
99 def test_const_create_specific_context_field_class(self):
100 ec_const = _create_const_event_class(self._tc, self._stream_class)
101 self.assertIs(
102 type(ec_const.specific_context_field_class),
103 bt2_field_class._StructureFieldClassConst,
104 )
9cf643d1 105
65531d55
SM
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')
9cf643d1 109
3321d9b0
SM
110 self.assertEqual(len(self._stream_class), 0)
111
65531d55
SM
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)
f0a42b33
FD
116 self.assertIs(
117 type(ec.payload_field_class), bt2_field_class._StructureFieldClass
118 )
119
120 def test_const_create_payload_field_class(self):
121 ec_const = _create_const_event_class(self._tc, self._stream_class)
122 self.assertIs(
123 type(ec_const.payload_field_class),
124 bt2_field_class._StructureFieldClassConst,
125 )
9cf643d1 126
65531d55 127 def test_create_invalid_payload_field_class(self):
9cf643d1 128 with self.assertRaises(TypeError):
65531d55
SM
129 self._stream_class.create_event_class(payload_field_class='lel')
130
3321d9b0
SM
131 self.assertEqual(len(self._stream_class), 0)
132
65531d55
SM
133 def test_create_name(self):
134 ec = self._stream_class.create_event_class(name='viande à chien')
135 self.assertEqual(ec.name, 'viande à chien')
136
137 def test_create_invalid_name(self):
138 with self.assertRaises(TypeError):
139 self._stream_class.create_event_class(name=2)
140
3321d9b0
SM
141 self.assertEqual(len(self._stream_class), 0)
142
65531d55
SM
143 def test_emf_uri(self):
144 ec = self._stream_class.create_event_class(emf_uri='salut')
145 self.assertEqual(ec.emf_uri, 'salut')
146
147 def test_create_invalid_emf_uri(self):
811644b8 148 with self.assertRaises(TypeError):
65531d55 149 self._stream_class.create_event_class(emf_uri=23)
9cf643d1 150
3321d9b0
SM
151 self.assertEqual(len(self._stream_class), 0)
152
65531d55 153 def test_create_log_level(self):
cfbd7cf3
FD
154 ec = self._stream_class.create_event_class(
155 log_level=bt2.EventClassLogLevel.EMERGENCY
156 )
65531d55 157 self.assertEqual(ec.log_level, bt2.EventClassLogLevel.EMERGENCY)
9cf643d1 158
65531d55 159 def test_create_invalid_log_level(self):
811644b8 160 with self.assertRaises(ValueError):
65531d55
SM
161 self._stream_class.create_event_class(log_level='zoom')
162
3321d9b0
SM
163 self.assertEqual(len(self._stream_class), 0)
164
5783664e
PP
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})
f0a42b33
FD
168 self.assertIs(type(ec.user_attributes), bt2_value.MapValue)
169
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)
5783664e
PP
173
174 def test_create_invalid_user_attributes(self):
175 with self.assertRaises(TypeError):
176 self._stream_class.create_event_class(user_attributes=object())
177
3321d9b0
SM
178 self.assertEqual(len(self._stream_class), 0)
179
5783664e
PP
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)
183
3321d9b0
SM
184 self.assertEqual(len(self._stream_class), 0)
185
65531d55
SM
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)
f0a42b33
FD
189 self.assertIs(type(ec.stream_class), bt2_stream_class._StreamClass)
190
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)
d14ddbba
SM
194
195
196if __name__ == '__main__':
197 unittest.main()
This page took 0.07142 seconds and 4 git commands to generate.