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

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