1 from collections
import OrderedDict
8 class StreamTestCase(unittest
.TestCase
):
10 self
._stream
= self
._create
_stream
()
12 def _create_stream(self
, name
='my_stream'):
14 eh
= bt2
.StructureFieldType()
16 ('id', bt2
.IntegerFieldType(8)),
17 ('ts', bt2
.IntegerFieldType(32)),
20 # stream event context
21 sec
= bt2
.StructureFieldType()
23 ('cpu_id', bt2
.IntegerFieldType(8)),
24 ('stuff', bt2
.FloatingPointNumberFieldType()),
28 pc
= bt2
.StructureFieldType()
30 ('something', bt2
.IntegerFieldType(8)),
31 ('something_else', bt2
.FloatingPointNumberFieldType()),
35 sc
= bt2
.StreamClass()
36 sc
.packet_context_field_type
= pc
37 sc
.event_header_field_type
= eh
38 sc
.event_context_field_type
= sec
41 ec
= bt2
.StructureFieldType()
43 ('ant', bt2
.IntegerFieldType(16, is_signed
=True)),
44 ('msg', bt2
.StringFieldType()),
48 ep
= bt2
.StructureFieldType()
50 ('giraffe', bt2
.IntegerFieldType(32)),
51 ('gnu', bt2
.IntegerFieldType(8)),
52 ('mosquito', bt2
.IntegerFieldType(8)),
56 event_class
= bt2
.EventClass('ec')
57 event_class
.context_field_type
= ec
58 event_class
.payload_field_type
= ep
59 sc
.add_event_class(event_class
)
62 ph
= bt2
.StructureFieldType()
64 ('magic', bt2
.IntegerFieldType(32)),
65 ('stream_id', bt2
.IntegerFieldType(16)),
70 tc
.packet_header_field_type
= ph
71 tc
.add_stream_class(sc
)
76 def test_attr_stream_class(self
):
77 self
.assertIsNotNone(self
._stream
.stream_class
)
79 def test_attr_name(self
):
80 self
.assertEqual(self
._stream
.name
, 'my_stream')
83 stream1
= self
._create
_stream
()
84 stream2
= self
._create
_stream
()
85 self
.assertEqual(stream1
, stream2
)
87 def test_ne_name(self
):
88 stream1
= self
._create
_stream
()
89 stream2
= self
._create
_stream
('lel')
90 self
.assertNotEqual(stream1
, stream2
)
92 def test_eq_invalid(self
):
93 self
.assertFalse(self
._stream
== 23)
95 def _test_copy(self
, func
):
96 stream
= self
._create
_stream
()
98 self
.assertIsNot(stream
, cpy
)
99 self
.assertNotEqual(stream
.addr
, cpy
.addr
)
100 self
.assertEqual(stream
, cpy
)
103 self
._test
_copy
(copy
.copy
)
105 def test_deepcopy(self
):
106 self
._test
_copy
(copy
.deepcopy
)