self.assertTrue(flt_iter_initialized)
# Test that creating a message iterator from a sink component on a
- # non-connected inport port raises.
+ # non-connected input port raises.
def test_create_from_sink_component_unconnected_port_raises(self):
class MySink(bt2._UserSinkComponent):
def __init__(comp_self, config, params, obj):
graph.run()
self.assertTrue(seen)
- # Test that creating a message iterator from a message iteartor on a
- # non-connected inport port raises.
+ # Test that creating a message iterator from a message iterator on a
+ # non-connected input port raises.
def test_create_from_message_iterator_unconnected_port_raises(self):
class MyFilterIter(bt2._UserMessageIterator):
def __init__(iter_self, config, port):
# This verifies that we are not missing an incref of Py_None, making the
# refcount of Py_None reach 0.
def test_try_again_many_times(self):
+ # Starting with Python 3.12, `None` is immortal: its reference
+ # count operations are no-op. Skip this test in that case.
+ before = sys.getrefcount(None)
+ dummy = None # noqa: F841
+
+ if before == sys.getrefcount(None):
+ raise unittest.SkipTest("`None` is immortal")
+
class MyIter(bt2._UserMessageIterator):
def __next__(self):
raise bt2.TryAgain
def __init__(self, config, params, obj):
self._add_output_port("out")
- class MyFilterIter(bt2._UserMessageIterator):
- def __init__(self, port):
- input_port = port.user_data
- self._upstream_iter = self._create_message_iterator(input_port)
-
- def __next__(self):
- return next(self._upstream_iter)
-
- def _user_seek_beginning(self):
- self._upstream_iter.seek_beginning()
-
- def _user_can_seek_beginning(self):
- return self._upstream_iter.can_seek_beginning()
-
- class MyFilter(bt2._UserFilterComponent, message_iterator_class=MyFilterIter):
- def __init__(self, config, params, obj):
- input_port = self._add_input_port("in")
- self._add_output_port("out", input_port)
-
graph = bt2.Graph()
src = graph.add_component(MySource, "src")
it = TestOutputPortMessageIterator(graph, src.output_ports["out"])