tests/data/auto-source-discovery/params-log-level/bt_plugin_test.py

   1 # SPDX-License-Identifier: GPL-2.0-only
   2 #
   3 # Copyright (C) 2019 EfficiOS Inc.
   4 #
   5
   6 import bt2
   7 import os
   8
   9 # This file defines source component classes to help verify the parameters an
  10 # log levels passed to components.  Each component creates one stream, with a
  11 # name derived from either:
  12 #
  13 #   - the received params that start with `test-`
  14 #   - the received log level
  15 #
  16 # The `what` parameter determines what is used.
  17
  18
  19 class TestIter(bt2._UserMessageIterator):
  20     def __init__(self, config, output_port):
  21         params = output_port.user_data["params"]
  22         obj = output_port.user_data["obj"]
  23
  24         comp_cls_name = self._component.__class__.__name__
  25
  26         if params["what"] == "test-params":
  27             items = sorted([str(x) for x in params.items() if x[0].startswith("test-")])
  28             stream_name = "{}: {}".format(comp_cls_name, ", ".join(items))
  29         elif params["what"] == "log-level":
  30             log_level = self._component.logging_level
  31             stream_name = "{}: {}".format(comp_cls_name, log_level)
  32         elif params["what"] == "python-obj":
  33             assert type(obj) == str or obj is None
  34             stream_name = "{}: {}".format(comp_cls_name, obj)
  35         else:
  36             assert False
  37
  38         sc = output_port.user_data["sc"]
  39         tc = sc.trace_class
  40         t = tc()
  41         s = t.create_stream(sc, name=stream_name)
  42
  43         self._msgs = [
  44             self._create_stream_beginning_message(s),
  45             self._create_stream_end_message(s),
  46         ]
  47
  48     def __next__(self):
  49         if len(self._msgs) == 0:
  50             raise StopIteration
  51
  52         return self._msgs.pop(0)
  53
  54
  55 class Base:
  56     def __init__(self, params, obj):
  57         tc = self._create_trace_class()
  58         sc = tc.create_stream_class()
  59
  60         self._add_output_port("out", {"params": params, "obj": obj, "sc": sc})
  61
  62
  63 @bt2.plugin_component_class
  64 class TestSourceA(Base, bt2._UserSourceComponent, message_iterator_class=TestIter):
  65     def __init__(self, config, params, obj):
  66         super().__init__(params, obj)
  67
  68     @staticmethod
  69     def _user_query(priv_query_exec, obj, params, method_obj):
  70         # Match files starting with 'aaa'.
  71
  72         if obj == "babeltrace.support-info":
  73             if params["type"] != "file":
  74                 return 0
  75
  76             name = os.path.basename(str(params["input"]))
  77
  78             if name.startswith("aaa"):
  79                 return {"weight": 1, "group": "aaa"}
  80             else:
  81                 return 0
  82         else:
  83             raise bt2.UnknownObject
  84
  85
  86 @bt2.plugin_component_class
  87 class TestSourceB(Base, bt2._UserSourceComponent, message_iterator_class=TestIter):
  88     def __init__(self, config, params, obj):
  89         super().__init__(params, obj)
  90
  91     @staticmethod
  92     def _user_query(priv_query_exec, obj, params, method_obj):
  93         # Match files starting with 'bbb'.
  94
  95         if obj == "babeltrace.support-info":
  96             if params["type"] != "file":
  97                 return 0
  98
  99             name = os.path.basename(str(params["input"]))
 100
 101             if name.startswith("bbb"):
 102                 return {"weight": 1, "group": "bbb"}
 103             else:
 104                 return 0
 105         else:
 106             raise bt2.UnknownObject
 107
 108
 109 bt2.register_plugin(module_name=__name__, name="test")