1 # SPDX-License-Identifier: GPL-2.0-only
3 # Copyright (C) 2019 EfficiOS Inc.
8 from bt2
import component
as bt2_component
11 class UserComponentTestCase(unittest
.TestCase
):
13 def _create_comp(comp_cls
, name
=None, log_level
=bt2
.LoggingLevel
.NONE
):
19 return graph
.add_component(comp_cls
, name
, logging_level
=log_level
)
22 class MySink(bt2
._UserSinkComponent
):
23 def __init__(comp_self
, config
, params
, obj
):
24 self
.assertEqual(comp_self
.name
, 'yaes')
26 def _user_consume(self
):
29 self
._create
_comp
(MySink
, 'yaes')
31 def test_logging_level(self
):
32 class MySink(bt2
._UserSinkComponent
):
33 def __init__(comp_self
, config
, params
, obj
):
34 self
.assertEqual(comp_self
.logging_level
, bt2
.LoggingLevel
.INFO
)
36 def _user_consume(self
):
39 self
._create
_comp
(MySink
, 'yaes', bt2
.LoggingLevel
.INFO
)
41 def test_graph_mip_version(self
):
42 class MySink(bt2
._UserSinkComponent
):
43 def __init__(comp_self
, config
, params
, obj
):
44 self
.assertEqual(comp_self
._graph
_mip
_version
, 0)
46 def _user_consume(self
):
49 self
._create
_comp
(MySink
, 'yaes', bt2
.LoggingLevel
.INFO
)
52 class MySink(bt2
._UserSinkComponent
):
53 def __init__(comp_self
, config
, params
, obj
):
54 self
.assertEqual(comp_self
.cls
, MySink
)
56 def _user_consume(self
):
59 self
._create
_comp
(MySink
)
62 class MySink(bt2
._UserSinkComponent
):
63 def __init__(comp_self
, config
, params
, obj
):
64 self
.assertIsInstance(comp_self
.addr
, int)
65 self
.assertNotEqual(comp_self
.addr
, 0)
67 def _user_consume(self
):
70 self
._create
_comp
(MySink
)
72 def test_finalize(self
):
75 class MySink(bt2
._UserSinkComponent
):
76 def _user_consume(self
):
79 def _user_finalize(comp_self
):
84 comp
= graph
.add_component(MySink
, 'lel')
88 self
.assertTrue(finalized
)
90 def test_source_component_config(self
):
92 bt2
._UserSourceComponent
, message_iterator_class
=bt2
._UserMessageIterator
94 def __init__(comp_self
, config
, params
, obj
):
96 cfg_type
= type(config
)
99 self
._create
_comp
(MySource
)
100 self
.assertIs(cfg_type
, bt2_component
._UserSourceComponentConfiguration
)
102 def test_filter_component_config(self
):
104 bt2
._UserFilterComponent
, message_iterator_class
=bt2
._UserMessageIterator
106 def __init__(comp_self
, config
, params
, obj
):
108 cfg_type
= type(config
)
111 self
._create
_comp
(MyFilter
)
112 self
.assertIs(cfg_type
, bt2_component
._UserFilterComponentConfiguration
)
114 def test_sink_component_config(self
):
115 class MySink(bt2
._UserSinkComponent
):
116 def __init__(comp_self
, config
, params
, obj
):
118 cfg_type
= type(config
)
120 def _user_consume(self
):
124 self
._create
_comp
(MySink
)
125 self
.assertIs(cfg_type
, bt2_component
._UserSinkComponentConfiguration
)
128 class GenericComponentTestCase(unittest
.TestCase
):
130 def _create_comp(comp_cls
, name
=None, log_level
=bt2
.LoggingLevel
.NONE
):
136 return graph
.add_component(comp_cls
, name
, logging_level
=log_level
)
139 class MySink(bt2
._UserSinkComponent
):
140 def _user_consume(self
):
143 comp
= self
._create
_comp
(MySink
, 'yaes')
144 self
.assertEqual(comp
.name
, 'yaes')
146 def test_logging_level(self
):
147 class MySink(bt2
._UserSinkComponent
):
148 def _user_consume(self
):
151 comp
= self
._create
_comp
(MySink
, 'yaes', bt2
.LoggingLevel
.WARNING
)
152 self
.assertEqual(comp
.logging_level
, bt2
.LoggingLevel
.WARNING
)
154 def test_class(self
):
155 class MySink(bt2
._UserSinkComponent
):
156 def _user_consume(self
):
159 comp
= self
._create
_comp
(MySink
)
160 self
.assertEqual(comp
.cls
, MySink
)
163 class MySink(bt2
._UserSinkComponent
):
164 def _user_consume(self
):
167 comp
= self
._create
_comp
(MySink
)
168 self
.assertIsInstance(comp
.addr
, int)
169 self
.assertNotEqual(comp
.addr
, 0)
172 if __name__
== '__main__':