1 # SPDX-License-Identifier: GPL-2.0-only
3 # Copyright (C) 2019 EfficiOS Inc.
8 from utils
import run_in_component_init
9 from bt2
import trace
as bt2_trace
10 from bt2
import stream
as bt2_stream
11 from bt2
import value
as bt2_value
12 from bt2
import stream_class
as bt2_stream_class
15 class StreamTestCase(unittest
.TestCase
):
18 return comp_self
._create
_trace
_class
()
20 self
._tc
= run_in_component_init(f
)
21 self
._sc
= self
._tc
.create_stream_class(assigns_automatic_stream_id
=True)
24 def test_create_default(self
):
25 stream
= self
._tr
.create_stream(self
._sc
)
26 self
.assertIsNone(stream
.name
)
27 self
.assertIs(type(stream
), bt2_stream
._Stream
)
28 self
.assertEqual(len(stream
.user_attributes
), 0)
31 stream
= self
._tr
.create_stream(self
._sc
, name
='équidistant')
32 self
.assertEqual(stream
.name
, 'équidistant')
34 def test_invalid_name(self
):
35 with self
.assertRaises(TypeError):
36 self
._tr
.create_stream(self
._sc
, name
=22)
38 def test_create_user_attributes(self
):
39 stream
= self
._tr
.create_stream(self
._sc
, user_attributes
={'salut': 23})
40 self
.assertEqual(stream
.user_attributes
, {'salut': 23})
41 self
.assertIs(type(stream
.user_attributes
), bt2_value
.MapValue
)
43 def test_const_user_attributes(self
):
44 stream
= utils
.get_const_stream_beginning_message().stream
45 self
.assertEqual(stream
.user_attributes
, {'salut': 23})
46 self
.assertIs(type(stream
.user_attributes
), bt2_value
._MapValueConst
)
48 def test_create_invalid_user_attributes(self
):
49 with self
.assertRaises(TypeError):
50 self
._tr
.create_stream(self
._sc
, user_attributes
=object())
52 def test_create_invalid_user_attributes_value_type(self
):
53 with self
.assertRaises(TypeError):
54 self
._tr
.create_stream(self
._sc
, user_attributes
=23)
56 def test_stream_class(self
):
57 stream
= self
._tr
.create_stream(self
._sc
)
58 self
.assertEqual(stream
.cls
, self
._sc
)
59 self
.assertIs(type(stream
.cls
), bt2_stream_class
._StreamClass
)
61 def test_const_stream_class(self
):
62 stream
= utils
.get_const_stream_beginning_message().stream
63 self
.assertIs(type(stream
.cls
), bt2_stream_class
._StreamClassConst
)
66 stream
= self
._tr
.create_stream(self
._sc
)
67 self
.assertEqual(stream
.trace
.addr
, self
._tr
.addr
)
68 self
.assertIs(type(stream
.trace
), bt2_trace
._Trace
)
70 def test_const_trace(self
):
71 stream
= utils
.get_const_stream_beginning_message().stream
72 self
.assertIs(type(stream
.trace
), bt2_trace
._TraceConst
)
74 def test_invalid_id(self
):
75 sc
= self
._tc
.create_stream_class(assigns_automatic_stream_id
=False)
77 with self
.assertRaises(TypeError):
78 self
._tr
.create_stream(sc
, id='string')
81 if __name__
== '__main__':