X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=tests%2Fbindings%2Fpython%2Fbt2%2Ftest_stream.py;h=d03c27601a30030be79bb2a824fa3691e1f3304c;hb=f0a42b33ac3951cd5cb2ee0f66ac04437a681621;hp=d1d37cc3215ca7e870c23be9c3aea091c51c93d0;hpb=976c241df475c9d99161a1100517c3c308074d7f;p=babeltrace.git diff --git a/tests/bindings/python/bt2/test_stream.py b/tests/bindings/python/bt2/test_stream.py index d1d37cc3..d03c2760 100644 --- a/tests/bindings/python/bt2/test_stream.py +++ b/tests/bindings/python/bt2/test_stream.py @@ -1,115 +1,91 @@ -from collections import OrderedDict -from bt2 import values +# +# 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. +# + import unittest -import copy -import bt2 +import utils +from utils import run_in_component_init +from bt2 import trace as bt2_trace +from bt2 import stream as bt2_stream +from bt2 import value as bt2_value +from bt2 import stream_class as bt2_stream_class -@unittest.skip("this is broken") class StreamTestCase(unittest.TestCase): def setUp(self): - self._stream = self._create_stream(stream_id=23) - - def tearDown(self): - del self._stream - - def _create_stream(self, name='my_stream', stream_id=None): - # event header - eh = bt2.StructureFieldType() - eh += OrderedDict(( - ('id', bt2.IntegerFieldType(8)), - ('ts', bt2.IntegerFieldType(32)), - )) - - # stream event context - sec = bt2.StructureFieldType() - sec += OrderedDict(( - ('cpu_id', bt2.IntegerFieldType(8)), - ('stuff', bt2.FloatingPointNumberFieldType()), - )) - - # packet context - pc = bt2.StructureFieldType() - pc += OrderedDict(( - ('something', bt2.IntegerFieldType(8)), - ('something_else', bt2.FloatingPointNumberFieldType()), - )) - - # stream class - sc = bt2.StreamClass() - sc.packet_context_field_type = pc - sc.event_header_field_type = eh - sc.event_context_field_type = sec - - # event context - ec = bt2.StructureFieldType() - ec += OrderedDict(( - ('ant', bt2.IntegerFieldType(16, is_signed=True)), - ('msg', bt2.StringFieldType()), - )) - - # event payload - ep = bt2.StructureFieldType() - ep += OrderedDict(( - ('giraffe', bt2.IntegerFieldType(32)), - ('gnu', bt2.IntegerFieldType(8)), - ('mosquito', bt2.IntegerFieldType(8)), - )) - - # event class - event_class = bt2.EventClass('ec') - event_class.context_field_type = ec - event_class.payload_field_type = ep - sc.add_event_class(event_class) - - # packet header - ph = bt2.StructureFieldType() - ph += OrderedDict(( - ('magic', bt2.IntegerFieldType(32)), - ('stream_id', bt2.IntegerFieldType(16)), - )) - - # trace c;ass - tc = bt2.Trace() - tc.packet_header_field_type = ph - tc.add_stream_class(sc) - - # stream - return sc(name=name, id=stream_id) - - def test_attr_stream_class(self): - self.assertIsNotNone(self._stream.stream_class) - - def test_attr_name(self): - self.assertEqual(self._stream.name, 'my_stream') - - def test_eq(self): - stream1 = self._create_stream(stream_id=17) - stream2 = self._create_stream(stream_id=17) - self.assertEqual(stream1, stream2) - - def test_ne_name(self): - stream1 = self._create_stream(stream_id=17) - stream2 = self._create_stream('lel', 17) - self.assertNotEqual(stream1, stream2) - - def test_ne_id(self): - stream1 = self._create_stream(stream_id=17) - stream2 = self._create_stream(stream_id=23) - self.assertNotEqual(stream1, stream2) - - def test_eq_invalid(self): - self.assertFalse(self._stream == 23) - - def _test_copy(self, func): - stream = self._create_stream() - cpy = func(stream) - self.assertIsNot(stream, cpy) - self.assertNotEqual(stream.addr, cpy.addr) - self.assertEqual(stream, cpy) - - def test_copy(self): - self._test_copy(copy.copy) - - def test_deepcopy(self): - self._test_copy(copy.deepcopy) + def f(comp_self): + return comp_self._create_trace_class() + + self._tc = run_in_component_init(f) + self._sc = self._tc.create_stream_class(assigns_automatic_stream_id=True) + self._tr = self._tc() + + def test_create_default(self): + stream = self._tr.create_stream(self._sc) + self.assertIsNone(stream.name) + self.assertIs(type(stream), bt2_stream._Stream) + self.assertEqual(len(stream.user_attributes), 0) + + def test_name(self): + stream = self._tr.create_stream(self._sc, name='équidistant') + self.assertEqual(stream.name, 'équidistant') + + def test_invalid_name(self): + with self.assertRaises(TypeError): + self._tr.create_stream(self._sc, name=22) + + def test_create_user_attributes(self): + stream = self._tr.create_stream(self._sc, user_attributes={'salut': 23}) + self.assertEqual(stream.user_attributes, {'salut': 23}) + self.assertIs(type(stream.user_attributes), bt2_value.MapValue) + + def test_const_user_attributes(self): + stream = utils.get_const_stream_beginning_message().stream + self.assertEqual(stream.user_attributes, {'salut': 23}) + self.assertIs(type(stream.user_attributes), bt2_value._MapValueConst) + + def test_create_invalid_user_attributes(self): + with self.assertRaises(TypeError): + self._tr.create_stream(self._sc, user_attributes=object()) + + def test_create_invalid_user_attributes_value_type(self): + with self.assertRaises(TypeError): + self._tr.create_stream(self._sc, user_attributes=23) + + def test_stream_class(self): + stream = self._tr.create_stream(self._sc) + self.assertEqual(stream.cls, self._sc) + self.assertIs(type(stream.cls), bt2_stream_class._StreamClass) + + def test_const_stream_class(self): + stream = utils.get_const_stream_beginning_message().stream + self.assertIs(type(stream.cls), bt2_stream_class._StreamClassConst) + + def test_trace(self): + stream = self._tr.create_stream(self._sc) + self.assertEqual(stream.trace.addr, self._tr.addr) + self.assertIs(type(stream.trace), bt2_trace._Trace) + + def test_const_trace(self): + stream = utils.get_const_stream_beginning_message().stream + self.assertIs(type(stream.trace), bt2_trace._TraceConst) + + def test_invalid_id(self): + sc = self._tc.create_stream_class(assigns_automatic_stream_id=False) + + with self.assertRaises(TypeError): + self._tr.create_stream(sc, id='string')