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, output_port):
  16         params = output_port.user_data['params']
  17
  18         comp_cls_name = self._component.__class__.__name__
  19
  20         if params['what'] == 'test-params':
  21             items = sorted([str(x) for x in params.items() if x[0].startswith('test-')])
  22             stream_name = '{}: {}'.format(comp_cls_name, ', '.join(items))
  23         elif params['what'] == 'log-level':
  24             log_level = self._component.logging_level
  25             stream_name = '{}: {}'.format(comp_cls_name, log_level)
  26         else:
  27             assert False
  28
  29         sc = output_port.user_data['sc']
  30         tc = sc.trace_class
  31         t = tc()
  32         s = t.create_stream(sc, name=stream_name)
  33
  34         self._msgs = [
  35             self._create_stream_beginning_message(s),
  36             self._create_stream_end_message(s),
  37         ]
  38
  39     def __next__(self):
  40         if len(self._msgs) == 0:
  41             raise StopIteration
  42
  43         return self._msgs.pop(0)
  44
  45
  46 class Base:
  47     def __init__(self, params):
  48         tc = self._create_trace_class()
  49         sc = tc.create_stream_class()
  50
  51         self._add_output_port('out', {'params': params, 'sc': sc})
  52
  53
  54 @bt2.plugin_component_class
  55 class TestSourceA(Base, bt2._UserSourceComponent, message_iterator_class=TestIter):
  56     def __init__(self, params, obj):
  57         super().__init__(params)
  58
  59     @staticmethod
  60     def _user_query(priv_query_exec, obj, params, method_obj):
  61         # Match files starting with 'aaa'.
  62
  63         if obj == 'babeltrace.support-info':
  64             if params['type'] != 'file':
  65                 return 0
  66
  67             name = os.path.basename(str(params['input']))
  68
  69             if name.startswith('aaa'):
  70                 return {'weight': 1, 'group': 'aaa'}
  71             else:
  72                 return 0
  73         else:
  74             raise bt2.UnknownObject
  75
  76
  77 @bt2.plugin_component_class
  78 class TestSourceB(Base, bt2._UserSourceComponent, message_iterator_class=TestIter):
  79     def __init__(self, params, obj):
  80         super().__init__(params)
  81
  82     @staticmethod
  83     def _user_query(priv_query_exec, obj, params, method_obj):
  84         # Match files starting with 'bbb'.
  85
  86         if obj == 'babeltrace.support-info':
  87             if params['type'] != 'file':
  88                 return 0
  89
  90             name = os.path.basename(str(params['input']))
  91
  92             if name.startswith('bbb'):
  93                 return {'weight': 1, 'group': 'bbb'}
  94             else:
  95                 return 0
  96         else:
  97             raise bt2.UnknownObject
  98
  99
 100 bt2.register_plugin(module_name=__name__, name="test")