res = bt2_value._create_from_ptr(res_ptr)
- assert type(res) == bt2.MapValue
+ assert type(res) is bt2.MapValue
assert 'status' in res
status = res['status']
comp_specs = []
comp_specs_raw = res['results']
- assert type(comp_specs_raw) == bt2.ArrayValue
+ assert type(comp_specs_raw) is bt2.ArrayValue
used_input_indices = set()
for comp_spec_raw in comp_specs_raw:
- assert type(comp_spec_raw) == bt2.ArrayValue
+ assert type(comp_spec_raw) is bt2.ArrayValue
assert len(comp_spec_raw) == 4
plugin_name = comp_spec_raw[0]
- assert type(plugin_name) == bt2.StringValue
+ assert type(plugin_name) is bt2.StringValue
plugin_name = str(plugin_name)
class_name = comp_spec_raw[1]
- assert type(class_name) == bt2.StringValue
+ assert type(class_name) is bt2.StringValue
class_name = str(class_name)
comp_inputs = comp_spec_raw[2]
- assert type(comp_inputs) == bt2.ArrayValue
+ assert type(comp_inputs) is bt2.ArrayValue
comp_orig_indices = comp_spec_raw[3]
assert type(comp_orig_indices)
class _TraceCollectionMessageIteratorProxySink(bt2_component._UserSinkComponent):
- def __init__(self, params, msg_list):
+ def __init__(self, config, params, msg_list):
assert type(msg_list) is list
self._msg_list = msg_list
self._add_input_port('in')
def _user_graph_is_configured(self):
- self._msg_iter = self._create_input_port_message_iterator(
- self._input_ports['in']
- )
+ self._msg_iter = self._create_message_iterator(self._input_ports['in'])
def _user_consume(self):
assert self._msg_list[0] is None