1 from collections
import OrderedDict
8 @unittest.skip("this is broken")
9 class StreamTestCase(unittest
.TestCase
):
11 self
._stream
= self
._create
_stream
(stream_id
=23)
16 def _create_stream(self
, name
='my_stream', stream_id
=None):
18 eh
= bt2
.StructureFieldClass()
20 ('id', bt2
.IntegerFieldClass(8)),
21 ('ts', bt2
.IntegerFieldClass(32)),
24 # stream event context
25 sec
= bt2
.StructureFieldClass()
27 ('cpu_id', bt2
.IntegerFieldClass(8)),
28 ('stuff', bt2
.FloatingPointNumberFieldClass()),
32 pc
= bt2
.StructureFieldClass()
34 ('something', bt2
.IntegerFieldClass(8)),
35 ('something_else', bt2
.FloatingPointNumberFieldClass()),
39 sc
= bt2
.StreamClass()
40 sc
.packet_context_field_class
= pc
41 sc
.event_header_field_class
= eh
42 sc
.event_context_field_class
= sec
45 ec
= bt2
.StructureFieldClass()
47 ('ant', bt2
.IntegerFieldClass(16, is_signed
=True)),
48 ('msg', bt2
.StringFieldClass()),
52 ep
= bt2
.StructureFieldClass()
54 ('giraffe', bt2
.IntegerFieldClass(32)),
55 ('gnu', bt2
.IntegerFieldClass(8)),
56 ('mosquito', bt2
.IntegerFieldClass(8)),
60 event_class
= bt2
.EventClass('ec')
61 event_class
.context_field_class
= ec
62 event_class
.payload_field_class
= ep
63 sc
.add_event_class(event_class
)
66 ph
= bt2
.StructureFieldClass()
68 ('magic', bt2
.IntegerFieldClass(32)),
69 ('stream_id', bt2
.IntegerFieldClass(16)),
74 tc
.packet_header_field_class
= ph
75 tc
.add_stream_class(sc
)
78 return sc(name
=name
, id=stream_id
)
80 def test_attr_stream_class(self
):
81 self
.assertIsNotNone(self
._stream
.stream_class
)
83 def test_attr_name(self
):
84 self
.assertEqual(self
._stream
.name
, 'my_stream')
87 stream1
= self
._create
_stream
(stream_id
=17)
88 stream2
= self
._create
_stream
(stream_id
=17)
89 self
.assertEqual(stream1
, stream2
)
91 def test_ne_name(self
):
92 stream1
= self
._create
_stream
(stream_id
=17)
93 stream2
= self
._create
_stream
('lel', 17)
94 self
.assertNotEqual(stream1
, stream2
)
97 stream1
= self
._create
_stream
(stream_id
=17)
98 stream2
= self
._create
_stream
(stream_id
=23)
99 self
.assertNotEqual(stream1
, stream2
)
101 def test_eq_invalid(self
):
102 self
.assertFalse(self
._stream
== 23)
104 def _test_copy(self
, func
):
105 stream
= self
._create
_stream
()
107 self
.assertIsNot(stream
, cpy
)
108 self
.assertNotEqual(stream
.addr
, cpy
.addr
)
109 self
.assertEqual(stream
, cpy
)
112 self
._test
_copy
(copy
.copy
)
114 def test_deepcopy(self
):
115 self
._test
_copy
(copy
.deepcopy
)