X-Git-Url: http://git.efficios.com/?p=babeltrace.git;a=blobdiff_plain;f=tests%2Fbindings%2Fpython%2Fbt2%2Ftest_stream_class.py;h=9010dcd4c7d68e6755f26a8e7488358b6def47c2;hp=ae005778b9d525a32b8190a5538cabd283d5fac2;hb=0235b0db7de5bcacdb3650c92461f2ce5eb2143d;hpb=811644b8fe5fb9946972a7ace9df02ed872f448a diff --git a/tests/bindings/python/bt2/test_stream_class.py b/tests/bindings/python/bt2/test_stream_class.py index ae005778..9010dcd4 100644 --- a/tests/bindings/python/bt2/test_stream_class.py +++ b/tests/bindings/python/bt2/test_stream_class.py @@ -1,264 +1,453 @@ -from bt2 import values +# SPDX-License-Identifier: GPL-2.0-only +# +# Copyright (C) 2019 EfficiOS Inc. +# + import unittest -import copy -import bt2 +from utils import run_in_component_init +from bt2 import stream_class as bt2_stream_class +from bt2 import trace_class as bt2_trace_class +from bt2 import clock_class as bt2_clock_class +from bt2 import event_class as bt2_event_class +from bt2 import field_class as bt2_field_class 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): - with self.assertRaises(TypeError): - self._sc.name = 17 - - def test_assign_id(self): - self._sc.id = 1717 - self.assertEqual(self._sc.id, 1717) - - def test_assign_invalid_id(self): - with self.assertRaises(TypeError): - self._sc.id = 'lel' - - def test_no_id(self): - sc = bt2.StreamClass() - self.assertIsNone(sc.id) - - 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) - - 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_assign_invalid_packet_context_field_type(self): - with self.assertRaises(TypeError): - self._sc.packet_context_field_type = 'lel' - - 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_assign_no_event_header_field_type(self): - self._sc.event_header_field_type = None - self.assertIsNone(self._sc.event_header_field_type) - - def test_assign_invalid_event_header_field_type(self): - with self.assertRaises(TypeError): - self._sc.event_header_field_type = 'lel' - - 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_assign_no_event_context_field_type(self): - self._sc.event_context_field_type = None - self.assertIsNone(self._sc.event_context_field_type) - - def test_assign_invalid_event_context_field_type(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) + 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.assertIs(type(sc), bt2_stream_class._StreamClass) + 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) + self.assertEqual(len(sc.user_attributes), 0) + + 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.assertRaisesRegex(TypeError, "'int' is not a 'str' object"): + self._tc.create_stream_class(name=17) + + self.assertEqual(len(self._tc), 0) + + 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) + self.assertIs( + type(sc.packet_context_field_class), bt2_field_class._StructureFieldClass + ) + + def test_create_invalid_packet_context_field_class(self): + with self.assertRaisesRegex( + TypeError, + "'int' is not a '' object", + ): + self._tc.create_stream_class( + packet_context_field_class=22, supports_packets=True + ) + + self.assertEqual(len(self._tc), 0) + + def test_create_invalid_packet_context_field_class_no_packets(self): + fc = self._tc.create_structure_field_class() + + with self.assertRaisesRegex( + ValueError, + "cannot have a packet context field class without supporting packets", + ): + self._tc.create_stream_class(packet_context_field_class=fc) + + self.assertEqual(len(self._tc), 0) + + 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) + self.assertIs( + type(sc.event_common_context_field_class), + bt2_field_class._StructureFieldClass, + ) + + def test_create_invalid_event_common_context_field_class(self): + with self.assertRaisesRegex( + TypeError, + "'int' is not a '' object", + ): + self._tc.create_stream_class(event_common_context_field_class=22) + + self.assertEqual(len(self._tc), 0) + + 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) + self.assertIs(type(sc.default_clock_class), bt2_clock_class._ClockClass) + + def test_create_invalid_default_clock_class(self): + with self.assertRaisesRegex( + TypeError, "'int' is not a '' object" + ): + self._tc.create_stream_class(default_clock_class=12) + + self.assertEqual(len(self._tc), 0) + + def test_create_user_attributes(self): + sc = self._tc.create_stream_class(user_attributes={'salut': 23}) + self.assertEqual(sc.user_attributes, {'salut': 23}) + + def test_create_invalid_user_attributes(self): + with self.assertRaisesRegex( + TypeError, "cannot create value object from 'object' object" + ): + self._tc.create_stream_class(user_attributes=object()) + + self.assertEqual(len(self._tc), 0) + + def test_create_invalid_user_attributes_value_type(self): + with self.assertRaisesRegex( + TypeError, + "'SignedIntegerValue' is not a '' object", + ): + self._tc.create_stream_class(user_attributes=23) + + self.assertEqual(len(self._tc), 0) + + 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.assertRaisesRegex( + ValueError, "id provided, but stream class assigns automatic stream ids" + ): + self._trace.create_stream(sc, id=123) + + self.assertEqual(len(self._trace), 0) + + def test_automatic_stream_ids_wrong_type(self): + with self.assertRaisesRegex(TypeError, "str' is not a 'bool' object"): + self._tc.create_stream_class(assigns_automatic_stream_id='True') + + self.assertEqual(len(self._tc), 0) + + 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.assertRaisesRegex( + ValueError, + "id not provided, but stream class does not assign automatic stream ids", + ): + self._trace.create_stream(sc) + + self.assertEqual(len(self._trace), 0) + + 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.assertRaisesRegex( + ValueError, + "id provided, but stream class assigns automatic event class ids", + ): + sc.create_event_class(id=123) + + self.assertEqual(len(sc), 0) + + def test_automatic_event_class_ids_wrong_type(self): + with self.assertRaisesRegex(TypeError, "'str' is not a 'bool' object"): + self._tc.create_stream_class(assigns_automatic_event_class_id='True') + + self.assertEqual(len(self._tc), 0) + + 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.assertRaisesRegex( + ValueError, + "id not provided, but stream class does not assign automatic event class ids", + ): + sc.create_event_class() + + self.assertEqual(len(sc), 0) + + 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.assertRaisesRegex(TypeError, "'int' is not a 'bool' object"): + self._tc.create_stream_class( + default_clock_class=self._cc, supports_packets=23 + ) + + self.assertEqual(len(self._tc), 0) + + def test_packets_have_begin_default_cs_raises_type_error(self): + with self.assertRaisesRegex(TypeError, "'int' is not a 'bool' object"): + self._tc.create_stream_class( + default_clock_class=self._cc, + packets_have_beginning_default_clock_snapshot=23, + ) + + self.assertEqual(len(self._tc), 0) + + def test_packets_have_end_default_cs_raises_type_error(self): + with self.assertRaisesRegex(TypeError, "'int' is not a 'bool' object"): + self._tc.create_stream_class( + default_clock_class=self._cc, packets_have_end_default_clock_snapshot=23 + ) + + self.assertEqual(len(self._tc), 0) + + def test_does_not_support_packets_raises_with_begin_cs(self): + with self.assertRaisesRegex( + ValueError, + "cannot not support packets, but have packet beginning default clock snapshot", + ): + self._tc.create_stream_class( + default_clock_class=self._cc, + packets_have_beginning_default_clock_snapshot=True, + ) + + self.assertEqual(len(self._tc), 0) + + def test_does_not_support_packets_raises_with_end_cs(self): + with self.assertRaisesRegex( + ValueError, + "cannot not support packets, but have packet end default clock snapshots", + ): + self._tc.create_stream_class( + default_clock_class=self._cc, + packets_have_end_default_clock_snapshot=True, + ) + + self.assertEqual(len(self._tc), 0) + + 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.assertRaisesRegex(TypeError, "'int' is not a 'bool' object"): + self._tc.create_stream_class( + default_clock_class=self._cc, supports_discarded_events=23 + ) + + self.assertEqual(len(self._tc), 0) + + def test_discarded_events_have_default_cs_raises_type_error(self): + with self.assertRaisesRegex(TypeError, "'int' is not a 'bool' object"): + self._tc.create_stream_class( + default_clock_class=self._cc, + discarded_events_have_default_clock_snapshots=23, + ) + + self.assertEqual(len(self._tc), 0) + + def test_does_not_support_discarded_events_raises_with_cs(self): + with self.assertRaisesRegex( + ValueError, + "cannot not support discarded events, but have default clock snapshots for discarded event messages", + ): + self._tc.create_stream_class( + default_clock_class=self._cc, + discarded_events_have_default_clock_snapshots=True, + ) + + self.assertEqual(len(self._tc), 0) + + 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.assertRaisesRegex( + ValueError, "cannot support discarded packets, but not support packets" + ): + self._tc.create_stream_class( + default_clock_class=self._cc, supports_discarded_packets=True + ) + + self.assertEqual(len(self._tc), 0) + + def test_supports_discarded_packets_raises_type_error(self): + with self.assertRaisesRegex(TypeError, "'int' is not a 'bool' object"): + self._tc.create_stream_class( + default_clock_class=self._cc, + supports_discarded_packets=23, + supports_packets=True, + ) + + self.assertEqual(len(self._tc), 0) + + def test_discarded_packets_have_default_cs_raises_type_error(self): + with self.assertRaisesRegex(TypeError, "'int' is not a 'bool' object"): + self._tc.create_stream_class( + default_clock_class=self._cc, + discarded_packets_have_default_clock_snapshots=23, + supports_packets=True, + ) + + self.assertEqual(len(self._tc), 0) + + def test_does_not_support_discarded_packets_raises_with_cs(self): + with self.assertRaisesRegex( + ValueError, + "cannot not support discarded packets, but have default clock snapshots for discarded packet messages", + ): + self._tc.create_stream_class( + default_clock_class=self._cc, + discarded_packets_have_default_clock_snapshots=True, + supports_packets=True, + ) + + self.assertEqual(len(self._tc), 0) + + def test_trace_class(self): + sc = self._tc.create_stream_class() + self.assertEqual(sc.trace_class.addr, self._tc.addr) + self.assertIs(type(sc.trace_class), bt2_trace_class._TraceClass) + + 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(type(sc[23]), bt2_event_class._EventClass) + self.assertEqual(sc[17].addr, ec2.addr) + self.assertEqual(type(sc[17]), bt2_event_class._EventClass) def test_getitem_wrong_key_type(self): - with self.assertRaises(TypeError): - self._sc['event23'] + sc, _, _ = self._create_stream_class_with_event_classes() + + with self.assertRaisesRegex(TypeError, "'str' is not an 'int' object"): + sc['event23'] def test_getitem_wrong_key(self): - with self.assertRaises(KeyError): - self._sc[19] + sc, _, _ = self._create_stream_class_with_event_classes() + + with self.assertRaisesRegex(KeyError, '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]) + + +if __name__ == '__main__': + unittest.main()