X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=tests%2Fbindings%2Fpython%2Fbt2%2Ftest_stream_class.py;h=f8d6446c0fa0a39793a7651be4081138279db9fc;hb=cfbd7cf3bde05e8a6606478889dcd663604ef7b5;hp=a1e320aee43bcbede33836e8d05c4da835b05eca;hpb=976c241df475c9d99161a1100517c3c308074d7f;p=babeltrace.git diff --git a/tests/bindings/python/bt2/test_stream_class.py b/tests/bindings/python/bt2/test_stream_class.py index a1e320ae..f8d6446c 100644 --- a/tests/bindings/python/bt2/test_stream_class.py +++ b/tests/bindings/python/bt2/test_stream_class.py @@ -1,265 +1,332 @@ -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 +from utils import run_in_component_init -@unittest.skip("this is broken") class StreamClassTestCase(unittest.TestCase): def setUp(self): - self._packet_context_ft = bt2.StructureFieldType() - self._packet_context_ft.append_field('menu', bt2.FloatingPointNumberFieldType()) - self._packet_context_ft.append_field('sticker', bt2.StringFieldType()) - self._event_header_ft = bt2.StructureFieldType() - self._event_header_ft.append_field('id', bt2.IntegerFieldType(19)) - self._event_context_ft = bt2.StructureFieldType() - self._event_context_ft.append_field('msg', bt2.StringFieldType()) - self._ec1, self._ec2 = self._create_event_classes() - self._sc = bt2.StreamClass(name='my_stream_class', id=12, - packet_context_field_type=self._packet_context_ft, - event_header_field_type=self._event_header_ft, - event_context_field_type=self._event_context_ft, - event_classes=(self._ec1, self._ec2)) - - def tearDown(self): - del self._packet_context_ft - del self._event_header_ft - del self._event_context_ft - del self._ec1 - del self._sc - - def _create_event_classes(self): - context_ft = bt2.StructureFieldType() - context_ft.append_field('allo', bt2.StringFieldType()) - context_ft.append_field('zola', bt2.IntegerFieldType(18)) - payload_ft = bt2.StructureFieldType() - payload_ft.append_field('zoom', bt2.StringFieldType()) - ec1 = bt2.EventClass('event23', id=23, context_field_type=context_ft, - payload_field_type=payload_ft) - ec2 = bt2.EventClass('event17', id=17, context_field_type=payload_ft, - payload_field_type=context_ft) - return ec1, ec2 - - def test_create(self): - self.assertEqual(self._sc.name, 'my_stream_class') - self.assertEqual(self._sc.id, 12) - self.assertEqual(self._sc.packet_context_field_type, self._packet_context_ft) - self.assertEqual(self._sc.event_header_field_type, self._event_header_ft) - self.assertEqual(self._sc.event_context_field_type, self._event_context_ft) - self.assertEqual(self._sc[23], self._ec1) - self.assertEqual(self._sc[17], self._ec2) - self.assertEqual(len(self._sc), 2) - - def test_assign_name(self): - self._sc.name = 'lel' - self.assertEqual(self._sc.name, 'lel') - - def test_assign_invalid_name(self): + def f(comp_self): + tc = comp_self._create_trace_class(assigns_automatic_stream_class_id=True) + cc = comp_self._create_clock_class() + return tc, cc + + self._tc, self._cc = run_in_component_init(f) + self._trace = self._tc() + + def test_create_default(self): + sc = self._tc.create_stream_class() + + self.assertIsNone(sc.name) + self.assertIsNone(sc.packet_context_field_class) + self.assertIsNone(sc.event_common_context_field_class) + self.assertIsNone(sc.default_clock_class) + self.assertTrue(sc.assigns_automatic_event_class_id) + self.assertTrue(sc.assigns_automatic_stream_id) + self.assertFalse(sc.supports_packets) + self.assertFalse(sc.packets_have_beginning_default_clock_snapshot) + self.assertFalse(sc.packets_have_end_default_clock_snapshot) + self.assertFalse(sc.supports_discarded_events) + self.assertFalse(sc.discarded_events_have_default_clock_snapshots) + self.assertFalse(sc.supports_discarded_packets) + self.assertFalse(sc.discarded_packets_have_default_clock_snapshots) + + def test_create_name(self): + sc = self._tc.create_stream_class(name='bozo') + self.assertEqual(sc.name, 'bozo') + + def test_create_invalid_name(self): with self.assertRaises(TypeError): - self._sc.name = 17 + self._tc.create_stream_class(name=17) - def test_assign_id(self): - self._sc.id = 1717 - self.assertEqual(self._sc.id, 1717) + def test_create_packet_context_field_class(self): + fc = self._tc.create_structure_field_class() + sc = self._tc.create_stream_class( + packet_context_field_class=fc, supports_packets=True + ) + self.assertEqual(sc.packet_context_field_class, fc) - def test_assign_invalid_id(self): + def test_create_invalid_packet_context_field_class(self): with self.assertRaises(TypeError): - self._sc.id = 'lel' + self._tc.create_stream_class(packet_context_field_class=22) - def test_no_id(self): - sc = bt2.StreamClass() - self.assertIsNone(sc.id) + def test_create_invalid_packet_context_field_class_no_packets(self): + fc = self._tc.create_structure_field_class() - def test_assign_packet_context_field_type(self): - self._sc.packet_context_field_type = self._event_context_ft - self.assertEqual(self._sc.packet_context_field_type, self._event_context_ft) + with self.assertRaises(ValueError): + self._tc.create_stream_class(packet_context_field_class=fc) - def test_assign_no_packet_context_field_type(self): - self._sc.packet_context_field_type = None - self.assertIsNone(self._sc.packet_context_field_type) + def test_create_event_common_context_field_class(self): + fc = self._tc.create_structure_field_class() + sc = self._tc.create_stream_class(event_common_context_field_class=fc) + self.assertEqual(sc.event_common_context_field_class, fc) - def test_assign_invalid_packet_context_field_type(self): + def test_create_invalid_event_common_context_field_class(self): with self.assertRaises(TypeError): - self._sc.packet_context_field_type = 'lel' + self._tc.create_stream_class(event_common_context_field_class=22) - def test_assign_event_header_field_type(self): - self._sc.event_header_field_type = self._event_header_ft - self.assertEqual(self._sc.event_header_field_type, self._event_header_ft) + def test_create_default_clock_class(self): + sc = self._tc.create_stream_class(default_clock_class=self._cc) + self.assertEqual(sc.default_clock_class.addr, self._cc.addr) - def test_assign_no_event_header_field_type(self): - self._sc.event_header_field_type = None - self.assertIsNone(self._sc.event_header_field_type) + def test_create_invalid_default_clock_class(self): + with self.assertRaises(TypeError): + self._tc.create_stream_class(default_clock_class=12) + + def test_automatic_stream_ids(self): + sc = self._tc.create_stream_class(assigns_automatic_stream_id=True) + self.assertTrue(sc.assigns_automatic_stream_id) + + stream = self._trace.create_stream(sc) + self.assertIsNotNone(stream.id) + + def test_automatic_stream_ids_raises(self): + sc = self._tc.create_stream_class(assigns_automatic_stream_id=True) + self.assertTrue(sc.assigns_automatic_stream_id) + + with self.assertRaises(ValueError): + self._trace.create_stream(sc, id=123) + + def test_no_automatic_stream_ids(self): + sc = self._tc.create_stream_class(assigns_automatic_stream_id=False) + self.assertFalse(sc.assigns_automatic_stream_id) + + stream = self._trace.create_stream(sc, id=333) + self.assertEqual(stream.id, 333) + + def test_no_automatic_stream_ids_raises(self): + sc = self._tc.create_stream_class(assigns_automatic_stream_id=False) + self.assertFalse(sc.assigns_automatic_stream_id) + + with self.assertRaises(ValueError): + self._trace.create_stream(sc) + + def test_automatic_event_class_ids(self): + sc = self._tc.create_stream_class(assigns_automatic_event_class_id=True) + self.assertTrue(sc.assigns_automatic_event_class_id) + + ec = sc.create_event_class() + self.assertIsNotNone(ec.id) + + def test_automatic_event_class_ids_raises(self): + sc = self._tc.create_stream_class(assigns_automatic_event_class_id=True) + self.assertTrue(sc.assigns_automatic_event_class_id) + + with self.assertRaises(ValueError): + sc.create_event_class(id=123) + + def test_no_automatic_event_class_ids(self): + sc = self._tc.create_stream_class(assigns_automatic_event_class_id=False) + self.assertFalse(sc.assigns_automatic_event_class_id) + + ec = sc.create_event_class(id=333) + self.assertEqual(ec.id, 333) + + def test_no_automatic_event_class_ids_raises(self): + sc = self._tc.create_stream_class(assigns_automatic_event_class_id=False) + self.assertFalse(sc.assigns_automatic_event_class_id) + + with self.assertRaises(ValueError): + sc.create_event_class() + + def test_supports_packets_without_cs(self): + sc = self._tc.create_stream_class( + default_clock_class=self._cc, supports_packets=True + ) + self.assertTrue(sc.supports_packets) + self.assertFalse(sc.packets_have_beginning_default_clock_snapshot) + self.assertFalse(sc.packets_have_end_default_clock_snapshot) + + def test_supports_packets_with_begin_cs(self): + sc = self._tc.create_stream_class( + default_clock_class=self._cc, + supports_packets=True, + packets_have_beginning_default_clock_snapshot=True, + ) + self.assertTrue(sc.supports_packets) + self.assertTrue(sc.packets_have_beginning_default_clock_snapshot) + self.assertFalse(sc.packets_have_end_default_clock_snapshot) + + def test_supports_packets_with_end_cs(self): + sc = self._tc.create_stream_class( + default_clock_class=self._cc, + supports_packets=True, + packets_have_end_default_clock_snapshot=True, + ) + self.assertTrue(sc.supports_packets) + self.assertFalse(sc.packets_have_beginning_default_clock_snapshot) + self.assertTrue(sc.packets_have_end_default_clock_snapshot) + + def test_supports_packets_raises_type_error(self): + with self.assertRaises(TypeError): + sc = self._tc.create_stream_class( + default_clock_class=self._cc, supports_packets=23 + ) - def test_assign_invalid_event_header_field_type(self): + def test_packets_have_begin_default_cs_raises_type_error(self): with self.assertRaises(TypeError): - self._sc.event_header_field_type = 'lel' + sc = self._tc.create_stream_class( + default_clock_class=self._cc, + packets_have_beginning_default_clock_snapshot=23, + ) - def test_assign_event_context_field_type(self): - self._sc.event_context_field_type = self._packet_context_ft - self.assertEqual(self._sc.event_context_field_type, self._packet_context_ft) + def test_packets_have_end_default_cs_raises_type_error(self): + with self.assertRaises(TypeError): + sc = self._tc.create_stream_class( + default_clock_class=self._cc, packets_have_end_default_clock_snapshot=23 + ) + + def test_does_not_support_packets_raises_with_begin_cs(self): + with self.assertRaises(ValueError): + sc = self._tc.create_stream_class( + default_clock_class=self._cc, + packets_have_beginning_default_clock_snapshot=True, + ) + + def test_does_not_support_packets_raises_with_end_cs(self): + with self.assertRaises(ValueError): + sc = self._tc.create_stream_class( + default_clock_class=self._cc, + packets_have_end_default_clock_snapshot=True, + ) + + def test_supports_discarded_events_without_cs(self): + sc = self._tc.create_stream_class( + default_clock_class=self._cc, supports_discarded_events=True + ) + self.assertTrue(sc.supports_discarded_events) + self.assertFalse(sc.discarded_events_have_default_clock_snapshots) + + def test_supports_discarded_events_with_cs(self): + sc = self._tc.create_stream_class( + default_clock_class=self._cc, + supports_discarded_events=True, + discarded_events_have_default_clock_snapshots=True, + ) + self.assertTrue(sc.supports_discarded_events) + self.assertTrue(sc.discarded_events_have_default_clock_snapshots) + + def test_supports_discarded_events_raises_type_error(self): + with self.assertRaises(TypeError): + sc = self._tc.create_stream_class( + default_clock_class=self._cc, supports_discarded_events=23 + ) - def test_assign_no_event_context_field_type(self): - self._sc.event_context_field_type = None - self.assertIsNone(self._sc.event_context_field_type) + def test_discarded_events_have_default_cs_raises_type_error(self): + with self.assertRaises(TypeError): + sc = self._tc.create_stream_class( + default_clock_class=self._cc, + discarded_events_have_default_clock_snapshots=23, + ) + + def test_does_not_support_discarded_events_raises_with_cs(self): + with self.assertRaises(ValueError): + sc = self._tc.create_stream_class( + default_clock_class=self._cc, + discarded_events_have_default_clock_snapshots=True, + ) + + def test_supports_discarded_packets_without_cs(self): + sc = self._tc.create_stream_class( + default_clock_class=self._cc, + supports_discarded_packets=True, + supports_packets=True, + ) + self.assertTrue(sc.supports_discarded_packets) + self.assertFalse(sc.discarded_packets_have_default_clock_snapshots) + + def test_supports_discarded_packets_with_cs(self): + sc = self._tc.create_stream_class( + default_clock_class=self._cc, + supports_discarded_packets=True, + discarded_packets_have_default_clock_snapshots=True, + supports_packets=True, + ) + self.assertTrue(sc.supports_discarded_packets) + self.assertTrue(sc.discarded_packets_have_default_clock_snapshots) + + def test_supports_discarded_packets_raises_without_packet_support(self): + with self.assertRaises(ValueError): + sc = self._tc.create_stream_class( + default_clock_class=self._cc, supports_discarded_packets=True + ) + + def test_supports_discarded_packets_raises_type_error(self): + with self.assertRaises(TypeError): + sc = self._tc.create_stream_class( + default_clock_class=self._cc, + supports_discarded_packets=23, + supports_packets=True, + ) - def test_assign_invalid_event_context_field_type(self): + def test_discarded_packets_have_default_cs_raises_type_error(self): with self.assertRaises(TypeError): - self._sc.event_context_field_type = 'lel' - - def test_trace_prop_no_tc(self): - self.assertIsNone(self._sc.trace) - - def test_trace_prop(self): - tc = bt2.Trace() - tc.add_stream_class(self._sc) - self.assertEqual(self._sc.trace.addr, tc.addr) - - def _test_copy(self, cpy): - self.assertIsNot(cpy, self._sc) - self.assertNotEqual(cpy.addr, self._sc.addr) - self.assertEqual(cpy, self._sc) - - def test_copy(self): - cpy = copy.copy(self._sc) - self._test_copy(cpy) - self.assertEqual(self._sc.packet_context_field_type.addr, cpy.packet_context_field_type.addr) - self.assertEqual(self._sc.event_header_field_type.addr, cpy.event_header_field_type.addr) - self.assertEqual(self._sc.event_context_field_type.addr, cpy.event_context_field_type.addr) - - def test_deepcopy(self): - cpy = copy.deepcopy(self._sc) - self._test_copy(cpy) - self.assertNotEqual(self._sc.packet_context_field_type.addr, cpy.packet_context_field_type.addr) - self.assertNotEqual(self._sc.event_header_field_type.addr, cpy.event_header_field_type.addr) - self.assertNotEqual(self._sc.event_context_field_type.addr, cpy.event_context_field_type.addr) + sc = self._tc.create_stream_class( + default_clock_class=self._cc, + discarded_packets_have_default_clock_snapshots=23, + supports_packets=True, + ) + + def test_does_not_support_discarded_packets_raises_with_cs(self): + with self.assertRaises(ValueError): + sc = self._tc.create_stream_class( + default_clock_class=self._cc, + discarded_packets_have_default_clock_snapshots=True, + supports_packets=True, + ) + + def test_trace_class(self): + sc = self._tc.create_stream_class() + self.assertEqual(sc.trace_class.addr, self._tc.addr) + + def _create_stream_class_with_event_classes(self): + sc = self._tc.create_stream_class(assigns_automatic_event_class_id=False) + ec1 = sc.create_event_class(id=23) + ec2 = sc.create_event_class(id=17) + return sc, ec1, ec2 def test_getitem(self): - self.assertEqual(self._sc[23], self._ec1) - self.assertEqual(self._sc[17], self._ec2) + sc, ec1, ec2 = self._create_stream_class_with_event_classes() + + self.assertEqual(sc[23].addr, ec1.addr) + self.assertEqual(sc[17].addr, ec2.addr) def test_getitem_wrong_key_type(self): + sc, _, _ = self._create_stream_class_with_event_classes() + with self.assertRaises(TypeError): - self._sc['event23'] + sc['event23'] def test_getitem_wrong_key(self): + sc, _, _ = self._create_stream_class_with_event_classes() + with self.assertRaises(KeyError): - self._sc[19] + sc[19] def test_len(self): - self.assertEqual(len(self._sc), 2) + sc, _, _ = self._create_stream_class_with_event_classes() + + self.assertEqual(len(sc), 2) def test_iter(self): - for ec_id, event_class in self._sc.items(): - self.assertIsInstance(event_class, bt2.EventClass) - - if ec_id == 23: - self.assertEqual(event_class, self._ec1) - elif ec_id == 17: - self.assertEqual(event_class, self._ec2) - - def test_eq(self): - ec1, ec2 = self._create_event_classes() - sc1 = bt2.StreamClass(name='my_stream_class', id=12, - packet_context_field_type=self._packet_context_ft, - event_header_field_type=self._event_header_ft, - event_context_field_type=self._event_context_ft, - event_classes=(ec1, ec2)) - ec1, ec2 = self._create_event_classes() - sc2 = bt2.StreamClass(name='my_stream_class', id=12, - packet_context_field_type=self._packet_context_ft, - event_header_field_type=self._event_header_ft, - event_context_field_type=self._event_context_ft, - event_classes=(ec1, ec2)) - self.assertEqual(sc1, sc2) - - def test_ne_name(self): - ec1, ec2 = self._create_event_classes() - sc1 = bt2.StreamClass(name='my_stream_class1', id=12, - packet_context_field_type=self._packet_context_ft, - event_header_field_type=self._event_header_ft, - event_context_field_type=self._event_context_ft, - event_classes=(ec1, ec2)) - ec1, ec2 = self._create_event_classes() - sc2 = bt2.StreamClass(name='my_stream_class', id=12, - packet_context_field_type=self._packet_context_ft, - event_header_field_type=self._event_header_ft, - event_context_field_type=self._event_context_ft, - event_classes=(ec1, ec2)) - self.assertNotEqual(sc1, sc2) - - def test_ne_id(self): - ec1, ec2 = self._create_event_classes() - sc1 = bt2.StreamClass(name='my_stream_class', id=13, - packet_context_field_type=self._packet_context_ft, - event_header_field_type=self._event_header_ft, - event_context_field_type=self._event_context_ft, - event_classes=(ec1, ec2)) - ec1, ec2 = self._create_event_classes() - sc2 = bt2.StreamClass(name='my_stream_class', id=12, - packet_context_field_type=self._packet_context_ft, - event_header_field_type=self._event_header_ft, - event_context_field_type=self._event_context_ft, - event_classes=(ec1, ec2)) - self.assertNotEqual(sc1, sc2) - - def test_ne_packet_context_field_type(self): - ec1, ec2 = self._create_event_classes() - sc1 = bt2.StreamClass(name='my_stream_class', id=12, - packet_context_field_type=self._event_context_ft, - event_header_field_type=self._event_header_ft, - event_context_field_type=self._event_context_ft, - event_classes=(ec1, ec2)) - ec1, ec2 = self._create_event_classes() - sc2 = bt2.StreamClass(name='my_stream_class', id=12, - packet_context_field_type=self._packet_context_ft, - event_header_field_type=self._event_header_ft, - event_context_field_type=self._event_context_ft, - event_classes=(ec1, ec2)) - self.assertNotEqual(sc1, sc2) - - def test_ne_event_header_field_type(self): - ec1, ec2 = self._create_event_classes() - sc1 = bt2.StreamClass(name='my_stream_class', id=12, - packet_context_field_type=self._packet_context_ft, - event_context_field_type=self._event_context_ft, - event_classes=(ec1, ec2)) - ec1, ec2 = self._create_event_classes() - sc2 = bt2.StreamClass(name='my_stream_class', id=12, - packet_context_field_type=self._packet_context_ft, - event_header_field_type=self._event_header_ft, - event_context_field_type=self._event_context_ft, - event_classes=(ec1, ec2)) - self.assertNotEqual(sc1, sc2) - - def test_ne_event_context_field_type(self): - ec1, ec2 = self._create_event_classes() - sc1 = bt2.StreamClass(name='my_stream_class', id=12, - packet_context_field_type=self._packet_context_ft, - event_header_field_type=self._event_header_ft, - event_context_field_type=self._packet_context_ft, - event_classes=(ec1, ec2)) - ec1, ec2 = self._create_event_classes() - sc2 = bt2.StreamClass(name='my_stream_class', id=12, - packet_context_field_type=self._packet_context_ft, - event_header_field_type=self._event_header_ft, - event_context_field_type=self._event_context_ft, - event_classes=(ec1, ec2)) - self.assertNotEqual(sc1, sc2) - - def test_ne_event_class(self): - ec1, ec2 = self._create_event_classes() - sc1 = bt2.StreamClass(name='my_stream_class', id=12, - packet_context_field_type=self._packet_context_ft, - event_header_field_type=self._event_header_ft, - event_context_field_type=self._event_context_ft, - event_classes=(ec1,)) - ec1, ec2 = self._create_event_classes() - sc2 = bt2.StreamClass(name='my_stream_class', id=12, - packet_context_field_type=self._packet_context_ft, - event_header_field_type=self._event_header_ft, - event_context_field_type=self._event_context_ft, - event_classes=(ec1, ec2)) - self.assertNotEqual(sc1, sc2) - - def test_eq_invalid(self): - self.assertFalse(self._sc == 23) + sc, _, _ = self._create_stream_class_with_event_classes() + + ec_ids = sorted(sc) + self.assertEqual(ec_ids, [17, 23])