-import bt2
+# SPDX-License-Identifier: GPL-2.0-only
+#
+# Copyright (C) 2019 EfficiOS Inc.
+#
+
import os
+import bt2
+
# Source component classes in this file recognize and group inputs in
# various ways. One stream is created by each component, with a name
# derived from the component class and the inputs. This is then checked
class TestIter(bt2._UserMessageIterator):
- def __init__(self, output_port):
- inputs = output_port.user_data['inputs']
- sc = output_port.user_data['sc']
+ def __init__(self, config, output_port):
+ inputs = output_port.user_data["inputs"]
+ sc = output_port.user_data["sc"]
tc = sc.trace_class
t = tc()
s = t.create_stream(sc, name=self._make_stream_name(inputs))
comp_cls_name = self._component.__class__.__name__
return (
comp_cls_name
- + ': '
- + ', '.join(sorted([os.path.basename(str(x)) for x in inputs]))
+ + ": "
+ + ", ".join(sorted([os.path.basename(str(x)) for x in inputs]))
)
def __next__(self):
tc = self._create_trace_class()
sc = tc.create_stream_class()
- self._add_output_port('out', {'inputs': params['inputs'], 'sc': sc})
+ self._add_output_port("out", {"inputs": params["inputs"], "sc": sc})
@bt2.plugin_component_class
files are not grouped.
"""
- def __init__(self, params, obj):
+ def __init__(self, config, params, obj):
super().__init__(params)
@staticmethod
def _user_query(priv_query_exec, obj, params, method_obj):
- if obj == 'babeltrace.support-info':
- if params['type'] == 'file':
- name = os.path.basename(str(params['input']))
-
- if name.startswith('aaa'):
- return {'weight': 1, 'group': 'aaa'}
- elif name.startswith('bbb'):
- return {'weight': 0.5, 'group': 'bbb'}
- elif name.startswith('ccc'):
- # Try two different ways of returning "no group", and two
- # different ways of returning 1 (an int and a float).
- if name[3] == '1':
- return {'weight': 1, 'group': None}
- elif name[3] == '2':
- return {'weight': 1.0, 'group': None}
- elif name[3] == '3':
+ if obj == "babeltrace.support-info":
+ if params["type"] == "file":
+ name = os.path.basename(str(params["input"]))
+
+ if name.startswith("aaa"):
+ return {"weight": 1, "group": "aaa"}
+ elif name.startswith("bbb"):
+ return {"weight": 0.5, "group": "bbb"}
+ elif name.startswith("ccc"):
+ # Try two different ways of returning 1 (an int and a float).
+ if name[3] == "1":
return 1
else:
return 1.0
directory "some-dir" won't be found by TestSourceExt, because we won't
recurse in "some-dir"."""
- def __init__(self, params, obj):
+ def __init__(self, config, params, obj):
super().__init__(params)
@staticmethod
def _user_query(priv_query_exec, obj, params, method_obj):
- if obj == 'babeltrace.support-info':
- if params['type'] == 'directory':
- name = os.path.basename(str(params['input']))
- return 1 if name == 'some-dir' else 0
+ if obj == "babeltrace.support-info":
+ if params["type"] == "directory":
+ name = os.path.basename(str(params["input"]))
+ return 1 if name == "some-dir" else 0
else:
return 0
else:
class TestSourceABCDE(Base, bt2._UserSourceComponent, message_iterator_class=TestIter):
"""A source that recognizes the arbitrary string input "ABCDE"."""
- def __init__(self, params, obj):
+ def __init__(self, config, params, obj):
super().__init__(params)
@staticmethod
def _user_query(priv_query_exec, obj, params, method_obj):
- if obj == 'babeltrace.support-info':
+ if obj == "babeltrace.support-info":
return (
1.0
- if params['type'] == 'string' and params['input'] == 'ABCDE'
+ if params["type"] == "string" and params["input"] == "ABCDE"
else 0.0
)
else: