2 # Copyright (C) 2019 EfficiOS Inc.
4 # This program is free software; you can redistribute it and/or
5 # modify it under the terms of the GNU General Public License
6 # as published by the Free Software Foundation; only version 2
9 # This program is distributed in the hope that it will be useful,
10 # but WITHOUT ANY WARRANTY; without even the implied warranty of
11 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
12 # GNU General Public License for more details.
14 # You should have received a copy of the GNU General Public License
15 # along with this program; if not, write to the Free Software
16 # Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
21 from utils
import run_in_component_init
22 from bt2
import trace
as bt2_trace
23 from bt2
import stream
as bt2_stream
24 from bt2
import value
as bt2_value
25 from bt2
import stream_class
as bt2_stream_class
28 class StreamTestCase(unittest
.TestCase
):
31 return comp_self
._create
_trace
_class
()
33 self
._tc
= run_in_component_init(f
)
34 self
._sc
= self
._tc
.create_stream_class(assigns_automatic_stream_id
=True)
37 def test_create_default(self
):
38 stream
= self
._tr
.create_stream(self
._sc
)
39 self
.assertIsNone(stream
.name
)
40 self
.assertIs(type(stream
), bt2_stream
._Stream
)
41 self
.assertEqual(len(stream
.user_attributes
), 0)
44 stream
= self
._tr
.create_stream(self
._sc
, name
='équidistant')
45 self
.assertEqual(stream
.name
, 'équidistant')
47 def test_invalid_name(self
):
48 with self
.assertRaises(TypeError):
49 self
._tr
.create_stream(self
._sc
, name
=22)
51 def test_create_user_attributes(self
):
52 stream
= self
._tr
.create_stream(self
._sc
, user_attributes
={'salut': 23})
53 self
.assertEqual(stream
.user_attributes
, {'salut': 23})
54 self
.assertIs(type(stream
.user_attributes
), bt2_value
.MapValue
)
56 def test_const_user_attributes(self
):
57 stream
= utils
.get_const_stream_beginning_message().stream
58 self
.assertEqual(stream
.user_attributes
, {'salut': 23})
59 self
.assertIs(type(stream
.user_attributes
), bt2_value
._MapValueConst
)
61 def test_create_invalid_user_attributes(self
):
62 with self
.assertRaises(TypeError):
63 self
._tr
.create_stream(self
._sc
, user_attributes
=object())
65 def test_create_invalid_user_attributes_value_type(self
):
66 with self
.assertRaises(TypeError):
67 self
._tr
.create_stream(self
._sc
, user_attributes
=23)
69 def test_stream_class(self
):
70 stream
= self
._tr
.create_stream(self
._sc
)
71 self
.assertEqual(stream
.cls
, self
._sc
)
72 self
.assertIs(type(stream
.cls
), bt2_stream_class
._StreamClass
)
74 def test_const_stream_class(self
):
75 stream
= utils
.get_const_stream_beginning_message().stream
76 self
.assertIs(type(stream
.cls
), bt2_stream_class
._StreamClassConst
)
79 stream
= self
._tr
.create_stream(self
._sc
)
80 self
.assertEqual(stream
.trace
.addr
, self
._tr
.addr
)
81 self
.assertIs(type(stream
.trace
), bt2_trace
._Trace
)
83 def test_const_trace(self
):
84 stream
= utils
.get_const_stream_beginning_message().stream
85 self
.assertIs(type(stream
.trace
), bt2_trace
._TraceConst
)
87 def test_invalid_id(self
):
88 sc
= self
._tc
.create_stream_class(assigns_automatic_stream_id
=False)
90 with self
.assertRaises(TypeError):
91 self
._tr
.create_stream(sc
, id='string')
94 if __name__
== '__main__':