lib: make can_seek_ns_from_origin logic use `can_seek_forward` property of iterator
[babeltrace.git] / tests / bindings / python / bt2 / test_clock_class.py
1 #
2 # Copyright (C) 2019 EfficiOS Inc.
3 #
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
7 # of the License.
8 #
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.
13 #
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.
17 #
18
19 import unittest
20 import uuid
21 import bt2
22 import utils
23 from utils import run_in_component_init, TestOutputPortMessageIterator
24 from bt2 import value as bt2_value
25 from bt2 import clock_class as bt2_clock_class
26
27
28 class ClockClassOffsetTestCase(unittest.TestCase):
29 def test_create_default(self):
30 cco = bt2.ClockClassOffset()
31 self.assertEqual(cco.seconds, 0)
32 self.assertEqual(cco.cycles, 0)
33
34 def test_create(self):
35 cco = bt2.ClockClassOffset(23, 4871232)
36 self.assertEqual(cco.seconds, 23)
37 self.assertEqual(cco.cycles, 4871232)
38
39 def test_create_kwargs(self):
40 cco = bt2.ClockClassOffset(seconds=23, cycles=4871232)
41 self.assertEqual(cco.seconds, 23)
42 self.assertEqual(cco.cycles, 4871232)
43
44 def test_create_invalid_seconds(self):
45 with self.assertRaises(TypeError):
46 bt2.ClockClassOffset('hello', 4871232)
47
48 def test_create_invalid_cycles(self):
49 with self.assertRaises(TypeError):
50 bt2.ClockClassOffset(23, 'hello')
51
52 def test_eq(self):
53 cco1 = bt2.ClockClassOffset(23, 42)
54 cco2 = bt2.ClockClassOffset(23, 42)
55 self.assertEqual(cco1, cco2)
56
57 def test_ne_seconds(self):
58 cco1 = bt2.ClockClassOffset(23, 42)
59 cco2 = bt2.ClockClassOffset(24, 42)
60 self.assertNotEqual(cco1, cco2)
61
62 def test_ne_cycles(self):
63 cco1 = bt2.ClockClassOffset(23, 42)
64 cco2 = bt2.ClockClassOffset(23, 43)
65 self.assertNotEqual(cco1, cco2)
66
67 def test_eq_invalid(self):
68 self.assertFalse(bt2.ClockClassOffset() == 23)
69
70
71 class ClockClassTestCase(unittest.TestCase):
72 def assertRaisesInComponentInit(self, expected_exc_type, user_code):
73 def f(comp_self):
74 try:
75 user_code(comp_self)
76 except Exception as exc:
77 return type(exc)
78
79 exc_type = run_in_component_init(f)
80 self.assertIsNotNone(exc_type)
81 self.assertEqual(exc_type, expected_exc_type)
82
83 def test_create_default(self):
84 cc = run_in_component_init(lambda comp_self: comp_self._create_clock_class())
85
86 self.assertIsNone(cc.name)
87 self.assertEqual(cc.frequency, 1000000000)
88 self.assertIsNone(cc.description)
89 self.assertEqual(cc.precision, 0)
90 self.assertEqual(cc.offset, bt2.ClockClassOffset())
91 self.assertTrue(cc.origin_is_unix_epoch)
92 self.assertIsNone(cc.uuid)
93 self.assertEqual(len(cc.user_attributes), 0)
94
95 def test_create_name(self):
96 def f(comp_self):
97 return comp_self._create_clock_class(name='the_clock')
98
99 cc = run_in_component_init(f)
100 self.assertEqual(cc.name, 'the_clock')
101
102 def test_create_invalid_name(self):
103 def f(comp_self):
104 comp_self._create_clock_class(name=23)
105
106 self.assertRaisesInComponentInit(TypeError, f)
107
108 def test_create_description(self):
109 def f(comp_self):
110 return comp_self._create_clock_class(description='hi people')
111
112 cc = run_in_component_init(f)
113 self.assertEqual(cc.description, 'hi people')
114
115 def test_create_invalid_description(self):
116 def f(comp_self):
117 return comp_self._create_clock_class(description=23)
118
119 self.assertRaisesInComponentInit(TypeError, f)
120
121 def test_create_frequency(self):
122 def f(comp_self):
123 return comp_self._create_clock_class(frequency=987654321)
124
125 cc = run_in_component_init(f)
126 self.assertEqual(cc.frequency, 987654321)
127
128 def test_create_invalid_frequency(self):
129 def f(comp_self):
130 return comp_self._create_clock_class(frequency='lel')
131
132 self.assertRaisesInComponentInit(TypeError, f)
133
134 def test_create_precision(self):
135 def f(comp_self):
136 return comp_self._create_clock_class(precision=12)
137
138 cc = run_in_component_init(f)
139 self.assertEqual(cc.precision, 12)
140
141 def test_create_invalid_precision(self):
142 def f(comp_self):
143 return comp_self._create_clock_class(precision='lel')
144
145 self.assertRaisesInComponentInit(TypeError, f)
146
147 def test_create_offset(self):
148 def f(comp_self):
149 return comp_self._create_clock_class(offset=bt2.ClockClassOffset(12, 56))
150
151 cc = run_in_component_init(f)
152 self.assertEqual(cc.offset, bt2.ClockClassOffset(12, 56))
153
154 def test_create_invalid_offset(self):
155 def f(comp_self):
156 return comp_self._create_clock_class(offset=object())
157
158 self.assertRaisesInComponentInit(TypeError, f)
159
160 def test_create_origin_is_unix_epoch(self):
161 def f(comp_self):
162 return comp_self._create_clock_class(origin_is_unix_epoch=False)
163
164 cc = run_in_component_init(f)
165 self.assertEqual(cc.origin_is_unix_epoch, False)
166
167 def test_create_invalid_origin_is_unix_epoch(self):
168 def f(comp_self):
169 return comp_self._create_clock_class(origin_is_unix_epoch=23)
170
171 self.assertRaisesInComponentInit(TypeError, f)
172
173 def test_cycles_to_ns_from_origin(self):
174 def f(comp_self):
175 return comp_self._create_clock_class(
176 frequency=10 ** 8, origin_is_unix_epoch=True
177 )
178
179 cc = run_in_component_init(f)
180 self.assertEqual(cc.cycles_to_ns_from_origin(112), 1120)
181
182 def test_cycles_to_ns_from_origin_overflow(self):
183 def f(comp_self):
184 return comp_self._create_clock_class(frequency=1000)
185
186 cc = run_in_component_init(f)
187 with self.assertRaises(bt2._OverflowError):
188 cc.cycles_to_ns_from_origin(2 ** 63)
189
190 def test_create_uuid(self):
191 def f(comp_self):
192 return comp_self._create_clock_class(
193 uuid=uuid.UUID('b43372c32ef0be28444dfc1c5cdafd33')
194 )
195
196 cc = run_in_component_init(f)
197 self.assertEqual(cc.uuid, uuid.UUID('b43372c32ef0be28444dfc1c5cdafd33'))
198
199 def test_create_invalid_uuid(self):
200 def f(comp_self):
201 return comp_self._create_clock_class(uuid=23)
202
203 self.assertRaisesInComponentInit(TypeError, f)
204
205 def test_create_user_attributes(self):
206 def f(comp_self):
207 return comp_self._create_clock_class(user_attributes={'salut': 23})
208
209 cc = run_in_component_init(f)
210 self.assertEqual(cc.user_attributes, {'salut': 23})
211 self.assertIs(type(cc.user_attributes), bt2_value.MapValue)
212
213 def test_create_invalid_user_attributes(self):
214 def f(comp_self):
215 return comp_self._create_clock_class(user_attributes=object())
216
217 self.assertRaisesInComponentInit(TypeError, f)
218
219 def test_create_invalid_user_attributes_value_type(self):
220 def f(comp_self):
221 return comp_self._create_clock_class(user_attributes=23)
222
223 self.assertRaisesInComponentInit(TypeError, f)
224
225 def test_const_user_attributes(self):
226 cc = utils.get_const_event_message().default_clock_snapshot.clock_class
227 self.assertIs(type(cc.user_attributes), bt2_value._MapValueConst)
228
229
230 class ClockSnapshotTestCase(unittest.TestCase):
231 def setUp(self):
232 def f(comp_self):
233 cc = comp_self._create_clock_class(
234 1000, 'my_cc', offset=bt2.ClockClassOffset(45, 354)
235 )
236 tc = comp_self._create_trace_class()
237
238 return (cc, tc)
239
240 _cc, _tc = run_in_component_init(f)
241 _trace = _tc()
242 _sc = _tc.create_stream_class(default_clock_class=_cc)
243 _ec = _sc.create_event_class(name='salut')
244 _stream = _trace.create_stream(_sc)
245 self._stream = _stream
246 self._ec = _ec
247 self._cc = _cc
248
249 class MyIter(bt2._UserMessageIterator):
250 def __init__(self, config, self_port_output):
251 self._at = 0
252
253 def __next__(self):
254 if self._at == 0:
255 notif = self._create_stream_beginning_message(_stream)
256 elif self._at == 1:
257 notif = self._create_event_message(_ec, _stream, 123)
258 elif self._at == 2:
259 notif = self._create_event_message(_ec, _stream, 2 ** 63)
260 elif self._at == 3:
261 notif = self._create_stream_end_message(_stream)
262 else:
263 raise bt2.Stop
264
265 self._at += 1
266 return notif
267
268 class MySrc(bt2._UserSourceComponent, message_iterator_class=MyIter):
269 def __init__(self, config, params, obj):
270 self._add_output_port('out')
271
272 self._graph = bt2.Graph()
273 self._src_comp = self._graph.add_component(MySrc, 'my_source')
274 self._msg_iter = TestOutputPortMessageIterator(
275 self._graph, self._src_comp.output_ports['out']
276 )
277
278 for i, msg in enumerate(self._msg_iter):
279 if i == 1:
280 self._msg = msg
281 elif i == 2:
282 self._msg_clock_overflow = msg
283 break
284
285 def tearDown(self):
286 del self._cc
287 del self._msg
288
289 def test_create_default(self):
290 self.assertEqual(
291 self._msg.default_clock_snapshot.clock_class.addr, self._cc.addr
292 )
293 self.assertEqual(self._msg.default_clock_snapshot.value, 123)
294
295 def test_clock_class(self):
296 cc = self._msg.default_clock_snapshot.clock_class
297 self.assertEqual(cc.addr, self._cc.addr)
298 self.assertIs(type(cc), bt2_clock_class._ClockClassConst)
299
300 def test_ns_from_origin(self):
301 s_from_origin = 45 + ((354 + 123) / 1000)
302 ns_from_origin = int(s_from_origin * 1e9)
303 self.assertEqual(
304 self._msg.default_clock_snapshot.ns_from_origin, ns_from_origin
305 )
306
307 def test_ns_from_origin_overflow(self):
308 with self.assertRaises(bt2._OverflowError):
309 self._msg_clock_overflow.default_clock_snapshot.ns_from_origin
310
311 def test_eq_int(self):
312 self.assertEqual(self._msg.default_clock_snapshot, 123)
313
314 def test_eq_invalid(self):
315 self.assertFalse(self._msg.default_clock_snapshot == 23)
316
317 def test_comparison(self):
318 self.assertTrue(self._msg.default_clock_snapshot > 100)
319 self.assertFalse(self._msg.default_clock_snapshot > 200)
320
321 self.assertTrue(self._msg.default_clock_snapshot >= 123)
322 self.assertFalse(self._msg.default_clock_snapshot >= 200)
323
324 self.assertTrue(self._msg.default_clock_snapshot < 200)
325 self.assertFalse(self._msg.default_clock_snapshot < 100)
326
327 self.assertTrue(self._msg.default_clock_snapshot <= 123)
328 self.assertFalse(self._msg.default_clock_snapshot <= 100)
329
330
331 if __name__ == '__main__':
332 unittest.main()
This page took 0.03653 seconds and 4 git commands to generate.