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 TestOutputPortMessageIterator
23 from bt2
import port
as bt2_port
26 class SimpleSink(bt2
._UserSinkComponent
):
27 # Straightforward sink that creates one input port (`in`) and consumes from
30 def __init__(self
, config
, params
, obj
):
31 self
._add
_input
_port
('in')
33 def _user_consume(self
):
36 def _user_graph_is_configured(self
):
37 self
._msg
_iter
= self
._create
_input
_port
_message
_iterator
(
38 self
._input
_ports
['in']
42 def _create_graph(src_comp_cls
, sink_comp_cls
, flt_comp_cls
=None):
45 src_comp
= graph
.add_component(src_comp_cls
, 'src')
46 sink_comp
= graph
.add_component(sink_comp_cls
, 'sink')
48 if flt_comp_cls
is not None:
49 flt_comp
= graph
.add_component(flt_comp_cls
, 'flt')
50 graph
.connect_ports(src_comp
.output_ports
['out'], flt_comp
.input_ports
['in'])
51 graph
.connect_ports(flt_comp
.output_ports
['out'], sink_comp
.input_ports
['in'])
53 graph
.connect_ports(src_comp
.output_ports
['out'], sink_comp
.input_ports
['in'])
58 class UserMessageIteratorTestCase(unittest
.TestCase
):
60 the_output_port_from_source
= None
61 the_output_port_from_iter
= None
63 class MyIter(bt2
._UserMessageIterator
):
64 def __init__(self
, self_port_output
):
66 nonlocal the_output_port_from_iter
68 the_output_port_from_iter
= self_port_output
70 class MySource(bt2
._UserSourceComponent
, message_iterator_class
=MyIter
):
71 def __init__(self
, config
, params
, obj
):
72 nonlocal the_output_port_from_source
73 the_output_port_from_source
= self
._add
_output
_port
('out', 'user data')
76 graph
= _create_graph(MySource
, SimpleSink
)
78 self
.assertTrue(initialized
)
80 the_output_port_from_source
.addr
, the_output_port_from_iter
.addr
82 self
.assertEqual(the_output_port_from_iter
.user_data
, 'user data')
84 def test_create_from_message_iterator(self
):
85 class MySourceIter(bt2
._UserMessageIterator
):
86 def __init__(self
, self_port_output
):
87 nonlocal src_iter_initialized
88 src_iter_initialized
= True
90 class MySource(bt2
._UserSourceComponent
, message_iterator_class
=MySourceIter
):
91 def __init__(self
, config
, params
, obj
):
92 self
._add
_output
_port
('out')
94 class MyFilterIter(bt2
._UserMessageIterator
):
95 def __init__(self
, self_port_output
):
96 nonlocal flt_iter_initialized
97 flt_iter_initialized
= True
98 self
._up
_iter
= self
._create
_input
_port
_message
_iterator
(
99 self
._component
._input
_ports
['in']
103 return next(self
._up
_iter
)
105 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyFilterIter
):
106 def __init__(self
, config
, params
, obj
):
107 self
._add
_input
_port
('in')
108 self
._add
_output
_port
('out')
110 src_iter_initialized
= False
111 flt_iter_initialized
= False
112 graph
= _create_graph(MySource
, SimpleSink
, MyFilter
)
114 self
.assertTrue(src_iter_initialized
)
115 self
.assertTrue(flt_iter_initialized
)
117 def test_create_user_error(self
):
118 # This tests both error handling by
119 # _UserSinkComponent._create_input_port_message_iterator
120 # and _UserMessageIterator._create_input_port_message_iterator, as they
121 # are both used in the graph.
122 class MySourceIter(bt2
._UserMessageIterator
):
123 def __init__(self
, self_port_output
):
124 raise ValueError('Very bad error')
126 class MySource(bt2
._UserSourceComponent
, message_iterator_class
=MySourceIter
):
127 def __init__(self
, config
, params
, obj
):
128 self
._add
_output
_port
('out')
130 class MyFilterIter(bt2
._UserMessageIterator
):
131 def __init__(self
, self_port_output
):
132 # This is expected to raise because of the error in
133 # MySourceIter.__init__.
134 self
._create
_input
_port
_message
_iterator
(
135 self
._component
._input
_ports
['in']
138 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyFilterIter
):
139 def __init__(self
, config
, params
, obj
):
140 self
._add
_input
_port
('in')
141 self
._add
_output
_port
('out')
143 graph
= _create_graph(MySource
, SimpleSink
, MyFilter
)
145 with self
.assertRaises(bt2
._Error
) as ctx
:
151 self
.assertIsInstance(cause
, bt2
._MessageIteratorErrorCause
)
152 self
.assertEqual(cause
.component_name
, 'src')
153 self
.assertEqual(cause
.component_output_port_name
, 'out')
154 self
.assertIn('ValueError: Very bad error', cause
.message
)
156 def test_finalize(self
):
157 class MyIter(bt2
._UserMessageIterator
):
158 def _user_finalize(self
):
162 class MySource(bt2
._UserSourceComponent
, message_iterator_class
=MyIter
):
163 def __init__(self
, config
, params
, obj
):
164 self
._add
_output
_port
('out')
167 graph
= _create_graph(MySource
, SimpleSink
)
170 self
.assertTrue(finalized
)
172 def test_component(self
):
173 class MyIter(bt2
._UserMessageIterator
):
174 def __init__(self
, self_port_output
):
176 salut
= self
._component
._salut
178 class MySource(bt2
._UserSourceComponent
, message_iterator_class
=MyIter
):
179 def __init__(self
, config
, params
, obj
):
180 self
._add
_output
_port
('out')
184 graph
= _create_graph(MySource
, SimpleSink
)
186 self
.assertEqual(salut
, 23)
189 class MyIter(bt2
._UserMessageIterator
):
190 def __init__(self_iter
, self_port_output
):
193 port
= self_iter
._port
194 self
.assertIs(type(self_port_output
), bt2_port
._UserComponentOutputPort
)
195 self
.assertIs(type(port
), bt2_port
._UserComponentOutputPort
)
196 self
.assertEqual(self_port_output
.addr
, port
.addr
)
198 class MySource(bt2
._UserSourceComponent
, message_iterator_class
=MyIter
):
199 def __init__(self
, config
, params
, obj
):
200 self
._add
_output
_port
('out')
203 graph
= _create_graph(MySource
, SimpleSink
)
205 self
.assertTrue(called
)
208 class MyIter(bt2
._UserMessageIterator
):
209 def __init__(self
, self_port_output
):
213 class MySource(bt2
._UserSourceComponent
, message_iterator_class
=MyIter
):
214 def __init__(self
, config
, params
, obj
):
215 self
._add
_output
_port
('out')
218 graph
= _create_graph(MySource
, SimpleSink
)
220 self
.assertIsNotNone(addr
)
221 self
.assertNotEqual(addr
, 0)
223 # Test that messages returned by _UserMessageIterator.__next__ remain valid
224 # and can be re-used.
225 def test_reuse_message(self
):
226 class MyIter(bt2
._UserMessageIterator
):
227 def __init__(self
, port
):
228 tc
, sc
, ec
= port
.user_data
230 stream
= trace
.create_stream(sc
)
231 packet
= stream
.create_packet()
233 # This message will be returned twice by __next__.
234 event_message
= self
._create
_event
_message
(ec
, packet
)
237 self
._create
_stream
_beginning
_message
(stream
),
238 self
._create
_packet
_beginning
_message
(packet
),
244 return self
._msgs
.pop(0)
246 class MySource(bt2
._UserSourceComponent
, message_iterator_class
=MyIter
):
247 def __init__(self
, config
, params
, obj
):
248 tc
= self
._create
_trace
_class
()
249 sc
= tc
.create_stream_class(supports_packets
=True)
250 ec
= sc
.create_event_class()
251 self
._add
_output
_port
('out', (tc
, sc
, ec
))
254 src
= graph
.add_component(MySource
, 'src')
255 it
= TestOutputPortMessageIterator(graph
, src
.output_ports
['out'])
257 # Skip beginning messages.
259 self
.assertIs(type(msg
), bt2
._StreamBeginningMessageConst
)
261 self
.assertIs(type(msg
), bt2
._PacketBeginningMessageConst
)
266 self
.assertIs(type(msg_ev1
), bt2
._EventMessageConst
)
267 self
.assertIs(type(msg_ev2
), bt2
._EventMessageConst
)
268 self
.assertEqual(msg_ev1
.addr
, msg_ev2
.addr
)
270 # Try consuming many times from an iterator that always returns TryAgain.
271 # This verifies that we are not missing an incref of Py_None, making the
272 # refcount of Py_None reach 0.
273 def test_try_again_many_times(self
):
274 class MyIter(bt2
._UserMessageIterator
):
278 class MySource(bt2
._UserSourceComponent
, message_iterator_class
=MyIter
):
279 def __init__(self
, config
, params
, obj
):
280 self
._add
_output
_port
('out')
282 class MyFilterIter(bt2
._UserMessageIterator
):
283 def __init__(self
, port
):
284 input_port
= port
.user_data
285 self
._upstream
_iter
= self
._create
_input
_port
_message
_iterator
(
290 return next(self
._upstream
_iter
)
292 def _user_seek_beginning(self
):
293 self
._upstream
_iter
.seek_beginning()
295 def _user_can_seek_beginning(self
):
296 return self
._upstream
_iter
.can_seek_beginning()
298 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyFilterIter
):
299 def __init__(self
, config
, params
, obj
):
300 input_port
= self
._add
_input
_port
('in')
301 self
._add
_output
_port
('out', input_port
)
304 src
= graph
.add_component(MySource
, 'src')
305 it
= TestOutputPortMessageIterator(graph
, src
.output_ports
['out'])
307 # Three times the initial ref count of `None` iterations should
308 # be enough to catch the bug even if there are small differences
309 # between configurations.
310 none_ref_count
= sys
.getrefcount(None) * 3
312 for i
in range(none_ref_count
):
313 with self
.assertRaises(bt2
.TryAgain
):
317 def _setup_seek_test(
319 user_seek_beginning
=None,
320 user_can_seek_beginning
=None,
321 user_seek_ns_from_origin
=None,
322 user_can_seek_ns_from_origin
=None,
324 class MySourceIter(bt2
._UserMessageIterator
):
325 def __init__(self
, port
):
326 tc
, sc
, ec
= port
.user_data
328 stream
= trace
.create_stream(sc
)
329 packet
= stream
.create_packet()
332 self
._create
_stream
_beginning
_message
(stream
),
333 self
._create
_packet
_beginning
_message
(packet
),
334 self
._create
_event
_message
(ec
, packet
),
335 self
._create
_event
_message
(ec
, packet
),
336 self
._create
_packet
_end
_message
(packet
),
337 self
._create
_stream
_end
_message
(stream
),
342 if self
._at
< len(self
._msgs
):
343 msg
= self
._msgs
[self
._at
]
349 if user_seek_beginning
is not None:
350 MySourceIter
._user
_seek
_beginning
= user_seek_beginning
352 if user_can_seek_beginning
is not None:
353 MySourceIter
._user
_can
_seek
_beginning
= user_can_seek_beginning
355 if user_seek_ns_from_origin
is not None:
356 MySourceIter
._user
_seek
_ns
_from
_origin
= user_seek_ns_from_origin
358 if user_can_seek_ns_from_origin
is not None:
359 MySourceIter
._user
_can
_seek
_ns
_from
_origin
= user_can_seek_ns_from_origin
361 class MySource(bt2
._UserSourceComponent
, message_iterator_class
=MySourceIter
):
362 def __init__(self
, config
, params
, obj
):
363 tc
= self
._create
_trace
_class
()
364 sc
= tc
.create_stream_class(supports_packets
=True)
365 ec
= sc
.create_event_class()
367 self
._add
_output
_port
('out', (tc
, sc
, ec
))
369 class MyFilterIter(bt2
._UserMessageIterator
):
370 def __init__(self
, port
):
371 self
._upstream
_iter
= self
._create
_input
_port
_message
_iterator
(
372 self
._component
._input
_ports
['in']
376 return next(self
._upstream
_iter
)
378 def _user_can_seek_beginning(self
):
379 return self
._upstream
_iter
.can_seek_beginning()
381 def _user_seek_beginning(self
):
382 self
._upstream
_iter
.seek_beginning()
384 def _user_can_seek_ns_from_origin(self
, ns_from_origin
):
385 return self
._upstream
_iter
.can_seek_ns_from_origin(ns_from_origin
)
387 def _user_seek_ns_from_origin(self
, ns_from_origin
):
388 self
._upstream
_iter
.seek_ns_from_origin(ns_from_origin
)
390 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyFilterIter
):
391 def __init__(self
, config
, params
, obj
):
392 self
._add
_input
_port
('in')
393 self
._add
_output
_port
('out')
395 return _create_graph(MySource
, sink_cls
, flt_comp_cls
=MyFilter
)
398 class UserMessageIteratorSeekBeginningTestCase(unittest
.TestCase
):
399 def test_can_seek_beginning(self
):
400 class MySink(bt2
._UserSinkComponent
):
401 def __init__(self
, config
, params
, obj
):
402 self
._add
_input
_port
('in')
404 def _user_graph_is_configured(self
):
405 self
._msg
_iter
= self
._create
_input
_port
_message
_iterator
(
406 self
._input
_ports
['in']
409 def _user_consume(self
):
410 nonlocal can_seek_beginning
411 can_seek_beginning
= self
._msg
_iter
.can_seek_beginning()
413 def _user_can_seek_beginning(self
):
414 nonlocal input_port_iter_can_seek_beginning
415 return input_port_iter_can_seek_beginning
417 graph
= _setup_seek_test(
418 MySink
, user_can_seek_beginning
=_user_can_seek_beginning
421 input_port_iter_can_seek_beginning
= True
422 can_seek_beginning
= None
424 self
.assertIs(can_seek_beginning
, True)
426 input_port_iter_can_seek_beginning
= False
427 can_seek_beginning
= None
429 self
.assertIs(can_seek_beginning
, False)
431 def test_no_can_seek_beginning_with_seek_beginning(self
):
432 # Test an iterator without a _user_can_seek_beginning method, but with
433 # a _user_seek_beginning method.
434 class MySink(bt2
._UserSinkComponent
):
435 def __init__(self
, config
, params
, obj
):
436 self
._add
_input
_port
('in')
438 def _user_graph_is_configured(self
):
439 self
._msg
_iter
= self
._create
_input
_port
_message
_iterator
(
440 self
._input
_ports
['in']
443 def _user_consume(self
):
444 nonlocal can_seek_beginning
445 can_seek_beginning
= self
._msg
_iter
.can_seek_beginning()
447 def _user_seek_beginning(self
):
450 graph
= _setup_seek_test(MySink
, user_seek_beginning
=_user_seek_beginning
)
451 can_seek_beginning
= None
453 self
.assertIs(can_seek_beginning
, True)
455 def test_no_can_seek_beginning(self
):
456 # Test an iterator without a _user_can_seek_beginning method, without
457 # a _user_seek_beginning method.
458 class MySink(bt2
._UserSinkComponent
):
459 def __init__(self
, config
, params
, obj
):
460 self
._add
_input
_port
('in')
462 def _user_graph_is_configured(self
):
463 self
._msg
_iter
= self
._create
_input
_port
_message
_iterator
(
464 self
._input
_ports
['in']
467 def _user_consume(self
):
468 nonlocal can_seek_beginning
469 can_seek_beginning
= self
._msg
_iter
.can_seek_beginning()
471 graph
= _setup_seek_test(MySink
)
472 can_seek_beginning
= None
474 self
.assertIs(can_seek_beginning
, False)
476 def test_can_seek_beginning_user_error(self
):
477 class MySink(bt2
._UserSinkComponent
):
478 def __init__(self
, config
, params
, obj
):
479 self
._add
_input
_port
('in')
481 def _user_graph_is_configured(self
):
482 self
._msg
_iter
= self
._create
_input
_port
_message
_iterator
(
483 self
._input
_ports
['in']
486 def _user_consume(self
):
487 # This is expected to raise.
488 self
._msg
_iter
.can_seek_beginning()
490 def _user_can_seek_beginning(self
):
491 raise ValueError('moustiquaire')
493 graph
= _setup_seek_test(
494 MySink
, user_can_seek_beginning
=_user_can_seek_beginning
497 with self
.assertRaises(bt2
._Error
) as ctx
:
500 cause
= ctx
.exception
[0]
501 self
.assertIn('ValueError: moustiquaire', cause
.message
)
503 def test_can_seek_beginning_wrong_return_value(self
):
504 class MySink(bt2
._UserSinkComponent
):
505 def __init__(self
, config
, params
, obj
):
506 self
._add
_input
_port
('in')
508 def _user_graph_is_configured(self
):
509 self
._msg
_iter
= self
._create
_input
_port
_message
_iterator
(
510 self
._input
_ports
['in']
513 def _user_consume(self
):
514 # This is expected to raise.
515 self
._msg
_iter
.can_seek_beginning()
517 def _user_can_seek_beginning(self
):
520 graph
= _setup_seek_test(
521 MySink
, user_can_seek_beginning
=_user_can_seek_beginning
524 with self
.assertRaises(bt2
._Error
) as ctx
:
527 cause
= ctx
.exception
[0]
528 self
.assertIn("TypeError: 'str' is not a 'bool' object", cause
.message
)
530 def test_seek_beginning(self
):
531 class MySink(bt2
._UserSinkComponent
):
532 def __init__(self
, config
, params
, obj
):
533 self
._add
_input
_port
('in')
535 def _user_graph_is_configured(self
):
536 self
._msg
_iter
= self
._create
_input
_port
_message
_iterator
(
537 self
._input
_ports
['in']
540 def _user_consume(self
):
541 nonlocal do_seek_beginning
544 if do_seek_beginning
:
545 self
._msg
_iter
.seek_beginning()
548 msg
= next(self
._msg
_iter
)
550 def _user_seek_beginning(self
):
554 graph
= _setup_seek_test(MySink
, user_seek_beginning
=_user_seek_beginning
)
557 do_seek_beginning
= False
559 self
.assertIs(type(msg
), bt2
._StreamBeginningMessageConst
)
563 self
.assertIs(type(msg
), bt2
._PacketBeginningMessageConst
)
566 do_seek_beginning
= True
570 do_seek_beginning
= False
572 self
.assertIs(type(msg
), bt2
._StreamBeginningMessageConst
)
574 def test_seek_beginning_user_error(self
):
575 class MySink(bt2
._UserSinkComponent
):
576 def __init__(self
, config
, params
, obj
):
577 self
._add
_input
_port
('in')
579 def _user_graph_is_configured(self
):
580 self
._msg
_iter
= self
._create
_input
_port
_message
_iterator
(
581 self
._input
_ports
['in']
584 def _user_consume(self
):
585 self
._msg
_iter
.seek_beginning()
587 def _user_seek_beginning(self
):
588 raise ValueError('ouch')
590 graph
= _setup_seek_test(MySink
, user_seek_beginning
=_user_seek_beginning
)
592 with self
.assertRaises(bt2
._Error
):
596 class UserMessageIteratorSeekNsFromOriginTestCase(unittest
.TestCase
):
597 def test_can_seek_ns_from_origin(self
):
598 class MySink(bt2
._UserSinkComponent
):
599 def __init__(self
, config
, params
, obj
):
600 self
._add
_input
_port
('in')
602 def _user_graph_is_configured(self
):
603 self
._msg
_iter
= self
._create
_input
_port
_message
_iterator
(
604 self
._input
_ports
['in']
607 def _user_consume(self
):
608 nonlocal can_seek_ns_from_origin
609 nonlocal test_ns_from_origin
610 can_seek_ns_from_origin
= self
._msg
_iter
.can_seek_ns_from_origin(
614 def _user_can_seek_ns_from_origin(iter_self
, ns_from_origin
):
615 nonlocal input_port_iter_can_seek_ns_from_origin
616 nonlocal test_ns_from_origin
617 self
.assertEqual(ns_from_origin
, test_ns_from_origin
)
618 return input_port_iter_can_seek_ns_from_origin
620 graph
= _setup_seek_test(
621 MySink
, user_can_seek_ns_from_origin
=_user_can_seek_ns_from_origin
624 input_port_iter_can_seek_ns_from_origin
= True
625 can_seek_ns_from_origin
= None
626 test_ns_from_origin
= 1
628 self
.assertIs(can_seek_ns_from_origin
, True)
630 input_port_iter_can_seek_ns_from_origin
= False
631 can_seek_ns_from_origin
= None
632 test_ns_from_origin
= 2
634 self
.assertIs(can_seek_ns_from_origin
, False)
636 def test_no_can_seek_ns_from_origin_with_seek_ns_from_origin(self
):
637 # Test an iterator without a _user_can_seek_ns_from_origin method, but
638 # with a _user_seek_ns_from_origin method.
639 class MySink(bt2
._UserSinkComponent
):
640 def __init__(self
, config
, params
, obj
):
641 self
._add
_input
_port
('in')
643 def _user_graph_is_configured(self
):
644 self
._msg
_iter
= self
._create
_input
_port
_message
_iterator
(
645 self
._input
_ports
['in']
648 def _user_consume(self
):
649 nonlocal can_seek_ns_from_origin
650 nonlocal test_ns_from_origin
651 can_seek_ns_from_origin
= self
._msg
_iter
.can_seek_ns_from_origin(
655 def _user_seek_ns_from_origin(self
):
658 graph
= _setup_seek_test(
659 MySink
, user_seek_ns_from_origin
=_user_seek_ns_from_origin
661 can_seek_ns_from_origin
= None
662 test_ns_from_origin
= 2
664 self
.assertIs(can_seek_ns_from_origin
, True)
666 def test_no_can_seek_ns_from_origin_with_seek_beginning(self
):
667 # Test an iterator without a _user_can_seek_ns_from_origin method, but
668 # with a _user_seek_beginning method.
669 class MySink(bt2
._UserSinkComponent
):
670 def __init__(self
, config
, params
, obj
):
671 self
._add
_input
_port
('in')
673 def _user_graph_is_configured(self
):
674 self
._msg
_iter
= self
._create
_input
_port
_message
_iterator
(
675 self
._input
_ports
['in']
678 def _user_consume(self
):
679 nonlocal can_seek_ns_from_origin
680 nonlocal test_ns_from_origin
681 can_seek_ns_from_origin
= self
._msg
_iter
.can_seek_ns_from_origin(
685 def _user_seek_beginning(self
):
688 graph
= _setup_seek_test(MySink
, user_seek_beginning
=_user_seek_beginning
)
689 can_seek_ns_from_origin
= None
690 test_ns_from_origin
= 2
692 self
.assertIs(can_seek_ns_from_origin
, True)
694 def test_no_can_seek_ns_from_origin(self
):
695 # Test an iterator without a _user_can_seek_ns_from_origin method
696 # and no other related method.
697 class MySink(bt2
._UserSinkComponent
):
698 def __init__(self
, config
, params
, obj
):
699 self
._add
_input
_port
('in')
701 def _user_graph_is_configured(self
):
702 self
._msg
_iter
= self
._create
_input
_port
_message
_iterator
(
703 self
._input
_ports
['in']
706 def _user_consume(self
):
707 nonlocal can_seek_ns_from_origin
708 nonlocal test_ns_from_origin
709 can_seek_ns_from_origin
= self
._msg
_iter
.can_seek_ns_from_origin(
713 graph
= _setup_seek_test(MySink
)
714 can_seek_ns_from_origin
= None
715 test_ns_from_origin
= 2
717 self
.assertIs(can_seek_ns_from_origin
, False)
719 def test_can_seek_ns_from_origin_user_error(self
):
720 class MySink(bt2
._UserSinkComponent
):
721 def __init__(self
, config
, params
, obj
):
722 self
._add
_input
_port
('in')
724 def _user_graph_is_configured(self
):
725 self
._msg
_iter
= self
._create
_input
_port
_message
_iterator
(
726 self
._input
_ports
['in']
729 def _user_consume(self
):
730 # This is expected to raise.
731 self
._msg
_iter
.can_seek_ns_from_origin(2)
733 def _user_can_seek_ns_from_origin(self
, ns_from_origin
):
734 raise ValueError('Joutel')
736 graph
= _setup_seek_test(
737 MySink
, user_can_seek_ns_from_origin
=_user_can_seek_ns_from_origin
740 with self
.assertRaises(bt2
._Error
) as ctx
:
743 cause
= ctx
.exception
[0]
744 self
.assertIn('ValueError: Joutel', cause
.message
)
746 def test_can_seek_ns_from_origin_wrong_return_value(self
):
747 class MySink(bt2
._UserSinkComponent
):
748 def __init__(self
, config
, params
, obj
):
749 self
._add
_input
_port
('in')
751 def _user_graph_is_configured(self
):
752 self
._msg
_iter
= self
._create
_input
_port
_message
_iterator
(
753 self
._input
_ports
['in']
756 def _user_consume(self
):
757 # This is expected to raise.
758 self
._msg
_iter
.can_seek_ns_from_origin(2)
760 def _user_can_seek_ns_from_origin(self
, ns_from_origin
):
763 graph
= _setup_seek_test(
764 MySink
, user_can_seek_ns_from_origin
=_user_can_seek_ns_from_origin
767 with self
.assertRaises(bt2
._Error
) as ctx
:
770 cause
= ctx
.exception
[0]
771 self
.assertIn("TypeError: 'str' is not a 'bool' object", cause
.message
)
773 def test_seek_ns_from_origin(self
):
774 class MySink(bt2
._UserSinkComponent
):
775 def __init__(self
, config
, params
, obj
):
776 self
._add
_input
_port
('in')
778 def _user_graph_is_configured(self
):
779 self
._msg
_iter
= self
._create
_input
_port
_message
_iterator
(
780 self
._input
_ports
['in']
783 def _user_consume(self
):
784 self
._msg
_iter
.seek_ns_from_origin(17)
786 def _user_seek_ns_from_origin(self
, ns_from_origin
):
787 nonlocal actual_ns_from_origin
788 actual_ns_from_origin
= ns_from_origin
791 graph
= _setup_seek_test(
792 MySink
, user_seek_ns_from_origin
=_user_seek_ns_from_origin
795 actual_ns_from_origin
= None
797 self
.assertEqual(actual_ns_from_origin
, 17)
800 if __name__
== '__main__':