1 # SPDX-License-Identifier: GPL-2.0-only
3 # Copyright (C) 2019 EfficiOS Inc.
9 from bt2
import component
as bt2_component
12 class UserComponentTestCase(unittest
.TestCase
):
14 def _create_comp(comp_cls
, name
=None, log_level
=bt2
.LoggingLevel
.NONE
):
20 return graph
.add_component(comp_cls
, name
, logging_level
=log_level
)
23 class MySink(bt2
._UserSinkComponent
):
24 def __init__(comp_self
, config
, params
, obj
):
25 self
.assertEqual(comp_self
.name
, "yaes")
27 def _user_consume(self
):
30 self
._create
_comp
(MySink
, "yaes")
32 def test_logging_level(self
):
33 class MySink(bt2
._UserSinkComponent
):
34 def __init__(comp_self
, config
, params
, obj
):
35 self
.assertEqual(comp_self
.logging_level
, bt2
.LoggingLevel
.INFO
)
37 def _user_consume(self
):
40 self
._create
_comp
(MySink
, "yaes", bt2
.LoggingLevel
.INFO
)
42 def test_graph_mip_version(self
):
43 class MySink(bt2
._UserSinkComponent
):
44 def __init__(comp_self
, config
, params
, obj
):
45 self
.assertEqual(comp_self
._graph
_mip
_version
, 0)
47 def _user_consume(self
):
50 self
._create
_comp
(MySink
, "yaes", bt2
.LoggingLevel
.INFO
)
53 class MySink(bt2
._UserSinkComponent
):
54 def __init__(comp_self
, config
, params
, obj
):
55 self
.assertEqual(comp_self
.cls
, MySink
)
57 def _user_consume(self
):
60 self
._create
_comp
(MySink
)
63 class MySink(bt2
._UserSinkComponent
):
64 def __init__(comp_self
, config
, params
, obj
):
65 self
.assertIsInstance(comp_self
.addr
, int)
66 self
.assertNotEqual(comp_self
.addr
, 0)
68 def _user_consume(self
):
71 self
._create
_comp
(MySink
)
73 def test_finalize(self
):
76 class MySink(bt2
._UserSinkComponent
):
77 def _user_consume(self
):
80 def _user_finalize(comp_self
):
85 comp
= graph
.add_component(MySink
, "lel")
89 self
.assertTrue(finalized
)
91 def test_source_component_config(self
):
93 bt2
._UserSourceComponent
, message_iterator_class
=bt2
._UserMessageIterator
95 def __init__(comp_self
, config
, params
, obj
):
97 cfg_type
= type(config
)
100 self
._create
_comp
(MySource
)
101 self
.assertIs(cfg_type
, bt2_component
._UserSourceComponentConfiguration
)
103 def test_filter_component_config(self
):
105 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
107 def __init__(comp_self
, config
, params
, obj
):
109 cfg_type
= type(config
)
112 self
._create
_comp
(MyFilter
)
113 self
.assertIs(cfg_type
, bt2_component
._UserFilterComponentConfiguration
)
115 def test_sink_component_config(self
):
116 class MySink(bt2
._UserSinkComponent
):
117 def __init__(comp_self
, config
, params
, obj
):
119 cfg_type
= type(config
)
121 def _user_consume(self
):
125 self
._create
_comp
(MySink
)
126 self
.assertIs(cfg_type
, bt2_component
._UserSinkComponentConfiguration
)
129 class GenericComponentTestCase(unittest
.TestCase
):
131 def _create_comp(comp_cls
, name
=None, log_level
=bt2
.LoggingLevel
.NONE
):
137 return graph
.add_component(comp_cls
, name
, logging_level
=log_level
)
140 class MySink(bt2
._UserSinkComponent
):
141 def _user_consume(self
):
144 comp
= self
._create
_comp
(MySink
, "yaes")
145 self
.assertEqual(comp
.name
, "yaes")
147 def test_logging_level(self
):
148 class MySink(bt2
._UserSinkComponent
):
149 def _user_consume(self
):
152 comp
= self
._create
_comp
(MySink
, "yaes", bt2
.LoggingLevel
.WARNING
)
153 self
.assertEqual(comp
.logging_level
, bt2
.LoggingLevel
.WARNING
)
155 def test_class(self
):
156 class MySink(bt2
._UserSinkComponent
):
157 def _user_consume(self
):
160 comp
= self
._create
_comp
(MySink
)
161 self
.assertEqual(comp
.cls
, MySink
)
164 class MySink(bt2
._UserSinkComponent
):
165 def _user_consume(self
):
168 comp
= self
._create
_comp
(MySink
)
169 self
.assertIsInstance(comp
.addr
, int)
170 self
.assertNotEqual(comp
.addr
, 0)
173 if __name__
== "__main__":