[babeltrace.git] / tests / bindings / python / bt2 / test_trace.py

from bt2 import values
import unittest
import copy
import uuid
import bt2


@unittest.skip("this is broken")
class TraceTestCase(unittest.TestCase):
    def setUp(self):
        self._sc = self._create_stream_class('sc1', 3)
        self._tc = bt2.Trace()

    def tearDown(self):
        del self._sc
        del self._tc

    def _create_stream_class(self, name, id):
        ec1, ec2 = self._create_event_classes()
        packet_context_fc = bt2.StructureFieldClass()
        packet_context_fc.append_field('menu', bt2.FloatingPointNumberFieldClass())
        packet_context_fc.append_field('sticker', bt2.StringFieldClass())
        event_header_fc = bt2.StructureFieldClass()
        event_header_fc.append_field('id', bt2.IntegerFieldClass(19))
        event_context_fc = bt2.StructureFieldClass()
        event_context_fc.append_field('msg', bt2.StringFieldClass())
        return bt2.StreamClass(name=name, id=id,
                               packet_context_field_class=packet_context_fc,
                               event_header_field_class=event_header_fc,
                               event_context_field_class=event_context_fc,
                               event_classes=(ec1, ec2))

    def _create_event_classes(self):
        context_fc = bt2.StructureFieldClass()
        context_fc.append_field('allo', bt2.StringFieldClass())
        context_fc.append_field('zola', bt2.IntegerFieldClass(18))
        payload_fc = bt2.StructureFieldClass()
        payload_fc.append_field('zoom', bt2.StringFieldClass())
        ec1 = bt2.EventClass('event23', id=23, context_field_class=context_fc,
                             payload_field_class=payload_fc)
        ec2 = bt2.EventClass('event17', id=17, context_field_class=payload_fc,
                             payload_field_class=context_fc)
        return ec1, ec2

    def test_create_default(self):
        self.assertEqual(len(self._tc), 0)

    def _get_std_header(self):
        header_fc = bt2.StructureFieldClass()
        header_fc.append_field('magic', bt2.IntegerFieldClass(32))
        header_fc.append_field('stream_id', bt2.IntegerFieldClass(32))
        return header_fc

    def test_create_full(self):
        clock_classes = bt2.ClockClass('cc1', 1000), bt2.ClockClass('cc2', 30)
        sc = self._create_stream_class('sc1', 3)
        tc = bt2.Trace(name='my name',
                       native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN,
                       env={'the_string': 'value', 'the_int': 23},
                       packet_header_field_class=self._get_std_header(),
                       clock_classes=clock_classes,
                       stream_classes=(sc,))
        self.assertEqual(tc.name, 'my name')
        self.assertEqual(tc.native_byte_order, bt2.ByteOrder.LITTLE_ENDIAN)
        self.assertEqual(tc.env['the_string'], 'value')
        self.assertEqual(tc.env['the_int'], 23)
        self.assertEqual(tc.packet_header_field_class, self._get_std_header())
        self.assertEqual(tc.clock_classes['cc1'], clock_classes[0])
        self.assertEqual(tc.clock_classes['cc2'], clock_classes[1])
        self.assertEqual(tc[3], sc)

    def test_assign_name(self):
        self._tc.name = 'lel'
        self.assertEqual(self._tc.name, 'lel')

    def test_assign_invalid_name(self):
        with self.assertRaises(TypeError):
            self._tc.name = 17

    def test_assign_static(self):
        self._tc.set_is_static()
        self.assertTrue(self._tc.is_static)

    def test_assign_native_byte_order(self):
        self._tc.native_byte_order = bt2.ByteOrder.BIG_ENDIAN
        self.assertEqual(self._tc.native_byte_order, bt2.ByteOrder.BIG_ENDIAN)

    def test_assign_invalid_native_byte_order(self):
        with self.assertRaises(TypeError):
            self._tc.native_byte_order = 'lel'

    def test_assign_packet_header_field_class(self):
        header_fc = bt2.StructureFieldClass()
        header_fc.append_field('magic', bt2.IntegerFieldClass(32))
        self._tc.packet_header_field_class = header_fc
        self.assertEqual(self._tc.packet_header_field_class, header_fc)

    def test_assign_no_packet_header_field_class(self):
        self._tc.packet_header_field_class = None
        self.assertIsNone(self._tc.packet_header_field_class)

    def _test_copy(self, cpy):
        self.assertIsNot(cpy, self._tc)
        self.assertNotEqual(cpy.addr, self._tc.addr)
        self.assertEqual(cpy, self._tc)
        self.assertEqual(len(self._tc), len(cpy))

    def _pre_copy(self):
        self._tc.packet_header_field_class = self._get_std_header()
        self._tc.name = 'the trace class'
        sc1 = self._create_stream_class('sc1', 3)
        sc2 = self._create_stream_class('sc2', 9)
        sc3 = self._create_stream_class('sc3', 17)
        self._tc.add_clock_class(bt2.ClockClass('cc1', 1000))
        self._tc.add_clock_class(bt2.ClockClass('cc2', 30))
        self._tc.env['allo'] = 'bateau'
        self._tc.env['bateau'] = 'cart'
        self._tc.add_stream_class(sc1)
        self._tc.add_stream_class(sc2)
        self._tc.add_stream_class(sc3)

    def test_copy(self):
        self._pre_copy()
        cpy = copy.copy(self._tc)
        self._test_copy(cpy)
        self.assertEqual(self._tc.packet_header_field_class.addr, cpy.packet_header_field_class.addr)
        self.assertEqual(self._tc.clock_classes['cc1'].addr, cpy.clock_classes['cc1'].addr)
        self.assertEqual(self._tc.clock_classes['cc2'].addr, cpy.clock_classes['cc2'].addr)
        self.assertEqual(self._tc.env['allo'].addr, cpy.env['allo'].addr)
        self.assertEqual(self._tc.env['bateau'].addr, cpy.env['bateau'].addr)

    def test_deepcopy(self):
        self._pre_copy()
        cpy = copy.deepcopy(self._tc)
        self._test_copy(cpy)
        self.assertNotEqual(self._tc.packet_header_field_class.addr, cpy.packet_header_field_class.addr)
        self.assertNotEqual(self._tc.clock_classes['cc1'].addr, cpy.clock_classes['cc1'].addr)
        self.assertNotEqual(self._tc.clock_classes['cc2'].addr, cpy.clock_classes['cc2'].addr)
        self.assertNotEqual(self._tc.env['allo'].addr, cpy.env['allo'].addr)
        self.assertNotEqual(self._tc.env['bateau'].addr, cpy.env['bateau'].addr)

    def test_getitem(self):
        self._tc.add_stream_class(self._sc)
        self.assertEqual(self._tc[3].addr, self._sc.addr)

    def test_getitem_wrong_key_type(self):
        self._tc.add_stream_class(self._sc)
        with self.assertRaises(TypeError):
            self._tc['hello']

    def test_getitem_wrong_key(self):
        self._tc.add_stream_class(self._sc)
        with self.assertRaises(KeyError):
            self._tc[4]

    def test_len(self):
        self.assertEqual(len(self._tc), 0)
        self._tc.add_stream_class(self._sc)
        self.assertEqual(len(self._tc), 1)

    def test_iter(self):
        self._tc.packet_header_field_class = self._get_std_header()
        sc1 = self._create_stream_class('sc1', 3)
        sc2 = self._create_stream_class('sc2', 9)
        sc3 = self._create_stream_class('sc3', 17)
        self._tc.add_stream_class(sc1)
        self._tc.add_stream_class(sc2)
        self._tc.add_stream_class(sc3)

        for sid, stream_class in self._tc.items():
            self.assertIsInstance(stream_class, bt2.StreamClass)

            if sid == 3:
                self.assertEqual(stream_class.addr, sc1.addr)
            elif sid == 9:
                self.assertEqual(stream_class.addr, sc2.addr)
            elif sid == 17:
                self.assertEqual(stream_class.addr, sc3.addr)

    def test_env_getitem_wrong_key(self):
        with self.assertRaises(KeyError):
            self._tc.env['lel']

    def test_clock_classes_getitem_wrong_key(self):
        with self.assertRaises(KeyError):
            self._tc.clock_classes['lel']

    def test_streams_none(self):
        self.assertEqual(len(self._tc.streams), 0)

    def test_streams_len(self):
        self._tc.add_stream_class(self._create_stream_class('sc1', 3))
        stream0 = self._tc[3]()
        stream1 = self._tc[3]()
        stream2 = self._tc[3]()
        self.assertEqual(len(self._tc.streams), 3)

    def test_streams_iter(self):
        self._tc.add_stream_class(self._create_stream_class('sc1', 3))
        stream0 = self._tc[3](id=12)
        stream1 = self._tc[3](id=15)
        stream2 = self._tc[3](id=17)
        sids = set()

        for stream in self._tc.streams:
            sids.add(stream.id)

        self.assertEqual(len(sids), 3)
        self.assertTrue(12 in sids and 15 in sids and 17 in sids)

    def _test_eq_create_objects(self):
        cc1_uuid = uuid.UUID('bc7f2f2d-2ee4-4e03-ab1f-2e0e1304e94f')
        cc1 = bt2.ClockClass('cc1', 1000, uuid=cc1_uuid)
        cc2_uuid = uuid.UUID('da7d6b6f-3108-4706-89bd-ab554732611b')
        cc2 = bt2.ClockClass('cc2', 30, uuid=cc2_uuid)
        sc1 = self._create_stream_class('sc1', 3)
        sc2 = self._create_stream_class('sc2', 9)
        return cc1, cc2, sc1, sc2, self._get_std_header()

    def test_eq(self):
        cc1, cc2, sc1, sc2, header_fc = self._test_eq_create_objects()
        tc1 = bt2.Trace(name='my name',
                        native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN,
                        env={'the_string': 'value', 'the_int': 23},
                        packet_header_field_class=header_fc,
                        clock_classes=(cc1, cc2),
                        stream_classes=(sc1, sc2))
        cc1, cc2, sc1, sc2, header_fc = self._test_eq_create_objects()
        tc2 = bt2.Trace(name='my name',
                        native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN,
                        env={'the_string': 'value', 'the_int': 23},
                        packet_header_field_class=header_fc,
                        clock_classes=(cc1, cc2),
                        stream_classes=(sc1, sc2))
        self.assertEqual(tc1, tc2)

    def test_ne_name(self):
        cc1, cc2, sc1, sc2, header_fc = self._test_eq_create_objects()
        tc1 = bt2.Trace(name='my name2',
                        native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN,
                        env={'the_string': 'value', 'the_int': 23},
                        packet_header_field_class=header_fc,
                        clock_classes=(cc1, cc2),
                        stream_classes=(sc1, sc2))
        cc1, cc2, sc1, sc2, header_fc = self._test_eq_create_objects()
        tc2 = bt2.Trace(name='my name',
                        native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN,
                        env={'the_string': 'value', 'the_int': 23},
                        packet_header_field_class=header_fc,
                        clock_classes=(cc1, cc2),
                        stream_classes=(sc1, sc2))
        self.assertNotEqual(tc1, tc2)

    def test_ne_packet_header_field_class(self):
        cc1, cc2, sc1, sc2, header_fc = self._test_eq_create_objects()
        tc1 = bt2.Trace(name='my name',
                        native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN,
                        env={'the_string': 'value', 'the_int': 23},
                        packet_header_field_class=header_fc,
                        clock_classes=(cc1, cc2),
                        stream_classes=(sc1, sc2))
        cc1, cc2, sc1, sc2, header_fc = self._test_eq_create_objects()
        header_fc.append_field('yes', bt2.StringFieldClass())
        tc2 = bt2.Trace(name='my name',
                        native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN,
                        env={'the_string': 'value', 'the_int': 23},
                        packet_header_field_class=header_fc,
                        clock_classes=(cc1, cc2),
                        stream_classes=(sc1, sc2))
        self.assertNotEqual(tc1, tc2)

    def test_ne_native_byte_order(self):
        cc1, cc2, sc1, sc2, header_fc = self._test_eq_create_objects()
        tc1 = bt2.Trace(name='my name',
                        native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN,
                        env={'the_string': 'value', 'the_int': 23},
                        packet_header_field_class=header_fc,
                        clock_classes=(cc1, cc2),
                        stream_classes=(sc1, sc2))
        cc1, cc2, sc1, sc2, header_fc = self._test_eq_create_objects()
        tc2 = bt2.Trace(name='my name',
                        native_byte_order=bt2.ByteOrder.BIG_ENDIAN,
                        env={'the_string': 'value', 'the_int': 23},
                        packet_header_field_class=header_fc,
                        clock_classes=(cc1, cc2),
                        stream_classes=(sc1, sc2))
        self.assertNotEqual(tc1, tc2)

    def test_ne_env(self):
        cc1, cc2, sc1, sc2, header_fc = self._test_eq_create_objects()
        tc1 = bt2.Trace(name='my name',
                        native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN,
                        env={'the_string': 'value', 'the_int2': 23},
                        packet_header_field_class=header_fc,
                        clock_classes=(cc1, cc2),
                        stream_classes=(sc1, sc2))
        cc1, cc2, sc1, sc2, header_fc = self._test_eq_create_objects()
        tc2 = bt2.Trace(name='my name',
                        native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN,
                        env={'the_string': 'value', 'the_int': 23},
                        packet_header_field_class=header_fc,
                        clock_classes=(cc1, cc2),
                        stream_classes=(sc1, sc2))
        self.assertNotEqual(tc1, tc2)

    def test_ne_clock_classes(self):
        cc1, cc2, sc1, sc2, header_fc = self._test_eq_create_objects()
        tc1 = bt2.Trace(name='my name',
                        native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN,
                        env={'the_string': 'value', 'the_int': 23},
                        packet_header_field_class=header_fc,
                        clock_classes=(cc1, cc2),
                        stream_classes=(sc1, sc2))
        cc1, cc2, sc1, sc2, header_fc = self._test_eq_create_objects()
        cc2.frequency = 1234
        tc2 = bt2.Trace(name='my name',
                        native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN,
                        env={'the_string': 'value', 'the_int': 23},
                        packet_header_field_class=header_fc,
                        clock_classes=(cc1, cc2),
                        stream_classes=(sc1, sc2))
        self.assertNotEqual(tc1, tc2)

    def test_ne_stream_classes(self):
        cc1, cc2, sc1, sc2, header_fc = self._test_eq_create_objects()
        tc1 = bt2.Trace(name='my name',
                        native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN,
                        env={'the_string': 'value', 'the_int': 23},
                        packet_header_field_class=header_fc,
                        clock_classes=(cc1, cc2),
                        stream_classes=(sc1, sc2))
        cc1, cc2, sc1, sc2, header_fc = self._test_eq_create_objects()
        sc2.id = 72632
        tc2 = bt2.Trace(name='my name',
                        native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN,
                        env={'the_string': 'value', 'the_int': 23},
                        packet_header_field_class=header_fc,
                        clock_classes=(cc1, cc2),
                        stream_classes=(sc1, sc2))
        self.assertNotEqual(tc1, tc2)

    def test_eq_invalid(self):
        self.assertFalse(self._tc == 23)
Commit	Line	Data
9cf643d1 PP	1	from bt2 import values
	2	import unittest
	3	import copy
	4	import uuid
	5	import bt2
	6
	7
90cfc012	8	@unittest.skip("this is broken")
9cf643d1 PP	9	class TraceTestCase(unittest.TestCase):
	10	def setUp(self):
	11	self._sc = self._create_stream_class('sc1', 3)
	12	self._tc = bt2.Trace()
	13
f6a5e476 PP	14	def tearDown(self):
	15	del self._sc
	16	del self._tc
	17
9cf643d1 PP	18	def _create_stream_class(self, name, id):
9cf643d1 PP	19	ec1, ec2 = self._create_event_classes()
32656995 SM	20	packet_context_fc = bt2.StructureFieldClass()
	21	packet_context_fc.append_field('menu', bt2.FloatingPointNumberFieldClass())
	22	packet_context_fc.append_field('sticker', bt2.StringFieldClass())
	23	event_header_fc = bt2.StructureFieldClass()
	24	event_header_fc.append_field('id', bt2.IntegerFieldClass(19))
	25	event_context_fc = bt2.StructureFieldClass()
	26	event_context_fc.append_field('msg', bt2.StringFieldClass())
9cf643d1	27	return bt2.StreamClass(name=name, id=id,
32656995 SM	28	packet_context_field_class=packet_context_fc,
	29	event_header_field_class=event_header_fc,
	30	event_context_field_class=event_context_fc,
9cf643d1 PP	31	event_classes=(ec1, ec2))
	32
	33	def _create_event_classes(self):
32656995 SM	34	context_fc = bt2.StructureFieldClass()
	35	context_fc.append_field('allo', bt2.StringFieldClass())
	36	context_fc.append_field('zola', bt2.IntegerFieldClass(18))
	37	payload_fc = bt2.StructureFieldClass()
	38	payload_fc.append_field('zoom', bt2.StringFieldClass())
	39	ec1 = bt2.EventClass('event23', id=23, context_field_class=context_fc,
	40	payload_field_class=payload_fc)
	41	ec2 = bt2.EventClass('event17', id=17, context_field_class=payload_fc,
	42	payload_field_class=context_fc)
9cf643d1 PP	43	return ec1, ec2
	44
	45	def test_create_default(self):
	46	self.assertEqual(len(self._tc), 0)
	47
f6a5e476	48	def _get_std_header(self):
32656995 SM	49	header_fc = bt2.StructureFieldClass()
	50	header_fc.append_field('magic', bt2.IntegerFieldClass(32))
	51	header_fc.append_field('stream_id', bt2.IntegerFieldClass(32))
	52	return header_fc
f6a5e476 PP	53
	54	def test_create_full(self):
	55	clock_classes = bt2.ClockClass('cc1', 1000), bt2.ClockClass('cc2', 30)
9cf643d1 PP	56	sc = self._create_stream_class('sc1', 3)
	57	tc = bt2.Trace(name='my name',
	58	native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN,
	59	env={'the_string': 'value', 'the_int': 23},
32656995	60	packet_header_field_class=self._get_std_header(),
9cf643d1 PP	61	clock_classes=clock_classes,
	62	stream_classes=(sc,))
	63	self.assertEqual(tc.name, 'my name')
	64	self.assertEqual(tc.native_byte_order, bt2.ByteOrder.LITTLE_ENDIAN)
	65	self.assertEqual(tc.env['the_string'], 'value')
	66	self.assertEqual(tc.env['the_int'], 23)
32656995	67	self.assertEqual(tc.packet_header_field_class, self._get_std_header())
9cf643d1 PP	68	self.assertEqual(tc.clock_classes['cc1'], clock_classes[0])
	69	self.assertEqual(tc.clock_classes['cc2'], clock_classes[1])
	70	self.assertEqual(tc[3], sc)
	71
	72	def test_assign_name(self):
	73	self._tc.name = 'lel'
	74	self.assertEqual(self._tc.name, 'lel')
	75
	76	def test_assign_invalid_name(self):
	77	with self.assertRaises(TypeError):
	78	self._tc.name = 17
	79
f6a5e476 PP	80	def test_assign_static(self):
	81	self._tc.set_is_static()
	82	self.assertTrue(self._tc.is_static)
	83
9cf643d1 PP	84	def test_assign_native_byte_order(self):
	85	self._tc.native_byte_order = bt2.ByteOrder.BIG_ENDIAN
	86	self.assertEqual(self._tc.native_byte_order, bt2.ByteOrder.BIG_ENDIAN)
	87
	88	def test_assign_invalid_native_byte_order(self):
	89	with self.assertRaises(TypeError):
	90	self._tc.native_byte_order = 'lel'
	91
32656995 SM	92	def test_assign_packet_header_field_class(self):
	93	header_fc = bt2.StructureFieldClass()
	94	header_fc.append_field('magic', bt2.IntegerFieldClass(32))
	95	self._tc.packet_header_field_class = header_fc
	96	self.assertEqual(self._tc.packet_header_field_class, header_fc)
9cf643d1	97
32656995 SM	98	def test_assign_no_packet_header_field_class(self):
	99	self._tc.packet_header_field_class = None
	100	self.assertIsNone(self._tc.packet_header_field_class)
9cf643d1 PP	101
	102	def _test_copy(self, cpy):
	103	self.assertIsNot(cpy, self._tc)
	104	self.assertNotEqual(cpy.addr, self._tc.addr)
	105	self.assertEqual(cpy, self._tc)
	106	self.assertEqual(len(self._tc), len(cpy))
	107
	108	def _pre_copy(self):
32656995	109	self._tc.packet_header_field_class = self._get_std_header()
9cf643d1 PP	110	self._tc.name = 'the trace class'
	111	sc1 = self._create_stream_class('sc1', 3)
	112	sc2 = self._create_stream_class('sc2', 9)
	113	sc3 = self._create_stream_class('sc3', 17)
f6a5e476 PP	114	self._tc.add_clock_class(bt2.ClockClass('cc1', 1000))
f6a5e476 PP	115	self._tc.add_clock_class(bt2.ClockClass('cc2', 30))
9cf643d1 PP	116	self._tc.env['allo'] = 'bateau'
	117	self._tc.env['bateau'] = 'cart'
	118	self._tc.add_stream_class(sc1)
	119	self._tc.add_stream_class(sc2)
	120	self._tc.add_stream_class(sc3)
	121
	122	def test_copy(self):
	123	self._pre_copy()
	124	cpy = copy.copy(self._tc)
	125	self._test_copy(cpy)
32656995	126	self.assertEqual(self._tc.packet_header_field_class.addr, cpy.packet_header_field_class.addr)
9cf643d1 PP	127	self.assertEqual(self._tc.clock_classes['cc1'].addr, cpy.clock_classes['cc1'].addr)
	128	self.assertEqual(self._tc.clock_classes['cc2'].addr, cpy.clock_classes['cc2'].addr)
	129	self.assertEqual(self._tc.env['allo'].addr, cpy.env['allo'].addr)
	130	self.assertEqual(self._tc.env['bateau'].addr, cpy.env['bateau'].addr)
	131
	132	def test_deepcopy(self):
	133	self._pre_copy()
	134	cpy = copy.deepcopy(self._tc)
	135	self._test_copy(cpy)
32656995	136	self.assertNotEqual(self._tc.packet_header_field_class.addr, cpy.packet_header_field_class.addr)
9cf643d1 PP	137	self.assertNotEqual(self._tc.clock_classes['cc1'].addr, cpy.clock_classes['cc1'].addr)
	138	self.assertNotEqual(self._tc.clock_classes['cc2'].addr, cpy.clock_classes['cc2'].addr)
	139	self.assertNotEqual(self._tc.env['allo'].addr, cpy.env['allo'].addr)
	140	self.assertNotEqual(self._tc.env['bateau'].addr, cpy.env['bateau'].addr)
	141
	142	def test_getitem(self):
	143	self._tc.add_stream_class(self._sc)
	144	self.assertEqual(self._tc[3].addr, self._sc.addr)
	145
	146	def test_getitem_wrong_key_type(self):
	147	self._tc.add_stream_class(self._sc)
	148	with self.assertRaises(TypeError):
	149	self._tc['hello']
	150
	151	def test_getitem_wrong_key(self):
	152	self._tc.add_stream_class(self._sc)
	153	with self.assertRaises(KeyError):
	154	self._tc[4]
	155
	156	def test_len(self):
	157	self.assertEqual(len(self._tc), 0)
	158	self._tc.add_stream_class(self._sc)
	159	self.assertEqual(len(self._tc), 1)
	160
	161	def test_iter(self):
32656995	162	self._tc.packet_header_field_class = self._get_std_header()
9cf643d1 PP	163	sc1 = self._create_stream_class('sc1', 3)
	164	sc2 = self._create_stream_class('sc2', 9)
	165	sc3 = self._create_stream_class('sc3', 17)
	166	self._tc.add_stream_class(sc1)
	167	self._tc.add_stream_class(sc2)
	168	self._tc.add_stream_class(sc3)
	169
	170	for sid, stream_class in self._tc.items():
	171	self.assertIsInstance(stream_class, bt2.StreamClass)
	172
	173	if sid == 3:
	174	self.assertEqual(stream_class.addr, sc1.addr)
	175	elif sid == 9:
	176	self.assertEqual(stream_class.addr, sc2.addr)
	177	elif sid == 17:
	178	self.assertEqual(stream_class.addr, sc3.addr)
	179
	180	def test_env_getitem_wrong_key(self):
	181	with self.assertRaises(KeyError):
	182	self._tc.env['lel']
	183
	184	def test_clock_classes_getitem_wrong_key(self):
	185	with self.assertRaises(KeyError):
	186	self._tc.clock_classes['lel']
	187
f6a5e476 PP	188	def test_streams_none(self):
	189	self.assertEqual(len(self._tc.streams), 0)
	190
	191	def test_streams_len(self):
	192	self._tc.add_stream_class(self._create_stream_class('sc1', 3))
	193	stream0 = self._tc[3]()
	194	stream1 = self._tc[3]()
	195	stream2 = self._tc[3]()
	196	self.assertEqual(len(self._tc.streams), 3)
	197
	198	def test_streams_iter(self):
	199	self._tc.add_stream_class(self._create_stream_class('sc1', 3))
	200	stream0 = self._tc[3](id=12)
	201	stream1 = self._tc[3](id=15)
	202	stream2 = self._tc[3](id=17)
	203	sids = set()
	204
	205	for stream in self._tc.streams:
	206	sids.add(stream.id)
	207
	208	self.assertEqual(len(sids), 3)
	209	self.assertTrue(12 in sids and 15 in sids and 17 in sids)
	210
9cf643d1 PP	211	def _test_eq_create_objects(self):
9cf643d1 PP	212	cc1_uuid = uuid.UUID('bc7f2f2d-2ee4-4e03-ab1f-2e0e1304e94f')
f6a5e476	213	cc1 = bt2.ClockClass('cc1', 1000, uuid=cc1_uuid)
9cf643d1	214	cc2_uuid = uuid.UUID('da7d6b6f-3108-4706-89bd-ab554732611b')
f6a5e476	215	cc2 = bt2.ClockClass('cc2', 30, uuid=cc2_uuid)
9cf643d1 PP	216	sc1 = self._create_stream_class('sc1', 3)
9cf643d1 PP	217	sc2 = self._create_stream_class('sc2', 9)
f6a5e476	218	return cc1, cc2, sc1, sc2, self._get_std_header()
9cf643d1 PP	219
9cf643d1 PP	220	def test_eq(self):
32656995	221	cc1, cc2, sc1, sc2, header_fc = self._test_eq_create_objects()
9cf643d1 PP	222	tc1 = bt2.Trace(name='my name',
	223	native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN,
	224	env={'the_string': 'value', 'the_int': 23},
32656995	225	packet_header_field_class=header_fc,
9cf643d1 PP	226	clock_classes=(cc1, cc2),
9cf643d1 PP	227	stream_classes=(sc1, sc2))
32656995	228	cc1, cc2, sc1, sc2, header_fc = self._test_eq_create_objects()
9cf643d1 PP	229	tc2 = bt2.Trace(name='my name',
	230	native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN,
	231	env={'the_string': 'value', 'the_int': 23},
32656995	232	packet_header_field_class=header_fc,
9cf643d1 PP	233	clock_classes=(cc1, cc2),
	234	stream_classes=(sc1, sc2))
	235	self.assertEqual(tc1, tc2)
	236
	237	def test_ne_name(self):
32656995	238	cc1, cc2, sc1, sc2, header_fc = self._test_eq_create_objects()
9cf643d1 PP	239	tc1 = bt2.Trace(name='my name2',
	240	native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN,
	241	env={'the_string': 'value', 'the_int': 23},
32656995	242	packet_header_field_class=header_fc,
9cf643d1 PP	243	clock_classes=(cc1, cc2),
9cf643d1 PP	244	stream_classes=(sc1, sc2))
32656995	245	cc1, cc2, sc1, sc2, header_fc = self._test_eq_create_objects()
9cf643d1 PP	246	tc2 = bt2.Trace(name='my name',
	247	native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN,
	248	env={'the_string': 'value', 'the_int': 23},
32656995	249	packet_header_field_class=header_fc,
9cf643d1 PP	250	clock_classes=(cc1, cc2),
	251	stream_classes=(sc1, sc2))
	252	self.assertNotEqual(tc1, tc2)
	253
32656995 SM	254	def test_ne_packet_header_field_class(self):
32656995 SM	255	cc1, cc2, sc1, sc2, header_fc = self._test_eq_create_objects()
9cf643d1 PP	256	tc1 = bt2.Trace(name='my name',
	257	native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN,
	258	env={'the_string': 'value', 'the_int': 23},
32656995	259	packet_header_field_class=header_fc,
9cf643d1 PP	260	clock_classes=(cc1, cc2),
9cf643d1 PP	261	stream_classes=(sc1, sc2))
32656995 SM	262	cc1, cc2, sc1, sc2, header_fc = self._test_eq_create_objects()
32656995 SM	263	header_fc.append_field('yes', bt2.StringFieldClass())
9cf643d1 PP	264	tc2 = bt2.Trace(name='my name',
	265	native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN,
	266	env={'the_string': 'value', 'the_int': 23},
32656995	267	packet_header_field_class=header_fc,
9cf643d1 PP	268	clock_classes=(cc1, cc2),
	269	stream_classes=(sc1, sc2))
	270	self.assertNotEqual(tc1, tc2)
	271
	272	def test_ne_native_byte_order(self):
32656995	273	cc1, cc2, sc1, sc2, header_fc = self._test_eq_create_objects()
9cf643d1 PP	274	tc1 = bt2.Trace(name='my name',
	275	native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN,
	276	env={'the_string': 'value', 'the_int': 23},
32656995	277	packet_header_field_class=header_fc,
9cf643d1 PP	278	clock_classes=(cc1, cc2),
9cf643d1 PP	279	stream_classes=(sc1, sc2))
32656995	280	cc1, cc2, sc1, sc2, header_fc = self._test_eq_create_objects()
9cf643d1 PP	281	tc2 = bt2.Trace(name='my name',
	282	native_byte_order=bt2.ByteOrder.BIG_ENDIAN,
	283	env={'the_string': 'value', 'the_int': 23},
32656995	284	packet_header_field_class=header_fc,
9cf643d1 PP	285	clock_classes=(cc1, cc2),
	286	stream_classes=(sc1, sc2))
	287	self.assertNotEqual(tc1, tc2)
	288
	289	def test_ne_env(self):
32656995	290	cc1, cc2, sc1, sc2, header_fc = self._test_eq_create_objects()
9cf643d1 PP	291	tc1 = bt2.Trace(name='my name',
	292	native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN,
	293	env={'the_string': 'value', 'the_int2': 23},
32656995	294	packet_header_field_class=header_fc,
9cf643d1 PP	295	clock_classes=(cc1, cc2),
9cf643d1 PP	296	stream_classes=(sc1, sc2))
32656995	297	cc1, cc2, sc1, sc2, header_fc = self._test_eq_create_objects()
9cf643d1 PP	298	tc2 = bt2.Trace(name='my name',
	299	native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN,
	300	env={'the_string': 'value', 'the_int': 23},
32656995	301	packet_header_field_class=header_fc,
9cf643d1 PP	302	clock_classes=(cc1, cc2),
	303	stream_classes=(sc1, sc2))
	304	self.assertNotEqual(tc1, tc2)
	305
	306	def test_ne_clock_classes(self):
32656995	307	cc1, cc2, sc1, sc2, header_fc = self._test_eq_create_objects()
9cf643d1 PP	308	tc1 = bt2.Trace(name='my name',
	309	native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN,
	310	env={'the_string': 'value', 'the_int': 23},
32656995	311	packet_header_field_class=header_fc,
9cf643d1 PP	312	clock_classes=(cc1, cc2),
9cf643d1 PP	313	stream_classes=(sc1, sc2))
32656995	314	cc1, cc2, sc1, sc2, header_fc = self._test_eq_create_objects()
9cf643d1 PP	315	cc2.frequency = 1234
	316	tc2 = bt2.Trace(name='my name',
	317	native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN,
	318	env={'the_string': 'value', 'the_int': 23},
32656995	319	packet_header_field_class=header_fc,
9cf643d1 PP	320	clock_classes=(cc1, cc2),
	321	stream_classes=(sc1, sc2))
	322	self.assertNotEqual(tc1, tc2)
	323
	324	def test_ne_stream_classes(self):
32656995	325	cc1, cc2, sc1, sc2, header_fc = self._test_eq_create_objects()
9cf643d1 PP	326	tc1 = bt2.Trace(name='my name',
	327	native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN,
	328	env={'the_string': 'value', 'the_int': 23},
32656995	329	packet_header_field_class=header_fc,
9cf643d1 PP	330	clock_classes=(cc1, cc2),
9cf643d1 PP	331	stream_classes=(sc1, sc2))
32656995	332	cc1, cc2, sc1, sc2, header_fc = self._test_eq_create_objects()
9cf643d1 PP	333	sc2.id = 72632
	334	tc2 = bt2.Trace(name='my name',
	335	native_byte_order=bt2.ByteOrder.LITTLE_ENDIAN,
	336	env={'the_string': 'value', 'the_int': 23},
32656995	337	packet_header_field_class=header_fc,
9cf643d1 PP	338	clock_classes=(cc1, cc2),
	339	stream_classes=(sc1, sc2))
	340	self.assertNotEqual(tc1, tc2)
	341
	342	def test_eq_invalid(self):
	343	self.assertFalse(self._tc == 23)