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.
23 from utils
import run_in_component_init
26 class ClockClassOffsetTestCase(unittest
.TestCase
):
27 def test_create_default(self
):
28 cco
= bt2
.ClockClassOffset()
29 self
.assertEqual(cco
.seconds
, 0)
30 self
.assertEqual(cco
.cycles
, 0)
32 def test_create(self
):
33 cco
= bt2
.ClockClassOffset(23, 4871232)
34 self
.assertEqual(cco
.seconds
, 23)
35 self
.assertEqual(cco
.cycles
, 4871232)
37 def test_create_kwargs(self
):
38 cco
= bt2
.ClockClassOffset(seconds
=23, cycles
=4871232)
39 self
.assertEqual(cco
.seconds
, 23)
40 self
.assertEqual(cco
.cycles
, 4871232)
42 def test_create_invalid_seconds(self
):
43 with self
.assertRaises(TypeError):
44 bt2
.ClockClassOffset('hello', 4871232)
46 def test_create_invalid_cycles(self
):
47 with self
.assertRaises(TypeError):
48 bt2
.ClockClassOffset(23, 'hello')
51 cco1
= bt2
.ClockClassOffset(23, 42)
52 cco2
= bt2
.ClockClassOffset(23, 42)
53 self
.assertEqual(cco1
, cco2
)
55 def test_ne_seconds(self
):
56 cco1
= bt2
.ClockClassOffset(23, 42)
57 cco2
= bt2
.ClockClassOffset(24, 42)
58 self
.assertNotEqual(cco1
, cco2
)
60 def test_ne_cycles(self
):
61 cco1
= bt2
.ClockClassOffset(23, 42)
62 cco2
= bt2
.ClockClassOffset(23, 43)
63 self
.assertNotEqual(cco1
, cco2
)
65 def test_eq_invalid(self
):
66 self
.assertFalse(bt2
.ClockClassOffset() == 23)
69 class ClockClassTestCase(unittest
.TestCase
):
70 def assertRaisesInComponentInit(self
, expected_exc_type
, user_code
):
74 except Exception as exc
:
77 exc_type
= run_in_component_init(f
)
78 self
.assertIsNotNone(exc_type
)
79 self
.assertEqual(exc_type
, expected_exc_type
)
81 def test_create_default(self
):
82 cc
= run_in_component_init(lambda comp_self
: comp_self
._create
_clock
_class
())
84 self
.assertIsNone(cc
.name
)
85 self
.assertEqual(cc
.frequency
, 1000000000)
86 self
.assertIsNone(cc
.description
)
87 self
.assertEqual(cc
.precision
, 0)
88 self
.assertEqual(cc
.offset
, bt2
.ClockClassOffset())
89 self
.assertTrue(cc
.origin_is_unix_epoch
)
90 self
.assertIsNone(cc
.uuid
)
92 def test_create_name(self
):
94 return comp_self
._create
_clock
_class
(name
='the_clock')
96 cc
= run_in_component_init(f
)
97 self
.assertEqual(cc
.name
, 'the_clock')
99 def test_create_invalid_name(self
):
101 comp_self
._create
_clock
_class
(name
=23)
103 self
.assertRaisesInComponentInit(TypeError, f
)
105 def test_create_description(self
):
107 return comp_self
._create
_clock
_class
(description
='hi people')
109 cc
= run_in_component_init(f
)
110 self
.assertEqual(cc
.description
, 'hi people')
112 def test_create_invalid_description(self
):
114 return comp_self
._create
_clock
_class
(description
=23)
116 self
.assertRaisesInComponentInit(TypeError, f
)
118 def test_create_frequency(self
):
120 return comp_self
._create
_clock
_class
(frequency
=987654321)
122 cc
= run_in_component_init(f
)
123 self
.assertEqual(cc
.frequency
, 987654321)
125 def test_create_invalid_frequency(self
):
127 return comp_self
._create
_clock
_class
(frequency
='lel')
129 self
.assertRaisesInComponentInit(TypeError, f
)
131 def test_create_precision(self
):
133 return comp_self
._create
_clock
_class
(precision
=12)
135 cc
= run_in_component_init(f
)
136 self
.assertEqual(cc
.precision
, 12)
138 def test_create_invalid_precision(self
):
140 return comp_self
._create
_clock
_class
(precision
='lel')
142 self
.assertRaisesInComponentInit(TypeError, f
)
144 def test_create_offset(self
):
146 return comp_self
._create
_clock
_class
(offset
=bt2
.ClockClassOffset(12, 56))
148 cc
= run_in_component_init(f
)
149 self
.assertEqual(cc
.offset
, bt2
.ClockClassOffset(12, 56))
151 def test_create_invalid_offset(self
):
153 return comp_self
._create
_clock
_class
(offset
=object())
155 self
.assertRaisesInComponentInit(TypeError, f
)
157 def test_create_origin_is_unix_epoch(self
):
159 return comp_self
._create
_clock
_class
(origin_is_unix_epoch
=False)
161 cc
= run_in_component_init(f
)
162 self
.assertEqual(cc
.origin_is_unix_epoch
, False)
164 def test_create_invalid_origin_is_unix_epoch(self
):
166 return comp_self
._create
_clock
_class
(origin_is_unix_epoch
=23)
168 self
.assertRaisesInComponentInit(TypeError, f
)
170 def test_cycles_to_ns_from_origin(self
):
172 return comp_self
._create
_clock
_class
(frequency
=10**8, origin_is_unix_epoch
=True)
174 cc
= run_in_component_init(f
)
175 self
.assertEqual(cc
.cycles_to_ns_from_origin(112), 1120)
177 def test_cycles_to_ns_from_origin_overflow(self
):
179 return comp_self
._create
_clock
_class
(frequency
=1000)
181 cc
= run_in_component_init(f
)
182 with self
.assertRaises(bt2
.OverflowError):
183 cc
.cycles_to_ns_from_origin(2**63)
185 def test_create_uuid(self
):
187 return comp_self
._create
_clock
_class
(uuid
=uuid
.UUID('b43372c32ef0be28444dfc1c5cdafd33'))
189 cc
= run_in_component_init(f
)
190 self
.assertEqual(cc
.uuid
, uuid
.UUID('b43372c32ef0be28444dfc1c5cdafd33'))
192 def test_create_invalid_uuid(self
):
194 return comp_self
._create
_clock
_class
(uuid
=23)
196 self
.assertRaisesInComponentInit(TypeError, f
)
199 class ClockSnapshotTestCase(unittest
.TestCase
):
202 cc
= comp_self
._create
_clock
_class
(1000, 'my_cc',
203 offset
=bt2
.ClockClassOffset(45, 354))
204 tc
= comp_self
._create
_trace
_class
()
208 _cc
, _tc
= run_in_component_init(f
)
210 _sc
= _tc
.create_stream_class(default_clock_class
=_cc
,
211 packets_have_beginning_default_clock_snapshot
=True,
212 packets_have_end_default_clock_snapshot
=True)
213 _ec
= _sc
.create_event_class(name
='salut')
214 _stream
= _trace
.create_stream(_sc
)
215 _packet
= _stream
.create_packet()
216 self
._packet
= _packet
217 self
._stream
= _stream
221 class MyIter(bt2
._UserMessageIterator
):
222 def __init__(self
, self_port_output
):
227 notif
= self
._create
_stream
_beginning
_message
(_stream
)
229 notif
= self
._create
_packet
_beginning
_message
(_packet
, 100)
231 notif
= self
._create
_event
_message
(_ec
, _packet
, 123)
233 notif
= self
._create
_event
_message
(_ec
, _packet
, 2**63)
235 notif
= self
._create
_packet
_end
_message
(_packet
)
237 notif
= self
._create
_stream
_end
_message
(_stream
)
244 class MySrc(bt2
._UserSourceComponent
, message_iterator_class
=MyIter
):
245 def __init__(self
, params
):
246 self
._add
_output
_port
('out')
248 self
._graph
= bt2
.Graph()
249 self
._src
_comp
= self
._graph
.add_component(MySrc
, 'my_source')
250 self
._msg
_iter
= self
._graph
.create_output_port_message_iterator(
251 self
._src
_comp
.output_ports
['out'])
253 for i
, msg
in enumerate(self
._msg
_iter
):
257 self
._msg
_clock
_overflow
= msg
264 def test_create_default(self
):
266 self
._msg
.default_clock_snapshot
.clock_class
.addr
, self
._cc
.addr
)
267 self
.assertEqual(self
._msg
.default_clock_snapshot
.value
, 123)
269 def test_clock_class(self
):
271 self
._msg
.default_clock_snapshot
.clock_class
.addr
, self
._cc
.addr
)
273 def test_ns_from_origin(self
):
274 s_from_origin
= 45 + ((354 + 123) / 1000)
275 ns_from_origin
= int(s_from_origin
* 1e9
)
277 self
._msg
.default_clock_snapshot
.ns_from_origin
, ns_from_origin
)
279 def test_ns_from_origin_overflow(self
):
280 with self
.assertRaises(bt2
.OverflowError):
281 self
._msg
_clock
_overflow
.default_clock_snapshot
.ns_from_origin
283 def test_eq_int(self
):
284 self
.assertEqual(self
._msg
.default_clock_snapshot
, 123)
286 def test_eq_invalid(self
):
287 self
.assertFalse(self
._msg
.default_clock_snapshot
== 23)
289 def test_comparison(self
):
290 self
.assertTrue(self
._msg
.default_clock_snapshot
> 100)
291 self
.assertFalse(self
._msg
.default_clock_snapshot
> 200)
293 self
.assertTrue(self
._msg
.default_clock_snapshot
>= 123)
294 self
.assertFalse(self
._msg
.default_clock_snapshot
>= 200)
296 self
.assertTrue(self
._msg
.default_clock_snapshot
< 200)
297 self
.assertFalse(self
._msg
.default_clock_snapshot
< 100)
299 self
.assertTrue(self
._msg
.default_clock_snapshot
<= 123)
300 self
.assertFalse(self
._msg
.default_clock_snapshot
<= 100)