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

   1 import bt2
   2 import os
   3
   4 # This file defines source component classes that can print (depending the on
   5 # the `print` param):
   6 #
   7 #  - Parameter names and values, for parameter whose name starts with `test-`.
   8 #  - Component log levels as integers
   9
  10
  11 class TestIter(bt2._UserMessageIterator):
  12     pass
  13
  14
  15 class Base:
  16     @classmethod
  17     def _print_test_params(cls, params):
  18         items = sorted([str(x) for x in params.items() if x[0].startswith('test-')])
  19         print('{}: {}'.format(cls.__name__, ', '.join(items)))
  20
  21     def _print_log_level(self):
  22         cls_name = self.__class__.__name__
  23         log_level = self.logging_level
  24         print('{}: {}'.format(cls_name, log_level))
  25
  26     def _print_info(self, params):
  27         what = params['print']
  28         if what == 'test-params':
  29             self._print_test_params(params)
  30         elif what == 'log-level':
  31             self._print_log_level()
  32         else:
  33             assert False
  34
  35
  36 @bt2.plugin_component_class
  37 class TestSourceA(Base, bt2._UserSourceComponent, message_iterator_class=TestIter):
  38     def __init__(self, params):
  39         self._print_info(params)
  40
  41     @staticmethod
  42     def _user_query(priv_query_exec, obj, params):
  43         # Match files starting with 'aaa'.
  44
  45         if obj == 'babeltrace.support-info':
  46             if params['type'] != 'file':
  47                 return 0
  48
  49             name = os.path.basename(str(params['input']))
  50
  51             if name.startswith('aaa'):
  52                 return {'weight': 1, 'group': 'aaa'}
  53             else:
  54                 return 0
  55         else:
  56             raise bt2.UnknownObject
  57
  58
  59 @bt2.plugin_component_class
  60 class TestSourceB(Base, bt2._UserSourceComponent, message_iterator_class=TestIter):
  61     def __init__(self, params):
  62         self._print_info(params)
  63
  64     @staticmethod
  65     def _user_query(priv_query_exec, obj, params):
  66         # Match files starting with 'bbb'.
  67
  68         if obj == 'babeltrace.support-info':
  69             if params['type'] != 'file':
  70                 return 0
  71
  72             name = os.path.basename(str(params['input']))
  73
  74             if name.startswith('bbb'):
  75                 return {'weight': 1, 'group': 'bbb'}
  76             else:
  77                 return 0
  78         else:
  79             raise bt2.UnknownObject
  80
  81
  82 bt2.register_plugin(module_name=__name__, name="test")