class UnknownObject(Exception):
- '''
+ """
Raised when a component class handles a query for an object it doesn't
know about.
- '''
+ """
pass
def _trim_docstring(docstring):
lines = docstring.expandtabs().splitlines()
+
+ if len(lines) == 0:
+ return ''
+
indent = sys.maxsize
- for line in lines[1:]:
- stripped = line.lstrip()
+ if len(lines) > 1:
+ for line in lines[1:]:
+ stripped = line.lstrip()
- if stripped:
- indent = min(indent, len(line) - len(stripped))
+ if stripped:
+ indent = min(indent, len(line) - len(stripped))
trimmed = [lines[0].strip()]
- if indent < sys.maxsize:
+ if indent < sys.maxsize and len(lines) > 1:
for line in lines[1:]:
trimmed.append(line[indent:].rstrip())
self._component_name = native_bt.error_cause_component_actor_get_component_name(
ptr
)
- self._component_class_type = native_bt.error_cause_component_actor_get_component_class_type(
- ptr
+ self._component_class_type = (
+ native_bt.error_cause_component_actor_get_component_class_type(ptr)
)
- self._component_class_name = native_bt.error_cause_component_actor_get_component_class_name(
- ptr
+ self._component_class_name = (
+ native_bt.error_cause_component_actor_get_component_class_name(ptr)
)
self._plugin_name = native_bt.error_cause_component_actor_get_plugin_name(ptr)
class _ComponentClassErrorCause(_ErrorCause):
def __init__(self, ptr):
super().__init__(ptr)
- self._component_class_type = native_bt.error_cause_component_class_actor_get_component_class_type(
- ptr
+ self._component_class_type = (
+ native_bt.error_cause_component_class_actor_get_component_class_type(ptr)
)
- self._component_class_name = native_bt.error_cause_component_class_actor_get_component_class_name(
- ptr
+ self._component_class_name = (
+ native_bt.error_cause_component_class_actor_get_component_class_name(ptr)
)
self._plugin_name = native_bt.error_cause_component_class_actor_get_plugin_name(
ptr
class _MessageIteratorErrorCause(_ErrorCause):
def __init__(self, ptr):
super().__init__(ptr)
- self._component_name = native_bt.error_cause_message_iterator_actor_get_component_name(
- ptr
+ self._component_name = (
+ native_bt.error_cause_message_iterator_actor_get_component_name(ptr)
)
- self._component_output_port_name = native_bt.error_cause_message_iterator_actor_get_component_output_port_name(
- ptr
+ self._component_output_port_name = (
+ native_bt.error_cause_message_iterator_actor_get_component_output_port_name(
+ ptr
+ )
)
- self._component_class_type = native_bt.error_cause_message_iterator_actor_get_component_class_type(
- ptr
+ self._component_class_type = (
+ native_bt.error_cause_message_iterator_actor_get_component_class_type(ptr)
)
- self._component_class_name = native_bt.error_cause_message_iterator_actor_get_component_class_name(
- ptr
+ self._component_class_name = (
+ native_bt.error_cause_message_iterator_actor_get_component_class_name(ptr)
)
- self._plugin_name = native_bt.error_cause_message_iterator_actor_get_plugin_name(
- ptr
+ self._plugin_name = (
+ native_bt.error_cause_message_iterator_actor_get_plugin_name(ptr)
)
@property
utils._check_uint64(beg_clock_snapshot)
utils._check_uint64(end_clock_snapshot)
- ptr = native_bt.message_discarded_events_create_with_default_clock_snapshots(
- self._bt_ptr, stream._ptr, beg_clock_snapshot, end_clock_snapshot
+ ptr = (
+ native_bt.message_discarded_events_create_with_default_clock_snapshots(
+ self._bt_ptr, stream._ptr, beg_clock_snapshot, end_clock_snapshot
+ )
)
else:
if beg_clock_snapshot is not None or end_clock_snapshot is not None:
utils._check_uint64(beg_clock_snapshot)
utils._check_uint64(end_clock_snapshot)
- ptr = native_bt.message_discarded_packets_create_with_default_clock_snapshots(
- self._bt_ptr, stream._ptr, beg_clock_snapshot, end_clock_snapshot
+ ptr = (
+ native_bt.message_discarded_packets_create_with_default_clock_snapshots(
+ self._bt_ptr, stream._ptr, beg_clock_snapshot, end_clock_snapshot
+ )
)
else:
if beg_clock_snapshot is not None or end_clock_snapshot is not None:
)
else:
utils._check_type(ranges, bt2_integer_range_set.SignedIntegerRangeSet)
- ptr = native_bt.field_class_option_with_selector_field_integer_signed_create(
- self._ptr, content_fc._ptr, selector_fc._ptr, ranges._ptr
+ ptr = (
+ native_bt.field_class_option_with_selector_field_integer_signed_create(
+ self._ptr, content_fc._ptr, selector_fc._ptr, ranges._ptr
+ )
)
self._check_field_class_create_status(ptr, 'option')
self.assertEqual(MySink.description, 'The description.')
- def test_empty_description(self):
+ def test_empty_description_no_lines(self):
class MySink(bt2._UserSinkComponent):
+ # fmt: off
+ """"""
+ # fmt: on
+
+ def _user_consume(self):
+ pass
+
+ self.assertIsNone(MySink.description)
+
+ def test_empty_description_no_contents(self):
+ class MySink(bt2._UserSinkComponent):
+ # fmt: off
"""
"""
+ # fmt: on
def _user_consume(self):
pass
self.assertIsNone(MySink.description)
+ def test_empty_description_single_line(self):
+ class MySink(bt2._UserSinkComponent):
+ """my description"""
+
+ def _user_consume(self):
+ pass
+
+ self.assertEqual(MySink.description, "my description")
+
def test_help(self):
class MySink(bt2._UserSinkComponent):
"""
class ComponentClassTestCase(unittest.TestCase):
def setUp(self):
class MySink(bt2._UserSinkComponent):
- '''
+ """
The description.
The help.
- '''
+ """
def _user_consume(self):
pass
event.payload_field['mosquito'] = 42
msg = self._create_test_const_event_message(
- event_fields_config=event_fields_config, with_ep=True,
+ event_fields_config=event_fields_config,
+ with_ep=True,
)
ev = msg.event
# Trying to create when the stream does not support discarded packets.
def test_create_unsupported_raises(self):
def create_stream_class(tc, cc):
- return tc.create_stream_class(supports_packets=True,)
+ return tc.create_stream_class(
+ supports_packets=True,
+ )
def msg_iter_next(msg_iter, stream):
with self.assertRaisesRegex(