2 # Copyright (C) 2019 EfficiOS Inc.
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
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.
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.
22 from utils
import run_in_component_init
, TestOutputPortMessageIterator
23 from bt2
import value
as bt2_value
24 from bt2
import clock_class
as bt2_clock_class
27 class ClockClassOffsetTestCase(unittest
.TestCase
):
28 def test_create_default(self
):
29 cco
= bt2
.ClockClassOffset()
30 self
.assertEqual(cco
.seconds
, 0)
31 self
.assertEqual(cco
.cycles
, 0)
33 def test_create(self
):
34 cco
= bt2
.ClockClassOffset(23, 4871232)
35 self
.assertEqual(cco
.seconds
, 23)
36 self
.assertEqual(cco
.cycles
, 4871232)
38 def test_create_kwargs(self
):
39 cco
= bt2
.ClockClassOffset(seconds
=23, cycles
=4871232)
40 self
.assertEqual(cco
.seconds
, 23)
41 self
.assertEqual(cco
.cycles
, 4871232)
43 def test_create_invalid_seconds(self
):
44 with self
.assertRaises(TypeError):
45 bt2
.ClockClassOffset('hello', 4871232)
47 def test_create_invalid_cycles(self
):
48 with self
.assertRaises(TypeError):
49 bt2
.ClockClassOffset(23, 'hello')
52 cco1
= bt2
.ClockClassOffset(23, 42)
53 cco2
= bt2
.ClockClassOffset(23, 42)
54 self
.assertEqual(cco1
, cco2
)
56 def test_ne_seconds(self
):
57 cco1
= bt2
.ClockClassOffset(23, 42)
58 cco2
= bt2
.ClockClassOffset(24, 42)
59 self
.assertNotEqual(cco1
, cco2
)
61 def test_ne_cycles(self
):
62 cco1
= bt2
.ClockClassOffset(23, 42)
63 cco2
= bt2
.ClockClassOffset(23, 43)
64 self
.assertNotEqual(cco1
, cco2
)
66 def test_eq_invalid(self
):
67 self
.assertFalse(bt2
.ClockClassOffset() == 23)
70 class ClockClassTestCase(unittest
.TestCase
):
71 def assertRaisesInComponentInit(self
, expected_exc_type
, user_code
):
75 except Exception as exc
:
78 exc_type
= run_in_component_init(f
)
79 self
.assertIsNotNone(exc_type
)
80 self
.assertEqual(exc_type
, expected_exc_type
)
82 def test_create_default(self
):
83 cc
= run_in_component_init(lambda comp_self
: comp_self
._create
_clock
_class
())
85 self
.assertIsNone(cc
.name
)
86 self
.assertEqual(cc
.frequency
, 1000000000)
87 self
.assertIsNone(cc
.description
)
88 self
.assertEqual(cc
.precision
, 0)
89 self
.assertEqual(cc
.offset
, bt2
.ClockClassOffset())
90 self
.assertTrue(cc
.origin_is_unix_epoch
)
91 self
.assertIsNone(cc
.uuid
)
92 self
.assertEqual(len(cc
.user_attributes
), 0)
94 def test_create_name(self
):
96 return comp_self
._create
_clock
_class
(name
='the_clock')
98 cc
= run_in_component_init(f
)
99 self
.assertEqual(cc
.name
, 'the_clock')
101 def test_create_invalid_name(self
):
103 comp_self
._create
_clock
_class
(name
=23)
105 self
.assertRaisesInComponentInit(TypeError, f
)
107 def test_create_description(self
):
109 return comp_self
._create
_clock
_class
(description
='hi people')
111 cc
= run_in_component_init(f
)
112 self
.assertEqual(cc
.description
, 'hi people')
114 def test_create_invalid_description(self
):
116 return comp_self
._create
_clock
_class
(description
=23)
118 self
.assertRaisesInComponentInit(TypeError, f
)
120 def test_create_frequency(self
):
122 return comp_self
._create
_clock
_class
(frequency
=987654321)
124 cc
= run_in_component_init(f
)
125 self
.assertEqual(cc
.frequency
, 987654321)
127 def test_create_invalid_frequency(self
):
129 return comp_self
._create
_clock
_class
(frequency
='lel')
131 self
.assertRaisesInComponentInit(TypeError, f
)
133 def test_create_precision(self
):
135 return comp_self
._create
_clock
_class
(precision
=12)
137 cc
= run_in_component_init(f
)
138 self
.assertEqual(cc
.precision
, 12)
140 def test_create_invalid_precision(self
):
142 return comp_self
._create
_clock
_class
(precision
='lel')
144 self
.assertRaisesInComponentInit(TypeError, f
)
146 def test_create_offset(self
):
148 return comp_self
._create
_clock
_class
(offset
=bt2
.ClockClassOffset(12, 56))
150 cc
= run_in_component_init(f
)
151 self
.assertEqual(cc
.offset
, bt2
.ClockClassOffset(12, 56))
153 def test_create_invalid_offset(self
):
155 return comp_self
._create
_clock
_class
(offset
=object())
157 self
.assertRaisesInComponentInit(TypeError, f
)
159 def test_create_origin_is_unix_epoch(self
):
161 return comp_self
._create
_clock
_class
(origin_is_unix_epoch
=False)
163 cc
= run_in_component_init(f
)
164 self
.assertEqual(cc
.origin_is_unix_epoch
, False)
166 def test_create_invalid_origin_is_unix_epoch(self
):
168 return comp_self
._create
_clock
_class
(origin_is_unix_epoch
=23)
170 self
.assertRaisesInComponentInit(TypeError, f
)
172 def test_cycles_to_ns_from_origin(self
):
174 return comp_self
._create
_clock
_class
(
175 frequency
=10 ** 8, origin_is_unix_epoch
=True
178 cc
= run_in_component_init(f
)
179 self
.assertEqual(cc
.cycles_to_ns_from_origin(112), 1120)
181 def test_cycles_to_ns_from_origin_overflow(self
):
183 return comp_self
._create
_clock
_class
(frequency
=1000)
185 cc
= run_in_component_init(f
)
186 with self
.assertRaises(bt2
._OverflowError
):
187 cc
.cycles_to_ns_from_origin(2 ** 63)
189 def test_create_uuid(self
):
191 return comp_self
._create
_clock
_class
(
192 uuid
=uuid
.UUID('b43372c32ef0be28444dfc1c5cdafd33')
195 cc
= run_in_component_init(f
)
196 self
.assertEqual(cc
.uuid
, uuid
.UUID('b43372c32ef0be28444dfc1c5cdafd33'))
198 def test_create_invalid_uuid(self
):
200 return comp_self
._create
_clock
_class
(uuid
=23)
202 self
.assertRaisesInComponentInit(TypeError, f
)
204 def test_create_user_attributes(self
):
206 return comp_self
._create
_clock
_class
(user_attributes
={'salut': 23})
208 cc
= run_in_component_init(f
)
209 self
.assertEqual(cc
.user_attributes
, {'salut': 23})
210 self
.assertIs(type(cc
.user_attributes
), bt2_value
.MapValue
)
212 def test_create_invalid_user_attributes(self
):
214 return comp_self
._create
_clock
_class
(user_attributes
=object())
216 self
.assertRaisesInComponentInit(TypeError, f
)
218 def test_create_invalid_user_attributes_value_type(self
):
220 return comp_self
._create
_clock
_class
(user_attributes
=23)
222 self
.assertRaisesInComponentInit(TypeError, f
)
225 class ClockSnapshotTestCase(unittest
.TestCase
):
228 cc
= comp_self
._create
_clock
_class
(
229 1000, 'my_cc', offset
=bt2
.ClockClassOffset(45, 354)
231 tc
= comp_self
._create
_trace
_class
()
235 _cc
, _tc
= run_in_component_init(f
)
237 _sc
= _tc
.create_stream_class(default_clock_class
=_cc
)
238 _ec
= _sc
.create_event_class(name
='salut')
239 _stream
= _trace
.create_stream(_sc
)
240 self
._stream
= _stream
244 class MyIter(bt2
._UserMessageIterator
):
245 def __init__(self
, self_port_output
):
250 notif
= self
._create
_stream
_beginning
_message
(_stream
)
252 notif
= self
._create
_event
_message
(_ec
, _stream
, 123)
254 notif
= self
._create
_event
_message
(_ec
, _stream
, 2 ** 63)
256 notif
= self
._create
_stream
_end
_message
(_stream
)
263 class MySrc(bt2
._UserSourceComponent
, message_iterator_class
=MyIter
):
264 def __init__(self
, params
, obj
):
265 self
._add
_output
_port
('out')
267 self
._graph
= bt2
.Graph()
268 self
._src
_comp
= self
._graph
.add_component(MySrc
, 'my_source')
269 self
._msg
_iter
= TestOutputPortMessageIterator(
270 self
._graph
, self
._src
_comp
.output_ports
['out']
273 for i
, msg
in enumerate(self
._msg
_iter
):
277 self
._msg
_clock
_overflow
= msg
284 def test_create_default(self
):
286 self
._msg
.default_clock_snapshot
.clock_class
.addr
, self
._cc
.addr
288 self
.assertEqual(self
._msg
.default_clock_snapshot
.value
, 123)
290 def test_clock_class(self
):
291 cc
= self
._msg
.default_clock_snapshot
.clock_class
292 self
.assertEqual(cc
.addr
, self
._cc
.addr
)
293 self
.assertIs(type(cc
), bt2_clock_class
._ClockClassConst
)
295 def test_ns_from_origin(self
):
296 s_from_origin
= 45 + ((354 + 123) / 1000)
297 ns_from_origin
= int(s_from_origin
* 1e9
)
299 self
._msg
.default_clock_snapshot
.ns_from_origin
, ns_from_origin
302 def test_ns_from_origin_overflow(self
):
303 with self
.assertRaises(bt2
._OverflowError
):
304 self
._msg
_clock
_overflow
.default_clock_snapshot
.ns_from_origin
306 def test_eq_int(self
):
307 self
.assertEqual(self
._msg
.default_clock_snapshot
, 123)
309 def test_eq_invalid(self
):
310 self
.assertFalse(self
._msg
.default_clock_snapshot
== 23)
312 def test_comparison(self
):
313 self
.assertTrue(self
._msg
.default_clock_snapshot
> 100)
314 self
.assertFalse(self
._msg
.default_clock_snapshot
> 200)
316 self
.assertTrue(self
._msg
.default_clock_snapshot
>= 123)
317 self
.assertFalse(self
._msg
.default_clock_snapshot
>= 200)
319 self
.assertTrue(self
._msg
.default_clock_snapshot
< 200)
320 self
.assertFalse(self
._msg
.default_clock_snapshot
< 100)
322 self
.assertTrue(self
._msg
.default_clock_snapshot
<= 123)
323 self
.assertFalse(self
._msg
.default_clock_snapshot
<= 100)