1 # SPDX-License-Identifier: GPL-2.0-only
3 # Copyright (C) 2019 EfficiOS Inc.
10 from utils
import run_in_component_init
, TestOutputPortMessageIterator
11 from bt2
import value
as bt2_value
12 from bt2
import clock_class
as bt2_clock_class
15 class ClockClassOffsetTestCase(unittest
.TestCase
):
16 def test_create_default(self
):
17 cco
= bt2
.ClockClassOffset()
18 self
.assertEqual(cco
.seconds
, 0)
19 self
.assertEqual(cco
.cycles
, 0)
21 def test_create(self
):
22 cco
= bt2
.ClockClassOffset(23, 4871232)
23 self
.assertEqual(cco
.seconds
, 23)
24 self
.assertEqual(cco
.cycles
, 4871232)
26 def test_create_kwargs(self
):
27 cco
= bt2
.ClockClassOffset(seconds
=23, cycles
=4871232)
28 self
.assertEqual(cco
.seconds
, 23)
29 self
.assertEqual(cco
.cycles
, 4871232)
31 def test_create_invalid_seconds(self
):
32 with self
.assertRaises(TypeError):
33 bt2
.ClockClassOffset('hello', 4871232)
35 def test_create_invalid_cycles(self
):
36 with self
.assertRaises(TypeError):
37 bt2
.ClockClassOffset(23, 'hello')
40 cco1
= bt2
.ClockClassOffset(23, 42)
41 cco2
= bt2
.ClockClassOffset(23, 42)
42 self
.assertEqual(cco1
, cco2
)
44 def test_ne_seconds(self
):
45 cco1
= bt2
.ClockClassOffset(23, 42)
46 cco2
= bt2
.ClockClassOffset(24, 42)
47 self
.assertNotEqual(cco1
, cco2
)
49 def test_ne_cycles(self
):
50 cco1
= bt2
.ClockClassOffset(23, 42)
51 cco2
= bt2
.ClockClassOffset(23, 43)
52 self
.assertNotEqual(cco1
, cco2
)
54 def test_eq_invalid(self
):
55 self
.assertFalse(bt2
.ClockClassOffset() == 23)
58 class ClockClassTestCase(unittest
.TestCase
):
59 def assertRaisesInComponentInit(self
, expected_exc_type
, user_code
):
63 except Exception as exc
:
66 exc_type
= run_in_component_init(f
)
67 self
.assertIsNotNone(exc_type
)
68 self
.assertEqual(exc_type
, expected_exc_type
)
70 def test_create_default(self
):
71 cc
= run_in_component_init(lambda comp_self
: comp_self
._create
_clock
_class
())
73 self
.assertIsNone(cc
.name
)
74 self
.assertEqual(cc
.frequency
, 1000000000)
75 self
.assertIsNone(cc
.description
)
76 self
.assertEqual(cc
.precision
, 0)
77 self
.assertEqual(cc
.offset
, bt2
.ClockClassOffset())
78 self
.assertTrue(cc
.origin_is_unix_epoch
)
79 self
.assertIsNone(cc
.uuid
)
80 self
.assertEqual(len(cc
.user_attributes
), 0)
82 def test_create_name(self
):
84 return comp_self
._create
_clock
_class
(name
='the_clock')
86 cc
= run_in_component_init(f
)
87 self
.assertEqual(cc
.name
, 'the_clock')
89 def test_create_invalid_name(self
):
91 comp_self
._create
_clock
_class
(name
=23)
93 self
.assertRaisesInComponentInit(TypeError, f
)
95 def test_create_description(self
):
97 return comp_self
._create
_clock
_class
(description
='hi people')
99 cc
= run_in_component_init(f
)
100 self
.assertEqual(cc
.description
, 'hi people')
102 def test_create_invalid_description(self
):
104 return comp_self
._create
_clock
_class
(description
=23)
106 self
.assertRaisesInComponentInit(TypeError, f
)
108 def test_create_frequency(self
):
110 return comp_self
._create
_clock
_class
(frequency
=987654321)
112 cc
= run_in_component_init(f
)
113 self
.assertEqual(cc
.frequency
, 987654321)
115 def test_create_invalid_frequency(self
):
117 return comp_self
._create
_clock
_class
(frequency
='lel')
119 self
.assertRaisesInComponentInit(TypeError, f
)
121 def test_create_precision(self
):
123 return comp_self
._create
_clock
_class
(precision
=12)
125 cc
= run_in_component_init(f
)
126 self
.assertEqual(cc
.precision
, 12)
128 def test_create_invalid_precision(self
):
130 return comp_self
._create
_clock
_class
(precision
='lel')
132 self
.assertRaisesInComponentInit(TypeError, f
)
134 def test_create_offset(self
):
136 return comp_self
._create
_clock
_class
(offset
=bt2
.ClockClassOffset(12, 56))
138 cc
= run_in_component_init(f
)
139 self
.assertEqual(cc
.offset
, bt2
.ClockClassOffset(12, 56))
141 def test_create_invalid_offset(self
):
143 return comp_self
._create
_clock
_class
(offset
=object())
145 self
.assertRaisesInComponentInit(TypeError, f
)
147 def test_create_origin_is_unix_epoch(self
):
149 return comp_self
._create
_clock
_class
(origin_is_unix_epoch
=False)
151 cc
= run_in_component_init(f
)
152 self
.assertEqual(cc
.origin_is_unix_epoch
, False)
154 def test_create_invalid_origin_is_unix_epoch(self
):
156 return comp_self
._create
_clock
_class
(origin_is_unix_epoch
=23)
158 self
.assertRaisesInComponentInit(TypeError, f
)
160 def test_cycles_to_ns_from_origin(self
):
162 return comp_self
._create
_clock
_class
(
163 frequency
=10 ** 8, origin_is_unix_epoch
=True
166 cc
= run_in_component_init(f
)
167 self
.assertEqual(cc
.cycles_to_ns_from_origin(112), 1120)
169 def test_cycles_to_ns_from_origin_overflow(self
):
171 return comp_self
._create
_clock
_class
(frequency
=1000)
173 cc
= run_in_component_init(f
)
174 with self
.assertRaises(bt2
._OverflowError
):
175 cc
.cycles_to_ns_from_origin(2 ** 63)
177 def test_create_uuid(self
):
179 return comp_self
._create
_clock
_class
(
180 uuid
=uuid
.UUID('b43372c32ef0be28444dfc1c5cdafd33')
183 cc
= run_in_component_init(f
)
184 self
.assertEqual(cc
.uuid
, uuid
.UUID('b43372c32ef0be28444dfc1c5cdafd33'))
186 def test_create_invalid_uuid(self
):
188 return comp_self
._create
_clock
_class
(uuid
=23)
190 self
.assertRaisesInComponentInit(TypeError, f
)
192 def test_create_user_attributes(self
):
194 return comp_self
._create
_clock
_class
(user_attributes
={'salut': 23})
196 cc
= run_in_component_init(f
)
197 self
.assertEqual(cc
.user_attributes
, {'salut': 23})
198 self
.assertIs(type(cc
.user_attributes
), bt2_value
.MapValue
)
200 def test_create_invalid_user_attributes(self
):
202 return comp_self
._create
_clock
_class
(user_attributes
=object())
204 self
.assertRaisesInComponentInit(TypeError, f
)
206 def test_create_invalid_user_attributes_value_type(self
):
208 return comp_self
._create
_clock
_class
(user_attributes
=23)
210 self
.assertRaisesInComponentInit(TypeError, f
)
212 def test_const_user_attributes(self
):
213 cc
= utils
.get_const_event_message().default_clock_snapshot
.clock_class
214 self
.assertIs(type(cc
.user_attributes
), bt2_value
._MapValueConst
)
217 class ClockSnapshotTestCase(unittest
.TestCase
):
220 cc
= comp_self
._create
_clock
_class
(
221 1000, 'my_cc', offset
=bt2
.ClockClassOffset(45, 354)
223 tc
= comp_self
._create
_trace
_class
()
227 _cc
, _tc
= run_in_component_init(f
)
229 _sc
= _tc
.create_stream_class(default_clock_class
=_cc
)
230 _ec
= _sc
.create_event_class(name
='salut')
231 _stream
= _trace
.create_stream(_sc
)
232 self
._stream
= _stream
236 class MyIter(bt2
._UserMessageIterator
):
237 def __init__(self
, config
, self_port_output
):
242 notif
= self
._create
_stream
_beginning
_message
(_stream
)
244 notif
= self
._create
_event
_message
(_ec
, _stream
, 123)
246 notif
= self
._create
_event
_message
(_ec
, _stream
, 2 ** 63)
248 notif
= self
._create
_stream
_end
_message
(_stream
)
255 class MySrc(bt2
._UserSourceComponent
, message_iterator_class
=MyIter
):
256 def __init__(self
, config
, params
, obj
):
257 self
._add
_output
_port
('out')
259 self
._graph
= bt2
.Graph()
260 self
._src
_comp
= self
._graph
.add_component(MySrc
, 'my_source')
261 self
._msg
_iter
= TestOutputPortMessageIterator(
262 self
._graph
, self
._src
_comp
.output_ports
['out']
265 for i
, msg
in enumerate(self
._msg
_iter
):
269 self
._msg
_clock
_overflow
= msg
276 def test_create_default(self
):
278 self
._msg
.default_clock_snapshot
.clock_class
.addr
, self
._cc
.addr
280 self
.assertEqual(self
._msg
.default_clock_snapshot
.value
, 123)
282 def test_clock_class(self
):
283 cc
= self
._msg
.default_clock_snapshot
.clock_class
284 self
.assertEqual(cc
.addr
, self
._cc
.addr
)
285 self
.assertIs(type(cc
), bt2_clock_class
._ClockClassConst
)
287 def test_ns_from_origin(self
):
288 s_from_origin
= 45 + ((354 + 123) / 1000)
289 ns_from_origin
= int(s_from_origin
* 1e9
)
291 self
._msg
.default_clock_snapshot
.ns_from_origin
, ns_from_origin
294 def test_ns_from_origin_overflow(self
):
295 with self
.assertRaises(bt2
._OverflowError
):
296 self
._msg
_clock
_overflow
.default_clock_snapshot
.ns_from_origin
298 def test_eq_int(self
):
299 self
.assertEqual(self
._msg
.default_clock_snapshot
, 123)
301 def test_eq_invalid(self
):
302 self
.assertFalse(self
._msg
.default_clock_snapshot
== 23)
304 def test_comparison(self
):
305 self
.assertTrue(self
._msg
.default_clock_snapshot
> 100)
306 self
.assertFalse(self
._msg
.default_clock_snapshot
> 200)
308 self
.assertTrue(self
._msg
.default_clock_snapshot
>= 123)
309 self
.assertFalse(self
._msg
.default_clock_snapshot
>= 200)
311 self
.assertTrue(self
._msg
.default_clock_snapshot
< 200)
312 self
.assertFalse(self
._msg
.default_clock_snapshot
< 100)
314 self
.assertTrue(self
._msg
.default_clock_snapshot
<= 123)
315 self
.assertFalse(self
._msg
.default_clock_snapshot
<= 100)
318 if __name__
== '__main__':