+#
+# Copyright (C) 2019 EfficiOS Inc.
+#
+# This program is free software; you can redistribute it and/or
+# modify it under the terms of the GNU General Public License
+# as published by the Free Software Foundation; only version 2
+# of the License.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program; if not, write to the Free Software
+# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
+#
+
from bt2 import value
import collections
import unittest
class _MyIter(bt2._UserMessageIterator):
- def __init__(self):
+ def __init__(self, self_output_port):
self._build_meta()
self._at = 0
comp = self._graph.add_component(MySink, 'salut')
assert comp
- comp2 = self._graph.add_component(comp.component_class, 'salut2')
+ comp2 = self._graph.add_component(comp.cls, 'salut2')
self.assertEqual(comp2.name, 'salut2')
def test_add_component_params(self):
with self.assertRaises(TypeError):
self._graph.add_component(int, 'salut')
+ def test_add_component_invalid_logging_level_type(self):
+ class MySink(bt2._UserSinkComponent):
+ def _consume(self):
+ pass
+
+ with self.assertRaises(TypeError):
+ self._graph.add_component(MySink, 'salut', logging_level='yo')
+
+ def test_add_component_invalid_logging_level_value(self):
+ class MySink(bt2._UserSinkComponent):
+ def _consume(self):
+ pass
+
+ with self.assertRaises(ValueError):
+ self._graph.add_component(MySink, 'salut', logging_level=12345)
+
+ def test_add_component_logging_level(self):
+ class MySink(bt2._UserSinkComponent):
+ def _consume(self):
+ pass
+
+ comp = self._graph.add_component(MySink, 'salut',
+ logging_level=bt2.LoggingLevel.DEBUG)
+ self.assertEqual(comp.logging_level, bt2.LoggingLevel.DEBUG)
+
def test_connect_ports(self):
class MyIter(bt2._UserMessageIterator):
def __next__(self):
self.assertIsInstance(msg, bt2.message._PacketBeginningMessage)
elif comp_self._at >= 2 and comp_self._at <= 6:
self.assertIsInstance(msg, bt2.message._EventMessage)
- self.assertEqual(msg.event.event_class.name, 'salut')
+ self.assertEqual(msg.event.cls.name, 'salut')
elif comp_self._at == 7:
self.assertIsInstance(msg, bt2.message._PacketEndMessage)
elif comp_self._at == 8:
def _consume(self):
raise bt2.Stop
- def ports_connected_listener(upstream_port, downstream_port):
+ def ports_connected_listener(upstream_component, upstream_port,
+ downstream_component, downstream_port):
raise ValueError('oh noes!')
graph = bt2.Graph()