tests/bindings/python/bt2/test_component.py

   1 from bt2 import value
   2 import unittest
   3 import copy
   4 import bt2
   5
   6
   7 class UserComponentTestCase(unittest.TestCase):
   8     @staticmethod
   9     def _create_comp(comp_cls, name=None, log_level=bt2.LoggingLevel.NONE):
  10         graph = bt2.Graph()
  11
  12         if name is None:
  13             name = 'comp'
  14
  15         return graph.add_component(comp_cls, name, logging_level=log_level)
  16
  17     def test_name(self):
  18         class MySink(bt2._UserSinkComponent):
  19             def __init__(comp_self, params):
  20                 self.assertEqual(comp_self.name, 'yaes')
  21
  22             def _consume(self):
  23                 pass
  24
  25         comp = self._create_comp(MySink, 'yaes')
  26
  27     def test_logging_level(self):
  28         class MySink(bt2._UserSinkComponent):
  29             def __init__(comp_self, params):
  30                 self.assertEqual(comp_self.logging_level, bt2.LoggingLevel.INFO)
  31
  32             def _consume(self):
  33                 pass
  34
  35         comp = self._create_comp(MySink, 'yaes', bt2.LoggingLevel.INFO)
  36
  37     def test_class(self):
  38         class MySink(bt2._UserSinkComponent):
  39             def __init__(comp_self, params):
  40                 self.assertEqual(comp_self.cls, MySink)
  41
  42             def _consume(self):
  43                 pass
  44
  45         self._create_comp(MySink)
  46
  47     def test_addr(self):
  48         class MySink(bt2._UserSinkComponent):
  49             def __init__(comp_self, params):
  50                 self.assertIsInstance(comp_self.addr, int)
  51                 self.assertNotEqual(comp_self.addr, 0)
  52
  53             def _consume(self):
  54                 pass
  55
  56         self._create_comp(MySink)
  57
  58     def test_finalize(self):
  59         finalized = False
  60
  61         class MySink(bt2._UserSinkComponent):
  62             def _consume(self):
  63                 pass
  64
  65             def _finalize(comp_self):
  66                 nonlocal finalized
  67                 finalized = True
  68
  69         graph = bt2.Graph()
  70         comp = graph.add_component(MySink, 'lel')
  71
  72         del graph
  73         del comp
  74         self.assertTrue(finalized)
  75
  76
  77 class GenericComponentTestCase(unittest.TestCase):
  78     @staticmethod
  79     def _create_comp(comp_cls, name=None, log_level=bt2.LoggingLevel.NONE):
  80         graph = bt2.Graph()
  81
  82         if name is None:
  83             name = 'comp'
  84
  85         return graph.add_component(comp_cls, name, logging_level=log_level)
  86
  87     def test_name(self):
  88         class MySink(bt2._UserSinkComponent):
  89             def _consume(self):
  90                 pass
  91
  92         comp = self._create_comp(MySink, 'yaes')
  93         self.assertEqual(comp.name, 'yaes')
  94
  95     def test_logging_level(self):
  96         class MySink(bt2._UserSinkComponent):
  97             def _consume(self):
  98                 pass
  99
 100         comp = self._create_comp(MySink, 'yaes', bt2.LoggingLevel.WARN)
 101         self.assertEqual(comp.logging_level, bt2.LoggingLevel.WARN)
 102
 103     def test_class(self):
 104         class MySink(bt2._UserSinkComponent):
 105             def _consume(self):
 106                 pass
 107
 108         comp = self._create_comp(MySink)
 109         self.assertEqual(comp.cls, MySink)
 110
 111     def test_addr(self):
 112         class MySink(bt2._UserSinkComponent):
 113             def _consume(self):
 114                 pass
 115
 116         comp = self._create_comp(MySink)
 117         self.assertIsInstance(comp.addr, int)
 118         self.assertNotEqual(comp.addr, 0)