def _user_consume(self):
pass
- comp = self._create_comp(MySink, 'yaes')
+ self._create_comp(MySink, 'yaes')
def test_logging_level(self):
class MySink(bt2._UserSinkComponent):
def _user_consume(self):
pass
- comp = self._create_comp(MySink, 'yaes', bt2.LoggingLevel.INFO)
+ self._create_comp(MySink, 'yaes', bt2.LoggingLevel.INFO)
def test_graph_mip_version(self):
class MySink(bt2._UserSinkComponent):
def _user_consume(self):
pass
- comp = self._create_comp(MySink, 'yaes', bt2.LoggingLevel.INFO)
+ self._create_comp(MySink, 'yaes', bt2.LoggingLevel.INFO)
def test_class(self):
class MySink(bt2._UserSinkComponent):
graph = bt2.Graph()
src = graph.add_component(MySource, 'src')
sink = graph.add_component(MySink, 'sink')
- conn = graph.connect_ports(src.output_ports['out'], sink.input_ports['in'])
+ graph.connect_ports(src.output_ports['out'], sink.input_ports['in'])
def test_downstream_port(self):
class MyIter(bt2._UserMessageIterator):
def test_current_thread_error_none(self):
# When a bt2._Error is raised, it steals the current thread's error.
# Verify that it is now NULL.
- exc = self._run_failing_graph(SourceWithFailingInit, WorkingSink)
+ self._run_failing_graph(SourceWithFailingInit, WorkingSink)
self.assertIsNone(native_bt.current_thread_take_error())
def test_len(self):
self._append_element_method(self._fc, 'yes', sub_fc2)
def test_iadd(self):
- other_fc = self._create_default_field_class()
a_field_class = self._tc.create_real_field_class()
b_field_class = self._tc.create_signed_integer_field_class(17)
self._append_element_method(self._fc, 'a_float', a_field_class)
)
def test_iadd(self):
- other_fc = self._create_default_field_class()
a_field_class = self._tc.create_real_field_class()
self._fc.append_option('a_float', a_field_class, self._ranges1)
c_field_class = self._tc.create_string_field_class()
pass
params = {'hello': 23, 'path': '/path/to/stuff'}
- comp = self._graph.add_component(MySink, 'salut', params)
+ self._graph.add_component(MySink, 'salut', params)
self.assertEqual(params, comp_params)
del comp_params
pass
obj = object()
- comp = self._graph.add_component(MySink, 'salut', obj=obj)
+ self._graph.add_component(MySink, 'salut', obj=obj)
self.assertIs(comp_obj, obj)
del comp_obj
def _user_consume(self):
pass
- comp = self._graph.add_component(MySink, 'salut')
+ self._graph.add_component(MySink, 'salut')
self.assertIsNone(comp_obj)
del comp_obj
def test_add_component_obj_non_python_comp_cls(self):
- comp_obj = None
-
plugin = bt2.find_plugin('text', find_in_user_dir=False, find_in_sys_dir=False)
assert plugin is not None
cc = plugin.source_component_classes['dmesg']
assert cc is not None
with self.assertRaises(ValueError):
- comp = self._graph.add_component(cc, 'salut', obj=57)
+ self._graph.add_component(cc, 'salut', obj=57)
def test_add_component_invalid_cls_type(self):
with self.assertRaises(TypeError):
sink = self._graph.add_component(MySink, 'sink')
with self.assertRaises(TypeError):
- conn = self._graph.connect_ports(
- sink.input_ports['in'], src.output_ports['out']
- )
+ self._graph.connect_ports(sink.input_ports['in'], src.output_ports['out'])
def test_add_interrupter(self):
class MyIter(bt2._UserMessageIterator):
src = self._graph.add_component(MySource, 'src')
sink = self._graph.add_component(MySink, 'sink')
- conn = self._graph.connect_ports(
- src.output_ports['out'], sink.input_ports['in']
- )
+ self._graph.connect_ports(src.output_ports['out'], sink.input_ports['in'])
self._graph.run()
def test_run_once(self):
run_count = 0
src = self._graph.add_component(MySource, 'src')
sink = self._graph.add_component(MySink, 'sink')
- conn = self._graph.connect_ports(
- src.output_ports['out'], sink.input_ports['in']
- )
+ self._graph.connect_ports(src.output_ports['out'], sink.input_ports['in'])
with self.assertRaises(bt2.TryAgain):
self._graph.run_once()
src = self._graph.add_component(MySource, 'src')
sink = self._graph.add_component(MySink, 'sink')
- conn = self._graph.connect_ports(
- src.output_ports['out'], sink.input_ports['in']
- )
+ self._graph.connect_ports(src.output_ports['out'], sink.input_ports['in'])
with self.assertRaises(bt2.Stop):
self._graph.run_once()
src = self._graph.add_component(MySource, 'src')
sink = self._graph.add_component(MySink, 'sink')
- conn = self._graph.connect_ports(
- src.output_ports['out'], sink.input_ports['in']
- )
+ self._graph.connect_ports(src.output_ports['out'], sink.input_ports['in'])
with self.assertRaises(bt2.TryAgain):
self._graph.run()
src = self._graph.add_component(MySource, 'src')
sink = self._graph.add_component(MySink, 'sink')
- conn = self._graph.connect_ports(
- src.output_ports['out'], sink.input_ports['in']
- )
+ self._graph.connect_ports(src.output_ports['out'], sink.input_ports['in'])
with self.assertRaises(bt2._Error):
self._graph.run()
def test_create_wrong_type_lower(self):
with self.assertRaises(TypeError):
- rg = self._CLS(19.3, self._def_upper)
+ self._CLS(19.3, self._def_upper)
def test_create_wrong_type_lower(self):
with self.assertRaises(TypeError):
- rg = self._CLS(self._def_lower, 19.3)
+ self._CLS(self._def_lower, 19.3)
def test_create_out_of_bound_lower(self):
with self.assertRaises(ValueError):
- rg = self._CLS(self._oob_lower, self._def_upper)
+ self._CLS(self._oob_lower, self._def_upper)
def test_create_out_of_bound_upper(self):
with self.assertRaises(ValueError):
- rg = self._CLS(self._def_lower, self._oob_upper)
+ self._CLS(self._def_lower, self._oob_upper)
def test_create_lower_gt_upper(self):
with self.assertRaises(ValueError):
- rg = self._CLS(self._def_lower, self._def_lower - 1)
+ self._CLS(self._def_lower, self._def_lower - 1)
def test_contains_lower(self):
self.assertTrue(self._rg.contains(self._def_lower))
def test_find_existing(self):
plugin = bt2.find_plugin('ctf', find_in_user_dir=False, find_in_sys_dir=False)
+ self.assertIsNotNone(plugin)
class PluginTestCase(unittest.TestCase):
user_datas = []
- comp = self._create_comp(MySource)
+ self._create_comp(MySource)
self.assertEqual(user_datas, [None, 2])
def test_filter_self_port_user_data(self):
user_datas = []
- comp = self._create_comp(MyFilter)
+ self._create_comp(MyFilter)
self.assertEqual(
user_datas, [None, 'user data string', None, {'user data': 'dict'}]
)
user_datas = []
- comp = self._create_comp(MySink)
+ self._create_comp(MySink)
self.assertEqual(user_datas, [None, set()])
query_params = params
query_params = 23
- res = bt2.QueryExecutor(MySink, 'obj', None).query()
+ bt2.QueryExecutor(MySink, 'obj', None).query()
self.assertIs(query_params, None)
del query_params
query_params = params
query_params = 23
- res = bt2.QueryExecutor(MySink, 'obj').query()
+ bt2.QueryExecutor(MySink, 'obj').query()
self.assertIs(query_params, None)
del query_params
query_method_obj = None
method_obj = object()
- res = bt2.QueryExecutor(MySink, 'obj', method_obj=method_obj).query()
+ bt2.QueryExecutor(MySink, 'obj', method_obj=method_obj).query()
self.assertIs(query_method_obj, method_obj)
del query_method_obj
query_method_obj = method_obj
query_method_obj = object()
- res = bt2.QueryExecutor(MySink, 'obj').query()
+ bt2.QueryExecutor(MySink, 'obj').query()
self.assertIsNone(query_method_obj)
del query_method_obj
raise ValueError
with self.assertRaises(bt2._Error) as ctx:
- res = bt2.QueryExecutor(MySink, 'obj', [17, 23]).query()
+ bt2.QueryExecutor(MySink, 'obj', [17, 23]).query()
exc = ctx.exception
self.assertEqual(len(exc), 3)
raise bt2.UnknownObject
with self.assertRaises(bt2.UnknownObject):
- res = bt2.QueryExecutor(MySink, 'obj', [17, 23]).query()
+ bt2.QueryExecutor(MySink, 'obj', [17, 23]).query()
def test_query_logging_level_invalid_type(self):
class MySink(bt2._UserSinkComponent):
raise bt2.TryAgain
with self.assertRaises(bt2.TryAgain):
- res = bt2.QueryExecutor(MySink, 'obj', [17, 23]).query()
+ bt2.QueryExecutor(MySink, 'obj', [17, 23]).query()
def test_query_add_interrupter(self):
class MySink(bt2._UserSinkComponent):
def test_supports_packets_raises_type_error(self):
with self.assertRaises(TypeError):
- sc = self._tc.create_stream_class(
+ self._tc.create_stream_class(
default_clock_class=self._cc, supports_packets=23
)
def test_packets_have_begin_default_cs_raises_type_error(self):
with self.assertRaises(TypeError):
- sc = self._tc.create_stream_class(
+ self._tc.create_stream_class(
default_clock_class=self._cc,
packets_have_beginning_default_clock_snapshot=23,
)
def test_packets_have_end_default_cs_raises_type_error(self):
with self.assertRaises(TypeError):
- sc = self._tc.create_stream_class(
+ self._tc.create_stream_class(
default_clock_class=self._cc, packets_have_end_default_clock_snapshot=23
)
def test_does_not_support_packets_raises_with_begin_cs(self):
with self.assertRaises(ValueError):
- sc = self._tc.create_stream_class(
+ self._tc.create_stream_class(
default_clock_class=self._cc,
packets_have_beginning_default_clock_snapshot=True,
)
def test_does_not_support_packets_raises_with_end_cs(self):
with self.assertRaises(ValueError):
- sc = self._tc.create_stream_class(
+ self._tc.create_stream_class(
default_clock_class=self._cc,
packets_have_end_default_clock_snapshot=True,
)
def test_supports_discarded_events_raises_type_error(self):
with self.assertRaises(TypeError):
- sc = self._tc.create_stream_class(
+ self._tc.create_stream_class(
default_clock_class=self._cc, supports_discarded_events=23
)
def test_discarded_events_have_default_cs_raises_type_error(self):
with self.assertRaises(TypeError):
- sc = self._tc.create_stream_class(
+ self._tc.create_stream_class(
default_clock_class=self._cc,
discarded_events_have_default_clock_snapshots=23,
)
def test_does_not_support_discarded_events_raises_with_cs(self):
with self.assertRaises(ValueError):
- sc = self._tc.create_stream_class(
+ self._tc.create_stream_class(
default_clock_class=self._cc,
discarded_events_have_default_clock_snapshots=True,
)
def test_supports_discarded_packets_raises_without_packet_support(self):
with self.assertRaises(ValueError):
- sc = self._tc.create_stream_class(
+ self._tc.create_stream_class(
default_clock_class=self._cc, supports_discarded_packets=True
)
def test_supports_discarded_packets_raises_type_error(self):
with self.assertRaises(TypeError):
- sc = self._tc.create_stream_class(
+ self._tc.create_stream_class(
default_clock_class=self._cc,
supports_discarded_packets=23,
supports_packets=True,
def test_discarded_packets_have_default_cs_raises_type_error(self):
with self.assertRaises(TypeError):
- sc = self._tc.create_stream_class(
+ self._tc.create_stream_class(
default_clock_class=self._cc,
discarded_packets_have_default_clock_snapshots=23,
supports_packets=True,
def test_does_not_support_discarded_packets_raises_with_cs(self):
with self.assertRaises(ValueError):
- sc = self._tc.create_stream_class(
+ self._tc.create_stream_class(
default_clock_class=self._cc,
discarded_packets_have_default_clock_snapshots=True,
supports_packets=True,
self.assertTrue(tc.assigns_automatic_stream_class_id)
with self.assertRaises(ValueError):
- sc1 = tc.create_stream_class(23)
+ tc.create_stream_class(23)
def test_no_assigns_automatic_stream_class_id(self):
def f(comp_self):
with self.assertRaisesRegex(
TypeError, "cannot create value object from 'A' object"
- ) as cm:
- v = bt2.create_value(a)
+ ):
+ bt2.create_value(a)
class BoolValueTestCase(_TestNumericValue, unittest.TestCase):
def test_create_from_int_non_zero(self):
with self.assertRaises(TypeError):
- b = bt2.BoolValue(23)
+ bt2.BoolValue(23)
def test_create_from_int_zero(self):
with self.assertRaises(TypeError):
- b = bt2.BoolValue(0)
+ bt2.BoolValue(0)
def test_assign_true(self):
b = bt2.BoolValue()
pass
with self._assert_expecting_int():
- i = self._CLS(A())
+ self._CLS(A())
def test_create_from_varray(self):
with self._assert_expecting_int():
- i = self._CLS(bt2.ArrayValue())
+ self._CLS(bt2.ArrayValue())
def test_assign_true(self):
raw = True
def test_create_pos_too_big(self):
with self._assert_expecting_int64():
- i = self._CLS(2 ** 63)
+ self._CLS(2 ** 63)
def test_create_neg_too_big(self):
with self._assert_expecting_int64():
- i = self._CLS(-(2 ** 63) - 1)
+ self._CLS(-(2 ** 63) - 1)
def test_assign_neg_int(self):
raw = -13
def test_create_pos_too_big(self):
with self._assert_expecting_uint64():
- i = self._CLS(2 ** 64)
+ self._CLS(2 ** 64)
def test_create_neg(self):
with self._assert_expecting_uint64():
- i = self._CLS(-1)
+ self._CLS(-1)
_inject_numeric_testing_methods(UnsignedIntegerValueTestCase)
pass
with self._assert_expecting_float():
- f = bt2.RealValue(A())
+ bt2.RealValue(A())
def test_create_from_varray(self):
with self._assert_expecting_float():
- f = bt2.RealValue(bt2.ArrayValue())
+ bt2.RealValue(bt2.ArrayValue())
def test_assign_true(self):
self._def.value = True
pass
with self._assert_expecting_str():
- i = bt2.StringValue(A())
+ bt2.StringValue(A())
def test_create_from_varray(self):
with self._assert_expecting_str():
- i = bt2.StringValue(bt2.ArrayValue())
+ bt2.StringValue(bt2.ArrayValue())
def test_assign_int(self):
with self._assert_expecting_str():
pass
with self._assert_type_error():
- a = bt2.ArrayValue(A())
+ bt2.ArrayValue(A())
def test_bool_op_true(self):
self.assertTrue(bool(self._def))
pass
with self.assertRaises(AttributeError):
- m = bt2.MapValue(A())
+ bt2.MapValue(A())
def test_bool_op_true(self):
self.assertTrue(bool(self._def))
bt2._UserSourceComponent, message_iterator_class=TheIteratorOfConfusion
):
def __init__(self, params, obj):
- tc = self._create_trace_class()
-
test_name = str(params['test-name'])
TEST_CASES[test_name].source_setup(self, test_name)