5 from utils
import run_in_component_init
8 class ClockClassOffsetTestCase(unittest
.TestCase
):
9 def test_create_default(self
):
10 cco
= bt2
.ClockClassOffset()
11 self
.assertEqual(cco
.seconds
, 0)
12 self
.assertEqual(cco
.cycles
, 0)
14 def test_create(self
):
15 cco
= bt2
.ClockClassOffset(23, 4871232)
16 self
.assertEqual(cco
.seconds
, 23)
17 self
.assertEqual(cco
.cycles
, 4871232)
19 def test_create_kwargs(self
):
20 cco
= bt2
.ClockClassOffset(seconds
=23, cycles
=4871232)
21 self
.assertEqual(cco
.seconds
, 23)
22 self
.assertEqual(cco
.cycles
, 4871232)
24 def test_create_invalid_seconds(self
):
25 with self
.assertRaises(TypeError):
26 bt2
.ClockClassOffset('hello', 4871232)
28 def test_create_invalid_cycles(self
):
29 with self
.assertRaises(TypeError):
30 bt2
.ClockClassOffset(23, 'hello')
33 cco1
= bt2
.ClockClassOffset(23, 42)
34 cco2
= bt2
.ClockClassOffset(23, 42)
35 self
.assertEqual(cco1
, cco2
)
37 def test_ne_seconds(self
):
38 cco1
= bt2
.ClockClassOffset(23, 42)
39 cco2
= bt2
.ClockClassOffset(24, 42)
40 self
.assertNotEqual(cco1
, cco2
)
42 def test_ne_cycles(self
):
43 cco1
= bt2
.ClockClassOffset(23, 42)
44 cco2
= bt2
.ClockClassOffset(23, 43)
45 self
.assertNotEqual(cco1
, cco2
)
47 def test_eq_invalid(self
):
48 self
.assertFalse(bt2
.ClockClassOffset() == 23)
51 class ClockClassTestCase(unittest
.TestCase
):
52 def assertRaisesInComponentInit(self
, expected_exc_type
, user_code
):
56 except Exception as exc
:
59 exc_type
= run_in_component_init(f
)
60 self
.assertIsNotNone(exc_type
)
61 self
.assertEqual(exc_type
, expected_exc_type
)
63 def test_create_default(self
):
64 cc
= run_in_component_init(lambda comp_self
: comp_self
._create
_clock
_class
())
66 self
.assertIsNone(cc
.name
)
67 self
.assertEqual(cc
.frequency
, 1000000000)
68 self
.assertIsNone(cc
.description
)
69 self
.assertEqual(cc
.precision
, 0)
70 self
.assertEqual(cc
.offset
, bt2
.ClockClassOffset())
71 self
.assertTrue(cc
.origin_is_unix_epoch
)
72 self
.assertIsNone(cc
.uuid
)
74 def test_create_name(self
):
76 return comp_self
._create
_clock
_class
(name
='the_clock')
78 cc
= run_in_component_init(f
)
79 self
.assertEqual(cc
.name
, 'the_clock')
81 def test_create_invalid_name(self
):
83 comp_self
._create
_clock
_class
(name
=23)
85 self
.assertRaisesInComponentInit(TypeError, f
)
87 def test_create_description(self
):
89 return comp_self
._create
_clock
_class
(description
='hi people')
91 cc
= run_in_component_init(f
)
92 self
.assertEqual(cc
.description
, 'hi people')
94 def test_create_invalid_description(self
):
96 return comp_self
._create
_clock
_class
(description
=23)
98 self
.assertRaisesInComponentInit(TypeError, f
)
100 def test_create_frequency(self
):
102 return comp_self
._create
_clock
_class
(frequency
=987654321)
104 cc
= run_in_component_init(f
)
105 self
.assertEqual(cc
.frequency
, 987654321)
107 def test_create_invalid_frequency(self
):
109 return comp_self
._create
_clock
_class
(frequency
='lel')
111 self
.assertRaisesInComponentInit(TypeError, f
)
113 def test_create_precision(self
):
115 return comp_self
._create
_clock
_class
(precision
=12)
117 cc
= run_in_component_init(f
)
118 self
.assertEqual(cc
.precision
, 12)
120 def test_create_invalid_precision(self
):
122 return comp_self
._create
_clock
_class
(precision
='lel')
124 self
.assertRaisesInComponentInit(TypeError, f
)
126 def test_create_offset(self
):
128 return comp_self
._create
_clock
_class
(offset
=bt2
.ClockClassOffset(12, 56))
130 cc
= run_in_component_init(f
)
131 self
.assertEqual(cc
.offset
, bt2
.ClockClassOffset(12, 56))
133 def test_create_invalid_offset(self
):
135 return comp_self
._create
_clock
_class
(offset
=object())
137 self
.assertRaisesInComponentInit(TypeError, f
)
139 def test_create_origin_is_unix_epoch(self
):
141 return comp_self
._create
_clock
_class
(origin_is_unix_epoch
=False)
143 cc
= run_in_component_init(f
)
144 self
.assertEqual(cc
.origin_is_unix_epoch
, False)
146 def test_create_invalid_origin_is_unix_epoch(self
):
148 return comp_self
._create
_clock
_class
(origin_is_unix_epoch
=23)
150 self
.assertRaisesInComponentInit(TypeError, f
)
152 def test_cycles_to_ns_from_origin(self
):
154 return comp_self
._create
_clock
_class
(frequency
=10**8, origin_is_unix_epoch
=True)
156 cc
= run_in_component_init(f
)
157 self
.assertEqual(cc
.cycles_to_ns_from_origin(112), 1120)
159 def test_cycles_to_ns_from_origin_overflow(self
):
161 return comp_self
._create
_clock
_class
(frequency
=1000)
163 cc
= run_in_component_init(f
)
164 with self
.assertRaises(OverflowError):
165 cc
.cycles_to_ns_from_origin(2**63)
167 def test_create_uuid(self
):
169 return comp_self
._create
_clock
_class
(uuid
=uuid
.UUID('b43372c32ef0be28444dfc1c5cdafd33'))
171 cc
= run_in_component_init(f
)
172 self
.assertEqual(cc
.uuid
, uuid
.UUID('b43372c32ef0be28444dfc1c5cdafd33'))
174 def test_create_invalid_uuid(self
):
176 return comp_self
._create
_clock
_class
(uuid
=23)
178 self
.assertRaisesInComponentInit(TypeError, f
)
181 class ClockSnapshotTestCase(unittest
.TestCase
):
184 cc
= comp_self
._create
_clock
_class
(1000, 'my_cc',
185 offset
=bt2
.ClockClassOffset(45, 354))
186 tc
= comp_self
._create
_trace
_class
()
190 _cc
, _tc
= run_in_component_init(f
)
192 _sc
= _tc
.create_stream_class(default_clock_class
=_cc
,
193 packets_have_beginning_default_clock_snapshot
=True,
194 packets_have_end_default_clock_snapshot
=True)
195 _ec
= _sc
.create_event_class(name
='salut')
196 _stream
= _trace
.create_stream(_sc
)
197 _packet
= _stream
.create_packet()
198 self
._packet
= _packet
199 self
._stream
= _stream
203 class MyIter(bt2
._UserMessageIterator
):
204 def __init__(self
, self_port_output
):
209 notif
= self
._create
_stream
_beginning
_message
(_stream
)
211 notif
= self
._create
_packet
_beginning
_message
(_packet
, 100)
213 notif
= self
._create
_event
_message
(_ec
, _packet
, 123)
215 notif
= self
._create
_event
_message
(_ec
, _packet
, 2**63)
217 notif
= self
._create
_packet
_end
_message
(_packet
)
219 notif
= self
._create
_stream
_end
_message
(_stream
)
226 class MySrc(bt2
._UserSourceComponent
, message_iterator_class
=MyIter
):
227 def __init__(self
, params
):
228 self
._add
_output
_port
('out')
230 self
._graph
= bt2
.Graph()
231 self
._src
_comp
= self
._graph
.add_component(MySrc
, 'my_source')
232 self
._msg
_iter
= self
._graph
.create_output_port_message_iterator(
233 self
._src
_comp
.output_ports
['out'])
235 for i
, msg
in enumerate(self
._msg
_iter
):
239 self
._msg
_clock
_overflow
= msg
246 def test_create_default(self
):
248 self
._msg
.default_clock_snapshot
.clock_class
.addr
, self
._cc
.addr
)
249 self
.assertEqual(self
._msg
.default_clock_snapshot
.value
, 123)
251 def test_clock_class(self
):
253 self
._msg
.default_clock_snapshot
.clock_class
.addr
, self
._cc
.addr
)
255 def test_ns_from_origin(self
):
256 s_from_origin
= 45 + ((354 + 123) / 1000)
257 ns_from_origin
= int(s_from_origin
* 1e9
)
259 self
._msg
.default_clock_snapshot
.ns_from_origin
, ns_from_origin
)
261 def test_ns_from_origin_overflow(self
):
262 with self
.assertRaises(OverflowError):
263 self
._msg
_clock
_overflow
.default_clock_snapshot
.ns_from_origin
265 def test_eq_int(self
):
266 self
.assertEqual(self
._msg
.default_clock_snapshot
, 123)
268 def test_eq_invalid(self
):
269 self
.assertFalse(self
._msg
.default_clock_snapshot
== 23)
271 def test_comparison(self
):
272 self
.assertTrue(self
._msg
.default_clock_snapshot
> 100)
273 self
.assertFalse(self
._msg
.default_clock_snapshot
> 200)
275 self
.assertTrue(self
._msg
.default_clock_snapshot
>= 123)
276 self
.assertFalse(self
._msg
.default_clock_snapshot
>= 200)
278 self
.assertTrue(self
._msg
.default_clock_snapshot
< 200)
279 self
.assertFalse(self
._msg
.default_clock_snapshot
< 100)
281 self
.assertTrue(self
._msg
.default_clock_snapshot
<= 123)
282 self
.assertFalse(self
._msg
.default_clock_snapshot
<= 100)