1 from collections
import OrderedDict
8 class StreamTestCase(unittest
.TestCase
):
10 self
._stream
= self
._create
_stream
(stream_id
=23)
15 def _create_stream(self
, name
='my_stream', stream_id
=None):
17 eh
= bt2
.StructureFieldType()
19 ('id', bt2
.IntegerFieldType(8)),
20 ('ts', bt2
.IntegerFieldType(32)),
23 # stream event context
24 sec
= bt2
.StructureFieldType()
26 ('cpu_id', bt2
.IntegerFieldType(8)),
27 ('stuff', bt2
.FloatingPointNumberFieldType()),
31 pc
= bt2
.StructureFieldType()
33 ('something', bt2
.IntegerFieldType(8)),
34 ('something_else', bt2
.FloatingPointNumberFieldType()),
38 sc
= bt2
.StreamClass()
39 sc
.packet_context_field_type
= pc
40 sc
.event_header_field_type
= eh
41 sc
.event_context_field_type
= sec
44 ec
= bt2
.StructureFieldType()
46 ('ant', bt2
.IntegerFieldType(16, is_signed
=True)),
47 ('msg', bt2
.StringFieldType()),
51 ep
= bt2
.StructureFieldType()
53 ('giraffe', bt2
.IntegerFieldType(32)),
54 ('gnu', bt2
.IntegerFieldType(8)),
55 ('mosquito', bt2
.IntegerFieldType(8)),
59 event_class
= bt2
.EventClass('ec')
60 event_class
.context_field_type
= ec
61 event_class
.payload_field_type
= ep
62 sc
.add_event_class(event_class
)
65 ph
= bt2
.StructureFieldType()
67 ('magic', bt2
.IntegerFieldType(32)),
68 ('stream_id', bt2
.IntegerFieldType(16)),
73 tc
.packet_header_field_type
= ph
74 tc
.add_stream_class(sc
)
77 return sc(name
=name
, id=stream_id
)
79 def test_attr_stream_class(self
):
80 self
.assertIsNotNone(self
._stream
.stream_class
)
82 def test_attr_name(self
):
83 self
.assertEqual(self
._stream
.name
, 'my_stream')
86 stream1
= self
._create
_stream
(stream_id
=17)
87 stream2
= self
._create
_stream
(stream_id
=17)
88 self
.assertEqual(stream1
, stream2
)
90 def test_ne_name(self
):
91 stream1
= self
._create
_stream
(stream_id
=17)
92 stream2
= self
._create
_stream
('lel', 17)
93 self
.assertNotEqual(stream1
, stream2
)
96 stream1
= self
._create
_stream
(stream_id
=17)
97 stream2
= self
._create
_stream
(stream_id
=23)
98 self
.assertNotEqual(stream1
, stream2
)
100 def test_eq_invalid(self
):
101 self
.assertFalse(self
._stream
== 23)
103 def _test_copy(self
, func
):
104 stream
= self
._create
_stream
()
106 self
.assertIsNot(stream
, cpy
)
107 self
.assertNotEqual(stream
.addr
, cpy
.addr
)
108 self
.assertEqual(stream
, cpy
)
111 self
._test
_copy
(copy
.copy
)
113 def test_deepcopy(self
):
114 self
._test
_copy
(copy
.deepcopy
)