1 # SPDX-License-Identifier: GPL-2.0-only
3 # Copyright (C) 2019 EfficiOS Inc.
11 from bt2
import value
as bt2_value
12 from bt2
import clock_class
as bt2_clock_class
13 from utils
import TestOutputPortMessageIterator
, run_in_component_init
16 class ClockClassOffsetTestCase(unittest
.TestCase
):
17 def test_create_default(self
):
18 cco
= bt2
.ClockClassOffset()
19 self
.assertEqual(cco
.seconds
, 0)
20 self
.assertEqual(cco
.cycles
, 0)
22 def test_create(self
):
23 cco
= bt2
.ClockClassOffset(23, 4871232)
24 self
.assertEqual(cco
.seconds
, 23)
25 self
.assertEqual(cco
.cycles
, 4871232)
27 def test_create_kwargs(self
):
28 cco
= bt2
.ClockClassOffset(seconds
=23, cycles
=4871232)
29 self
.assertEqual(cco
.seconds
, 23)
30 self
.assertEqual(cco
.cycles
, 4871232)
32 def test_create_invalid_seconds(self
):
33 with self
.assertRaises(TypeError):
34 bt2
.ClockClassOffset("hello", 4871232)
36 def test_create_invalid_cycles(self
):
37 with self
.assertRaises(TypeError):
38 bt2
.ClockClassOffset(23, "hello")
41 cco1
= bt2
.ClockClassOffset(23, 42)
42 cco2
= bt2
.ClockClassOffset(23, 42)
43 self
.assertEqual(cco1
, cco2
)
45 def test_ne_seconds(self
):
46 cco1
= bt2
.ClockClassOffset(23, 42)
47 cco2
= bt2
.ClockClassOffset(24, 42)
48 self
.assertNotEqual(cco1
, cco2
)
50 def test_ne_cycles(self
):
51 cco1
= bt2
.ClockClassOffset(23, 42)
52 cco2
= bt2
.ClockClassOffset(23, 43)
53 self
.assertNotEqual(cco1
, cco2
)
55 def test_eq_invalid(self
):
56 self
.assertFalse(bt2
.ClockClassOffset() == 23)
59 class ClockClassTestCase(unittest
.TestCase
):
60 def assertRaisesInComponentInit(self
, expected_exc_type
, user_code
):
64 except Exception as exc
:
67 exc_type
= run_in_component_init(f
)
68 self
.assertIsNotNone(exc_type
)
69 self
.assertEqual(exc_type
, expected_exc_type
)
71 def test_create_default(self
):
72 cc
= run_in_component_init(lambda comp_self
: comp_self
._create
_clock
_class
())
74 self
.assertIsNone(cc
.name
)
75 self
.assertEqual(cc
.frequency
, 1000000000)
76 self
.assertIsNone(cc
.description
)
77 self
.assertEqual(cc
.precision
, 0)
78 self
.assertEqual(cc
.offset
, bt2
.ClockClassOffset())
79 self
.assertTrue(cc
.origin_is_unix_epoch
)
80 self
.assertIsNone(cc
.uuid
)
81 self
.assertEqual(len(cc
.user_attributes
), 0)
83 def test_create_name(self
):
85 return comp_self
._create
_clock
_class
(name
="the_clock")
87 cc
= run_in_component_init(f
)
88 self
.assertEqual(cc
.name
, "the_clock")
90 def test_create_invalid_name(self
):
92 comp_self
._create
_clock
_class
(name
=23)
94 self
.assertRaisesInComponentInit(TypeError, f
)
96 def test_create_description(self
):
98 return comp_self
._create
_clock
_class
(description
="hi people")
100 cc
= run_in_component_init(f
)
101 self
.assertEqual(cc
.description
, "hi people")
103 def test_create_invalid_description(self
):
105 return comp_self
._create
_clock
_class
(description
=23)
107 self
.assertRaisesInComponentInit(TypeError, f
)
109 def test_create_frequency(self
):
111 return comp_self
._create
_clock
_class
(frequency
=987654321)
113 cc
= run_in_component_init(f
)
114 self
.assertEqual(cc
.frequency
, 987654321)
116 def test_create_invalid_frequency(self
):
118 return comp_self
._create
_clock
_class
(frequency
="lel")
120 self
.assertRaisesInComponentInit(TypeError, f
)
122 def test_create_precision(self
):
124 return comp_self
._create
_clock
_class
(precision
=12)
126 cc
= run_in_component_init(f
)
127 self
.assertEqual(cc
.precision
, 12)
129 def test_create_invalid_precision(self
):
131 return comp_self
._create
_clock
_class
(precision
="lel")
133 self
.assertRaisesInComponentInit(TypeError, f
)
135 def test_create_offset(self
):
137 return comp_self
._create
_clock
_class
(offset
=bt2
.ClockClassOffset(12, 56))
139 cc
= run_in_component_init(f
)
140 self
.assertEqual(cc
.offset
, bt2
.ClockClassOffset(12, 56))
142 def test_create_invalid_offset(self
):
144 return comp_self
._create
_clock
_class
(offset
=object())
146 self
.assertRaisesInComponentInit(TypeError, f
)
148 def test_create_origin_is_unix_epoch(self
):
150 return comp_self
._create
_clock
_class
(origin_is_unix_epoch
=False)
152 cc
= run_in_component_init(f
)
153 self
.assertEqual(cc
.origin_is_unix_epoch
, False)
155 def test_create_invalid_origin_is_unix_epoch(self
):
157 return comp_self
._create
_clock
_class
(origin_is_unix_epoch
=23)
159 self
.assertRaisesInComponentInit(TypeError, f
)
161 def test_cycles_to_ns_from_origin(self
):
163 return comp_self
._create
_clock
_class
(
164 frequency
=10**8, origin_is_unix_epoch
=True
167 cc
= run_in_component_init(f
)
168 self
.assertEqual(cc
.cycles_to_ns_from_origin(112), 1120)
170 def test_cycles_to_ns_from_origin_overflow(self
):
172 return comp_self
._create
_clock
_class
(frequency
=1000)
174 cc
= run_in_component_init(f
)
175 with self
.assertRaises(bt2
._OverflowError
):
176 cc
.cycles_to_ns_from_origin(2**63)
178 def test_create_uuid(self
):
180 return comp_self
._create
_clock
_class
(
181 uuid
=uuid
.UUID("b43372c32ef0be28444dfc1c5cdafd33")
184 cc
= run_in_component_init(f
)
185 self
.assertEqual(cc
.uuid
, uuid
.UUID("b43372c32ef0be28444dfc1c5cdafd33"))
187 def test_create_invalid_uuid(self
):
189 return comp_self
._create
_clock
_class
(uuid
=23)
191 self
.assertRaisesInComponentInit(TypeError, f
)
193 def test_create_user_attributes(self
):
195 return comp_self
._create
_clock
_class
(user_attributes
={"salut": 23})
197 cc
= run_in_component_init(f
)
198 self
.assertEqual(cc
.user_attributes
, {"salut": 23})
199 self
.assertIs(type(cc
.user_attributes
), bt2_value
.MapValue
)
201 def test_create_invalid_user_attributes(self
):
203 return comp_self
._create
_clock
_class
(user_attributes
=object())
205 self
.assertRaisesInComponentInit(TypeError, f
)
207 def test_create_invalid_user_attributes_value_type(self
):
209 return comp_self
._create
_clock
_class
(user_attributes
=23)
211 self
.assertRaisesInComponentInit(TypeError, f
)
213 def test_const_user_attributes(self
):
214 cc
= utils
.get_const_event_message().default_clock_snapshot
.clock_class
215 self
.assertIs(type(cc
.user_attributes
), bt2_value
._MapValueConst
)
218 class ClockSnapshotTestCase(unittest
.TestCase
):
221 cc
= comp_self
._create
_clock
_class
(
222 1000, "my_cc", offset
=bt2
.ClockClassOffset(45, 354)
224 tc
= comp_self
._create
_trace
_class
()
228 _cc
, _tc
= run_in_component_init(f
)
230 _sc
= _tc
.create_stream_class(default_clock_class
=_cc
)
231 _ec
= _sc
.create_event_class(name
="salut")
232 _stream
= _trace
.create_stream(_sc
)
233 self
._stream
= _stream
237 class MyIter(bt2
._UserMessageIterator
):
238 def __init__(self
, config
, self_port_output
):
243 notif
= self
._create
_stream
_beginning
_message
(_stream
)
245 notif
= self
._create
_event
_message
(_ec
, _stream
, 123)
247 notif
= self
._create
_event
_message
(_ec
, _stream
, 2**63)
249 notif
= self
._create
_stream
_end
_message
(_stream
)
256 class MySrc(bt2
._UserSourceComponent
, message_iterator_class
=MyIter
):
257 def __init__(self
, config
, params
, obj
):
258 self
._add
_output
_port
("out")
260 self
._graph
= bt2
.Graph()
261 self
._src
_comp
= self
._graph
.add_component(MySrc
, "my_source")
262 self
._msg
_iter
= TestOutputPortMessageIterator(
263 self
._graph
, self
._src
_comp
.output_ports
["out"]
266 for i
, msg
in enumerate(self
._msg
_iter
):
270 self
._msg
_clock
_overflow
= msg
277 def test_create_default(self
):
279 self
._msg
.default_clock_snapshot
.clock_class
.addr
, self
._cc
.addr
281 self
.assertEqual(self
._msg
.default_clock_snapshot
.value
, 123)
283 def test_clock_class(self
):
284 cc
= self
._msg
.default_clock_snapshot
.clock_class
285 self
.assertEqual(cc
.addr
, self
._cc
.addr
)
286 self
.assertIs(type(cc
), bt2_clock_class
._ClockClassConst
)
288 def test_ns_from_origin(self
):
289 s_from_origin
= 45 + ((354 + 123) / 1000)
290 ns_from_origin
= int(s_from_origin
* 1e9
)
292 self
._msg
.default_clock_snapshot
.ns_from_origin
, ns_from_origin
295 def test_ns_from_origin_overflow(self
):
296 with self
.assertRaises(bt2
._OverflowError
):
297 self
._msg
_clock
_overflow
.default_clock_snapshot
.ns_from_origin
299 def test_eq_int(self
):
300 self
.assertEqual(self
._msg
.default_clock_snapshot
, 123)
302 def test_eq_invalid(self
):
303 self
.assertFalse(self
._msg
.default_clock_snapshot
== 23)
305 def test_comparison(self
):
306 self
.assertTrue(self
._msg
.default_clock_snapshot
> 100)
307 self
.assertFalse(self
._msg
.default_clock_snapshot
> 200)
309 self
.assertTrue(self
._msg
.default_clock_snapshot
>= 123)
310 self
.assertFalse(self
._msg
.default_clock_snapshot
>= 200)
312 self
.assertTrue(self
._msg
.default_clock_snapshot
< 200)
313 self
.assertFalse(self
._msg
.default_clock_snapshot
< 100)
315 self
.assertTrue(self
._msg
.default_clock_snapshot
<= 123)
316 self
.assertFalse(self
._msg
.default_clock_snapshot
<= 100)
319 if __name__
== "__main__":