# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
#
-from bt2 import value
-import collections
import unittest
-import copy
import bt2
def tearDown(self):
del self._graph
- def test_create_empty(self):
- graph = bt2.Graph()
+ def test_create_default(self):
+ bt2.Graph()
+
+ def test_create_known_mip_version(self):
+ bt2.Graph(0)
+
+ def test_create_invalid_mip_version_type(self):
+ with self.assertRaises(TypeError):
+ bt2.Graph('')
+
+ def test_create_unknown_mip_version(self):
+ with self.assertRaisesRegex(ValueError, 'unknown MIP version'):
+ bt2.Graph(1)
def test_add_component_user_cls(self):
class MySink(bt2._UserSinkComponent):
pass
params = {'hello': 23, 'path': '/path/to/stuff'}
- comp = self._graph.add_component(MySink, 'salut', params)
+ self._graph.add_component(MySink, 'salut', params)
self.assertEqual(params, comp_params)
del comp_params
pass
obj = object()
- comp = self._graph.add_component(MySink, 'salut', obj=obj)
+ self._graph.add_component(MySink, 'salut', obj=obj)
self.assertIs(comp_obj, obj)
del comp_obj
def _user_consume(self):
pass
- comp = self._graph.add_component(MySink, 'salut')
+ self._graph.add_component(MySink, 'salut')
self.assertIsNone(comp_obj)
del comp_obj
def test_add_component_obj_non_python_comp_cls(self):
- comp_obj = None
-
plugin = bt2.find_plugin('text', find_in_user_dir=False, find_in_sys_dir=False)
assert plugin is not None
cc = plugin.source_component_classes['dmesg']
assert cc is not None
with self.assertRaises(ValueError):
- comp = self._graph.add_component(cc, 'salut', obj=57)
+ self._graph.add_component(cc, 'salut', obj=57)
def test_add_component_invalid_cls_type(self):
with self.assertRaises(TypeError):
sink = self._graph.add_component(MySink, 'sink')
with self.assertRaises(TypeError):
- conn = self._graph.connect_ports(
- sink.input_ports['in'], src.output_ports['out']
- )
+ self._graph.connect_ports(sink.input_ports['in'], src.output_ports['out'])
def test_add_interrupter(self):
class MyIter(bt2._UserMessageIterator):
msg = next(comp_self._msg_iter)
if comp_self._at == 0:
- self.assertIsInstance(msg, bt2._StreamBeginningMessage)
+ self.assertIs(type(msg), bt2._StreamBeginningMessageConst)
elif comp_self._at == 1:
- self.assertIsInstance(msg, bt2._PacketBeginningMessage)
+ self.assertIs(type(msg), bt2._PacketBeginningMessageConst)
elif comp_self._at >= 2 and comp_self._at <= 6:
- self.assertIsInstance(msg, bt2._EventMessage)
+ self.assertIs(type(msg), bt2._EventMessageConst)
self.assertEqual(msg.event.cls.name, 'salut')
elif comp_self._at == 7:
- self.assertIsInstance(msg, bt2._PacketEndMessage)
+ self.assertIs(type(msg), bt2._PacketEndMessageConst)
elif comp_self._at == 8:
- self.assertIsInstance(msg, bt2._StreamEndMessage)
+ self.assertIs(type(msg), bt2._StreamEndMessageConst)
comp_self._at += 1
src = self._graph.add_component(MySource, 'src')
sink = self._graph.add_component(MySink, 'sink')
- conn = self._graph.connect_ports(
- src.output_ports['out'], sink.input_ports['in']
- )
+ self._graph.connect_ports(src.output_ports['out'], sink.input_ports['in'])
self._graph.run()
def test_run_once(self):
run_count = 0
src = self._graph.add_component(MySource, 'src')
sink = self._graph.add_component(MySink, 'sink')
- conn = self._graph.connect_ports(
- src.output_ports['out'], sink.input_ports['in']
- )
+ self._graph.connect_ports(src.output_ports['out'], sink.input_ports['in'])
with self.assertRaises(bt2.TryAgain):
self._graph.run_once()
src = self._graph.add_component(MySource, 'src')
sink = self._graph.add_component(MySink, 'sink')
- conn = self._graph.connect_ports(
- src.output_ports['out'], sink.input_ports['in']
- )
+ self._graph.connect_ports(src.output_ports['out'], sink.input_ports['in'])
with self.assertRaises(bt2.Stop):
self._graph.run_once()
def _user_consume(comp_self):
msg = next(comp_self._msg_iter)
if comp_self._at == 0:
- self.assertIsInstance(msg, bt2._StreamBeginningMessage)
+ self.assertIs(type(msg), bt2._StreamBeginningMessageConst)
elif comp_self._at == 1:
- self.assertIsInstance(msg, bt2._PacketBeginningMessage)
+ self.assertIs(type(msg), bt2._PacketBeginningMessageConst)
elif comp_self._at == 2:
- self.assertIsInstance(msg, bt2._EventMessage)
+ self.assertIs(type(msg), bt2._EventMessageConst)
raise bt2.TryAgain
else:
pass
src = self._graph.add_component(MySource, 'src')
sink = self._graph.add_component(MySink, 'sink')
- conn = self._graph.connect_ports(
- src.output_ports['out'], sink.input_ports['in']
- )
+ self._graph.connect_ports(src.output_ports['out'], sink.input_ports['in'])
with self.assertRaises(bt2.TryAgain):
self._graph.run()
def _user_consume(comp_self):
msg = next(comp_self._msg_iter)
if comp_self._at == 0:
- self.assertIsInstance(msg, bt2._StreamBeginningMessage)
+ self.assertIs(type(msg), bt2._StreamBeginningMessageConst)
elif comp_self._at == 1:
- self.assertIsInstance(msg, bt2._PacketBeginningMessage)
+ self.assertIs(type(msg), bt2._PacketBeginningMessageConst)
elif comp_self._at == 2:
- self.assertIsInstance(msg, bt2._EventMessage)
+ self.assertIs(type(msg), bt2._EventMessageConst)
elif comp_self._at == 3:
nonlocal raised_in_sink
raised_in_sink = True
src = self._graph.add_component(MySource, 'src')
sink = self._graph.add_component(MySink, 'sink')
- conn = self._graph.connect_ports(
- src.output_ports['out'], sink.input_ports['in']
- )
+ self._graph.connect_ports(src.output_ports['out'], sink.input_ports['in'])
with self.assertRaises(bt2._Error):
self._graph.run()
with self.assertRaises(bt2._Error):
graph.connect_ports(up.output_ports['out'], down.input_ports['in'])
+
+
+if __name__ == '__main__':
+ unittest.main()