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
24 from bt2
import message_iterator
as bt2_message_iterator
27 class SimpleSink(bt2
._UserSinkComponent
):
28 # Straightforward sink that creates one input port (`in`) and consumes from
31 def __init__(self
, config
, params
, obj
):
32 self
._add
_input
_port
('in')
34 def _user_consume(self
):
37 def _user_graph_is_configured(self
):
38 self
._msg
_iter
= self
._create
_input
_port
_message
_iterator
(
39 self
._input
_ports
['in']
43 def _create_graph(src_comp_cls
, sink_comp_cls
, flt_comp_cls
=None):
46 src_comp
= graph
.add_component(src_comp_cls
, 'src')
47 sink_comp
= graph
.add_component(sink_comp_cls
, 'sink')
49 if flt_comp_cls
is not None:
50 flt_comp
= graph
.add_component(flt_comp_cls
, 'flt')
51 graph
.connect_ports(src_comp
.output_ports
['out'], flt_comp
.input_ports
['in'])
52 graph
.connect_ports(flt_comp
.output_ports
['out'], sink_comp
.input_ports
['in'])
54 graph
.connect_ports(src_comp
.output_ports
['out'], sink_comp
.input_ports
['in'])
59 class UserMessageIteratorTestCase(unittest
.TestCase
):
61 the_output_port_from_source
= None
62 the_output_port_from_iter
= None
64 class MyIter(bt2
._UserMessageIterator
):
65 def __init__(self
, config
, self_port_output
):
67 nonlocal the_output_port_from_iter
69 the_output_port_from_iter
= self_port_output
71 class MySource(bt2
._UserSourceComponent
, message_iterator_class
=MyIter
):
72 def __init__(self
, config
, params
, obj
):
73 nonlocal the_output_port_from_source
74 the_output_port_from_source
= self
._add
_output
_port
('out', 'user data')
77 graph
= _create_graph(MySource
, SimpleSink
)
79 self
.assertTrue(initialized
)
81 the_output_port_from_source
.addr
, the_output_port_from_iter
.addr
83 self
.assertEqual(the_output_port_from_iter
.user_data
, 'user data')
85 def test_create_from_message_iterator(self
):
86 class MySourceIter(bt2
._UserMessageIterator
):
87 def __init__(self
, config
, self_port_output
):
88 nonlocal src_iter_initialized
89 src_iter_initialized
= True
91 class MySource(bt2
._UserSourceComponent
, message_iterator_class
=MySourceIter
):
92 def __init__(self
, config
, params
, obj
):
93 self
._add
_output
_port
('out')
95 class MyFilterIter(bt2
._UserMessageIterator
):
96 def __init__(self
, config
, self_port_output
):
97 nonlocal flt_iter_initialized
98 flt_iter_initialized
= True
99 self
._up
_iter
= self
._create
_input
_port
_message
_iterator
(
100 self
._component
._input
_ports
['in']
104 return next(self
._up
_iter
)
106 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyFilterIter
):
107 def __init__(self
, config
, params
, obj
):
108 self
._add
_input
_port
('in')
109 self
._add
_output
_port
('out')
111 src_iter_initialized
= False
112 flt_iter_initialized
= False
113 graph
= _create_graph(MySource
, SimpleSink
, MyFilter
)
115 self
.assertTrue(src_iter_initialized
)
116 self
.assertTrue(flt_iter_initialized
)
118 def test_create_user_error(self
):
119 # This tests both error handling by
120 # _UserSinkComponent._create_input_port_message_iterator
121 # and _UserMessageIterator._create_input_port_message_iterator, as they
122 # are both used in the graph.
123 class MySourceIter(bt2
._UserMessageIterator
):
124 def __init__(self
, config
, self_port_output
):
125 raise ValueError('Very bad error')
127 class MySource(bt2
._UserSourceComponent
, message_iterator_class
=MySourceIter
):
128 def __init__(self
, config
, params
, obj
):
129 self
._add
_output
_port
('out')
131 class MyFilterIter(bt2
._UserMessageIterator
):
132 def __init__(self
, config
, self_port_output
):
133 # This is expected to raise because of the error in
134 # MySourceIter.__init__.
135 self
._create
_input
_port
_message
_iterator
(
136 self
._component
._input
_ports
['in']
139 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyFilterIter
):
140 def __init__(self
, config
, params
, obj
):
141 self
._add
_input
_port
('in')
142 self
._add
_output
_port
('out')
144 graph
= _create_graph(MySource
, SimpleSink
, MyFilter
)
146 with self
.assertRaises(bt2
._Error
) as ctx
:
152 self
.assertIsInstance(cause
, bt2
._MessageIteratorErrorCause
)
153 self
.assertEqual(cause
.component_name
, 'src')
154 self
.assertEqual(cause
.component_output_port_name
, 'out')
155 self
.assertIn('ValueError: Very bad error', cause
.message
)
157 def test_finalize(self
):
158 class MyIter(bt2
._UserMessageIterator
):
159 def _user_finalize(self
):
163 class MySource(bt2
._UserSourceComponent
, message_iterator_class
=MyIter
):
164 def __init__(self
, config
, params
, obj
):
165 self
._add
_output
_port
('out')
168 graph
= _create_graph(MySource
, SimpleSink
)
171 self
.assertTrue(finalized
)
173 def test_config_parameter(self
):
174 class MyIter(bt2
._UserMessageIterator
):
175 def __init__(self
, config
, port
):
177 config_type
= type(config
)
179 class MySource(bt2
._UserSourceComponent
, message_iterator_class
=MyIter
):
180 def __init__(self
, config
, params
, obj
):
181 self
._add
_output
_port
('out')
184 graph
= _create_graph(MySource
, SimpleSink
)
186 self
.assertIs(config_type
, bt2_message_iterator
._MessageIteratorConfiguration
)
188 def _test_config_can_seek_forward(self
, set_can_seek_forward
):
189 class MyIter(bt2
._UserMessageIterator
):
190 def __init__(self
, config
, port
):
191 if set_can_seek_forward
:
192 config
.can_seek_forward
= True
194 class MySource(bt2
._UserSourceComponent
, message_iterator_class
=MyIter
):
195 def __init__(self
, config
, params
, obj
):
196 self
._add
_output
_port
('out')
198 class MySink(bt2
._UserSinkComponent
):
199 def __init__(self
, config
, params
, obj
):
200 self
._add
_input
_port
('in')
202 def _user_graph_is_configured(self
):
203 self
._msg
_iter
= self
._create
_input
_port
_message
_iterator
(
204 self
._input
_ports
['in']
207 def _user_consume(self
):
208 nonlocal can_seek_forward
209 can_seek_forward
= self
._msg
_iter
.can_seek_forward
211 can_seek_forward
= None
212 graph
= _create_graph(MySource
, MySink
)
214 self
.assertIs(can_seek_forward
, set_can_seek_forward
)
216 def test_config_can_seek_forward_default(self
):
217 self
._test
_config
_can
_seek
_forward
(False)
219 def test_config_can_seek_forward(self
):
220 self
._test
_config
_can
_seek
_forward
(True)
222 def test_config_can_seek_forward_wrong_type(self
):
223 class MyIter(bt2
._UserMessageIterator
):
224 def __init__(self
, config
, port
):
225 config
.can_seek_forward
= 1
227 class MySource(bt2
._UserSourceComponent
, message_iterator_class
=MyIter
):
228 def __init__(self
, config
, params
, obj
):
229 self
._add
_output
_port
('out')
231 graph
= _create_graph(MySource
, SimpleSink
)
232 with self
.assertRaises(bt2
._Error
) as ctx
:
235 root_cause
= ctx
.exception
[0]
236 self
.assertIn("TypeError: 'int' is not a 'bool' object", root_cause
.message
)
238 def test_component(self
):
239 class MyIter(bt2
._UserMessageIterator
):
240 def __init__(self
, config
, self_port_output
):
242 salut
= self
._component
._salut
244 class MySource(bt2
._UserSourceComponent
, message_iterator_class
=MyIter
):
245 def __init__(self
, config
, params
, obj
):
246 self
._add
_output
_port
('out')
250 graph
= _create_graph(MySource
, SimpleSink
)
252 self
.assertEqual(salut
, 23)
255 class MyIter(bt2
._UserMessageIterator
):
256 def __init__(self_iter
, config
, self_port_output
):
259 port
= self_iter
._port
260 self
.assertIs(type(self_port_output
), bt2_port
._UserComponentOutputPort
)
261 self
.assertIs(type(port
), bt2_port
._UserComponentOutputPort
)
262 self
.assertEqual(self_port_output
.addr
, port
.addr
)
264 class MySource(bt2
._UserSourceComponent
, message_iterator_class
=MyIter
):
265 def __init__(self
, config
, params
, obj
):
266 self
._add
_output
_port
('out')
269 graph
= _create_graph(MySource
, SimpleSink
)
271 self
.assertTrue(called
)
274 class MyIter(bt2
._UserMessageIterator
):
275 def __init__(self
, config
, self_port_output
):
279 class MySource(bt2
._UserSourceComponent
, message_iterator_class
=MyIter
):
280 def __init__(self
, config
, params
, obj
):
281 self
._add
_output
_port
('out')
284 graph
= _create_graph(MySource
, SimpleSink
)
286 self
.assertIsNotNone(addr
)
287 self
.assertNotEqual(addr
, 0)
289 # Test that messages returned by _UserMessageIterator.__next__ remain valid
290 # and can be re-used.
291 def test_reuse_message(self
):
292 class MyIter(bt2
._UserMessageIterator
):
293 def __init__(self
, config
, port
):
294 tc
, sc
, ec
= port
.user_data
296 stream
= trace
.create_stream(sc
)
297 packet
= stream
.create_packet()
299 # This message will be returned twice by __next__.
300 event_message
= self
._create
_event
_message
(ec
, packet
)
303 self
._create
_stream
_beginning
_message
(stream
),
304 self
._create
_packet
_beginning
_message
(packet
),
310 return self
._msgs
.pop(0)
312 class MySource(bt2
._UserSourceComponent
, message_iterator_class
=MyIter
):
313 def __init__(self
, config
, params
, obj
):
314 tc
= self
._create
_trace
_class
()
315 sc
= tc
.create_stream_class(supports_packets
=True)
316 ec
= sc
.create_event_class()
317 self
._add
_output
_port
('out', (tc
, sc
, ec
))
320 src
= graph
.add_component(MySource
, 'src')
321 it
= TestOutputPortMessageIterator(graph
, src
.output_ports
['out'])
323 # Skip beginning messages.
325 self
.assertIs(type(msg
), bt2
._StreamBeginningMessageConst
)
327 self
.assertIs(type(msg
), bt2
._PacketBeginningMessageConst
)
332 self
.assertIs(type(msg_ev1
), bt2
._EventMessageConst
)
333 self
.assertIs(type(msg_ev2
), bt2
._EventMessageConst
)
334 self
.assertEqual(msg_ev1
.addr
, msg_ev2
.addr
)
336 # Try consuming many times from an iterator that always returns TryAgain.
337 # This verifies that we are not missing an incref of Py_None, making the
338 # refcount of Py_None reach 0.
339 def test_try_again_many_times(self
):
340 class MyIter(bt2
._UserMessageIterator
):
344 class MySource(bt2
._UserSourceComponent
, message_iterator_class
=MyIter
):
345 def __init__(self
, config
, params
, obj
):
346 self
._add
_output
_port
('out')
348 class MyFilterIter(bt2
._UserMessageIterator
):
349 def __init__(self
, port
):
350 input_port
= port
.user_data
351 self
._upstream
_iter
= self
._create
_input
_port
_message
_iterator
(
356 return next(self
._upstream
_iter
)
358 def _user_seek_beginning(self
):
359 self
._upstream
_iter
.seek_beginning()
361 def _user_can_seek_beginning(self
):
362 return self
._upstream
_iter
.can_seek_beginning()
364 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyFilterIter
):
365 def __init__(self
, config
, params
, obj
):
366 input_port
= self
._add
_input
_port
('in')
367 self
._add
_output
_port
('out', input_port
)
370 src
= graph
.add_component(MySource
, 'src')
371 it
= TestOutputPortMessageIterator(graph
, src
.output_ports
['out'])
373 # Three times the initial ref count of `None` iterations should
374 # be enough to catch the bug even if there are small differences
375 # between configurations.
376 none_ref_count
= sys
.getrefcount(None) * 3
378 for i
in range(none_ref_count
):
379 with self
.assertRaises(bt2
.TryAgain
):
383 def _setup_seek_test(
385 user_seek_beginning
=None,
386 user_can_seek_beginning
=None,
387 user_seek_ns_from_origin
=None,
388 user_can_seek_ns_from_origin
=None,
390 class MySourceIter(bt2
._UserMessageIterator
):
391 def __init__(self
, config
, port
):
392 tc
, sc
, ec
= port
.user_data
394 stream
= trace
.create_stream(sc
)
395 packet
= stream
.create_packet()
398 self
._create
_stream
_beginning
_message
(stream
),
399 self
._create
_packet
_beginning
_message
(packet
),
400 self
._create
_event
_message
(ec
, packet
),
401 self
._create
_event
_message
(ec
, packet
),
402 self
._create
_packet
_end
_message
(packet
),
403 self
._create
_stream
_end
_message
(stream
),
408 if self
._at
< len(self
._msgs
):
409 msg
= self
._msgs
[self
._at
]
415 if user_seek_beginning
is not None:
416 MySourceIter
._user
_seek
_beginning
= user_seek_beginning
418 if user_can_seek_beginning
is not None:
419 MySourceIter
._user
_can
_seek
_beginning
= user_can_seek_beginning
421 if user_seek_ns_from_origin
is not None:
422 MySourceIter
._user
_seek
_ns
_from
_origin
= user_seek_ns_from_origin
424 if user_can_seek_ns_from_origin
is not None:
425 MySourceIter
._user
_can
_seek
_ns
_from
_origin
= user_can_seek_ns_from_origin
427 class MySource(bt2
._UserSourceComponent
, message_iterator_class
=MySourceIter
):
428 def __init__(self
, config
, params
, obj
):
429 tc
= self
._create
_trace
_class
()
430 sc
= tc
.create_stream_class(supports_packets
=True)
431 ec
= sc
.create_event_class()
433 self
._add
_output
_port
('out', (tc
, sc
, ec
))
435 class MyFilterIter(bt2
._UserMessageIterator
):
436 def __init__(self
, config
, port
):
437 self
._upstream
_iter
= self
._create
_input
_port
_message
_iterator
(
438 self
._component
._input
_ports
['in']
442 return next(self
._upstream
_iter
)
444 def _user_can_seek_beginning(self
):
445 return self
._upstream
_iter
.can_seek_beginning()
447 def _user_seek_beginning(self
):
448 self
._upstream
_iter
.seek_beginning()
450 def _user_can_seek_ns_from_origin(self
, ns_from_origin
):
451 return self
._upstream
_iter
.can_seek_ns_from_origin(ns_from_origin
)
453 def _user_seek_ns_from_origin(self
, ns_from_origin
):
454 self
._upstream
_iter
.seek_ns_from_origin(ns_from_origin
)
456 class MyFilter(bt2
._UserFilterComponent
, message_iterator_class
=MyFilterIter
):
457 def __init__(self
, config
, params
, obj
):
458 self
._add
_input
_port
('in')
459 self
._add
_output
_port
('out')
461 return _create_graph(MySource
, sink_cls
, flt_comp_cls
=MyFilter
)
464 class UserMessageIteratorSeekBeginningTestCase(unittest
.TestCase
):
465 def test_can_seek_beginning(self
):
466 class MySink(bt2
._UserSinkComponent
):
467 def __init__(self
, config
, params
, obj
):
468 self
._add
_input
_port
('in')
470 def _user_graph_is_configured(self
):
471 self
._msg
_iter
= self
._create
_input
_port
_message
_iterator
(
472 self
._input
_ports
['in']
475 def _user_consume(self
):
476 nonlocal can_seek_beginning
477 can_seek_beginning
= self
._msg
_iter
.can_seek_beginning()
479 def _user_can_seek_beginning(self
):
480 nonlocal input_port_iter_can_seek_beginning
481 return input_port_iter_can_seek_beginning
483 graph
= _setup_seek_test(
484 MySink
, user_can_seek_beginning
=_user_can_seek_beginning
487 input_port_iter_can_seek_beginning
= True
488 can_seek_beginning
= None
490 self
.assertIs(can_seek_beginning
, True)
492 input_port_iter_can_seek_beginning
= False
493 can_seek_beginning
= None
495 self
.assertIs(can_seek_beginning
, False)
497 def test_no_can_seek_beginning_with_seek_beginning(self
):
498 # Test an iterator without a _user_can_seek_beginning method, but with
499 # a _user_seek_beginning method.
500 class MySink(bt2
._UserSinkComponent
):
501 def __init__(self
, config
, params
, obj
):
502 self
._add
_input
_port
('in')
504 def _user_graph_is_configured(self
):
505 self
._msg
_iter
= self
._create
_input
_port
_message
_iterator
(
506 self
._input
_ports
['in']
509 def _user_consume(self
):
510 nonlocal can_seek_beginning
511 can_seek_beginning
= self
._msg
_iter
.can_seek_beginning()
513 def _user_seek_beginning(self
):
516 graph
= _setup_seek_test(MySink
, user_seek_beginning
=_user_seek_beginning
)
517 can_seek_beginning
= None
519 self
.assertIs(can_seek_beginning
, True)
521 def test_no_can_seek_beginning(self
):
522 # Test an iterator without a _user_can_seek_beginning method, without
523 # a _user_seek_beginning method.
524 class MySink(bt2
._UserSinkComponent
):
525 def __init__(self
, config
, params
, obj
):
526 self
._add
_input
_port
('in')
528 def _user_graph_is_configured(self
):
529 self
._msg
_iter
= self
._create
_input
_port
_message
_iterator
(
530 self
._input
_ports
['in']
533 def _user_consume(self
):
534 nonlocal can_seek_beginning
535 can_seek_beginning
= self
._msg
_iter
.can_seek_beginning()
537 graph
= _setup_seek_test(MySink
)
538 can_seek_beginning
= None
540 self
.assertIs(can_seek_beginning
, False)
542 def test_can_seek_beginning_user_error(self
):
543 class MySink(bt2
._UserSinkComponent
):
544 def __init__(self
, config
, params
, obj
):
545 self
._add
_input
_port
('in')
547 def _user_graph_is_configured(self
):
548 self
._msg
_iter
= self
._create
_input
_port
_message
_iterator
(
549 self
._input
_ports
['in']
552 def _user_consume(self
):
553 # This is expected to raise.
554 self
._msg
_iter
.can_seek_beginning()
556 def _user_can_seek_beginning(self
):
557 raise ValueError('moustiquaire')
559 graph
= _setup_seek_test(
560 MySink
, user_can_seek_beginning
=_user_can_seek_beginning
563 with self
.assertRaises(bt2
._Error
) as ctx
:
566 cause
= ctx
.exception
[0]
567 self
.assertIn('ValueError: moustiquaire', cause
.message
)
569 def test_can_seek_beginning_wrong_return_value(self
):
570 class MySink(bt2
._UserSinkComponent
):
571 def __init__(self
, config
, params
, obj
):
572 self
._add
_input
_port
('in')
574 def _user_graph_is_configured(self
):
575 self
._msg
_iter
= self
._create
_input
_port
_message
_iterator
(
576 self
._input
_ports
['in']
579 def _user_consume(self
):
580 # This is expected to raise.
581 self
._msg
_iter
.can_seek_beginning()
583 def _user_can_seek_beginning(self
):
586 graph
= _setup_seek_test(
587 MySink
, user_can_seek_beginning
=_user_can_seek_beginning
590 with self
.assertRaises(bt2
._Error
) as ctx
:
593 cause
= ctx
.exception
[0]
594 self
.assertIn("TypeError: 'str' is not a 'bool' object", cause
.message
)
596 def test_seek_beginning(self
):
597 class MySink(bt2
._UserSinkComponent
):
598 def __init__(self
, config
, params
, obj
):
599 self
._add
_input
_port
('in')
601 def _user_graph_is_configured(self
):
602 self
._msg
_iter
= self
._create
_input
_port
_message
_iterator
(
603 self
._input
_ports
['in']
606 def _user_consume(self
):
607 nonlocal do_seek_beginning
610 if do_seek_beginning
:
611 self
._msg
_iter
.seek_beginning()
614 msg
= next(self
._msg
_iter
)
616 def _user_seek_beginning(self
):
620 graph
= _setup_seek_test(MySink
, user_seek_beginning
=_user_seek_beginning
)
623 do_seek_beginning
= False
625 self
.assertIs(type(msg
), bt2
._StreamBeginningMessageConst
)
629 self
.assertIs(type(msg
), bt2
._PacketBeginningMessageConst
)
632 do_seek_beginning
= True
636 do_seek_beginning
= False
638 self
.assertIs(type(msg
), bt2
._StreamBeginningMessageConst
)
640 def test_seek_beginning_user_error(self
):
641 class MySink(bt2
._UserSinkComponent
):
642 def __init__(self
, config
, params
, obj
):
643 self
._add
_input
_port
('in')
645 def _user_graph_is_configured(self
):
646 self
._msg
_iter
= self
._create
_input
_port
_message
_iterator
(
647 self
._input
_ports
['in']
650 def _user_consume(self
):
651 self
._msg
_iter
.seek_beginning()
653 def _user_seek_beginning(self
):
654 raise ValueError('ouch')
656 graph
= _setup_seek_test(MySink
, user_seek_beginning
=_user_seek_beginning
)
658 with self
.assertRaises(bt2
._Error
):
662 class UserMessageIteratorSeekNsFromOriginTestCase(unittest
.TestCase
):
663 def test_can_seek_ns_from_origin(self
):
664 class MySink(bt2
._UserSinkComponent
):
665 def __init__(self
, config
, params
, obj
):
666 self
._add
_input
_port
('in')
668 def _user_graph_is_configured(self
):
669 self
._msg
_iter
= self
._create
_input
_port
_message
_iterator
(
670 self
._input
_ports
['in']
673 def _user_consume(self
):
674 nonlocal can_seek_ns_from_origin
675 nonlocal test_ns_from_origin
676 can_seek_ns_from_origin
= self
._msg
_iter
.can_seek_ns_from_origin(
680 def _user_can_seek_ns_from_origin(iter_self
, ns_from_origin
):
681 nonlocal input_port_iter_can_seek_ns_from_origin
682 nonlocal test_ns_from_origin
683 self
.assertEqual(ns_from_origin
, test_ns_from_origin
)
684 return input_port_iter_can_seek_ns_from_origin
686 graph
= _setup_seek_test(
687 MySink
, user_can_seek_ns_from_origin
=_user_can_seek_ns_from_origin
690 input_port_iter_can_seek_ns_from_origin
= True
691 can_seek_ns_from_origin
= None
692 test_ns_from_origin
= 1
694 self
.assertIs(can_seek_ns_from_origin
, True)
696 input_port_iter_can_seek_ns_from_origin
= False
697 can_seek_ns_from_origin
= None
698 test_ns_from_origin
= 2
700 self
.assertIs(can_seek_ns_from_origin
, False)
702 def test_no_can_seek_ns_from_origin_with_seek_ns_from_origin(self
):
703 # Test an iterator without a _user_can_seek_ns_from_origin method, but
704 # with a _user_seek_ns_from_origin method.
705 class MySink(bt2
._UserSinkComponent
):
706 def __init__(self
, config
, params
, obj
):
707 self
._add
_input
_port
('in')
709 def _user_graph_is_configured(self
):
710 self
._msg
_iter
= self
._create
_input
_port
_message
_iterator
(
711 self
._input
_ports
['in']
714 def _user_consume(self
):
715 nonlocal can_seek_ns_from_origin
716 nonlocal test_ns_from_origin
717 can_seek_ns_from_origin
= self
._msg
_iter
.can_seek_ns_from_origin(
721 def _user_seek_ns_from_origin(self
):
724 graph
= _setup_seek_test(
725 MySink
, user_seek_ns_from_origin
=_user_seek_ns_from_origin
727 can_seek_ns_from_origin
= None
728 test_ns_from_origin
= 2
730 self
.assertIs(can_seek_ns_from_origin
, True)
732 def test_no_can_seek_ns_from_origin_with_seek_beginning(self
):
733 # Test an iterator without a _user_can_seek_ns_from_origin method, but
734 # with a _user_seek_beginning method.
735 class MySink(bt2
._UserSinkComponent
):
736 def __init__(self
, config
, params
, obj
):
737 self
._add
_input
_port
('in')
739 def _user_graph_is_configured(self
):
740 self
._msg
_iter
= self
._create
_input
_port
_message
_iterator
(
741 self
._input
_ports
['in']
744 def _user_consume(self
):
745 nonlocal can_seek_ns_from_origin
746 nonlocal test_ns_from_origin
747 can_seek_ns_from_origin
= self
._msg
_iter
.can_seek_ns_from_origin(
751 def _user_seek_beginning(self
):
754 graph
= _setup_seek_test(MySink
, user_seek_beginning
=_user_seek_beginning
)
755 can_seek_ns_from_origin
= None
756 test_ns_from_origin
= 2
758 self
.assertIs(can_seek_ns_from_origin
, True)
760 def test_no_can_seek_ns_from_origin(self
):
761 # Test an iterator without a _user_can_seek_ns_from_origin method
762 # and no other related method.
763 class MySink(bt2
._UserSinkComponent
):
764 def __init__(self
, config
, params
, obj
):
765 self
._add
_input
_port
('in')
767 def _user_graph_is_configured(self
):
768 self
._msg
_iter
= self
._create
_input
_port
_message
_iterator
(
769 self
._input
_ports
['in']
772 def _user_consume(self
):
773 nonlocal can_seek_ns_from_origin
774 nonlocal test_ns_from_origin
775 can_seek_ns_from_origin
= self
._msg
_iter
.can_seek_ns_from_origin(
779 graph
= _setup_seek_test(MySink
)
780 can_seek_ns_from_origin
= None
781 test_ns_from_origin
= 2
783 self
.assertIs(can_seek_ns_from_origin
, False)
785 def test_can_seek_ns_from_origin_user_error(self
):
786 class MySink(bt2
._UserSinkComponent
):
787 def __init__(self
, config
, params
, obj
):
788 self
._add
_input
_port
('in')
790 def _user_graph_is_configured(self
):
791 self
._msg
_iter
= self
._create
_input
_port
_message
_iterator
(
792 self
._input
_ports
['in']
795 def _user_consume(self
):
796 # This is expected to raise.
797 self
._msg
_iter
.can_seek_ns_from_origin(2)
799 def _user_can_seek_ns_from_origin(self
, ns_from_origin
):
800 raise ValueError('Joutel')
802 graph
= _setup_seek_test(
803 MySink
, user_can_seek_ns_from_origin
=_user_can_seek_ns_from_origin
806 with self
.assertRaises(bt2
._Error
) as ctx
:
809 cause
= ctx
.exception
[0]
810 self
.assertIn('ValueError: Joutel', cause
.message
)
812 def test_can_seek_ns_from_origin_wrong_return_value(self
):
813 class MySink(bt2
._UserSinkComponent
):
814 def __init__(self
, config
, params
, obj
):
815 self
._add
_input
_port
('in')
817 def _user_graph_is_configured(self
):
818 self
._msg
_iter
= self
._create
_input
_port
_message
_iterator
(
819 self
._input
_ports
['in']
822 def _user_consume(self
):
823 # This is expected to raise.
824 self
._msg
_iter
.can_seek_ns_from_origin(2)
826 def _user_can_seek_ns_from_origin(self
, ns_from_origin
):
829 graph
= _setup_seek_test(
830 MySink
, user_can_seek_ns_from_origin
=_user_can_seek_ns_from_origin
833 with self
.assertRaises(bt2
._Error
) as ctx
:
836 cause
= ctx
.exception
[0]
837 self
.assertIn("TypeError: 'str' is not a 'bool' object", cause
.message
)
839 def test_seek_ns_from_origin(self
):
840 class MySink(bt2
._UserSinkComponent
):
841 def __init__(self
, config
, params
, obj
):
842 self
._add
_input
_port
('in')
844 def _user_graph_is_configured(self
):
845 self
._msg
_iter
= self
._create
_input
_port
_message
_iterator
(
846 self
._input
_ports
['in']
849 def _user_consume(self
):
850 self
._msg
_iter
.seek_ns_from_origin(17)
852 def _user_seek_ns_from_origin(self
, ns_from_origin
):
853 nonlocal actual_ns_from_origin
854 actual_ns_from_origin
= ns_from_origin
856 graph
= _setup_seek_test(
857 MySink
, user_seek_ns_from_origin
=_user_seek_ns_from_origin
860 actual_ns_from_origin
= None
862 self
.assertEqual(actual_ns_from_origin
, 17)
865 if __name__
== '__main__':