tests/bindings/python/bt2/test_stream.py

   1 #
   2 # Copyright (C) 2019 EfficiOS Inc.
   3 #
   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
   7 # of the License.
   8 #
   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.
  13 #
  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.
  17 #
  18
  19 import unittest
  20 import utils
  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
  26
  27
  28 class StreamTestCase(unittest.TestCase):
  29     def setUp(self):
  30         def f(comp_self):
  31             return comp_self._create_trace_class()
  32
  33         self._tc = run_in_component_init(f)
  34         self._sc = self._tc.create_stream_class(assigns_automatic_stream_id=True)
  35         self._tr = self._tc()
  36
  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)
  42
  43     def test_name(self):
  44         stream = self._tr.create_stream(self._sc, name='équidistant')
  45         self.assertEqual(stream.name, 'équidistant')
  46
  47     def test_invalid_name(self):
  48         with self.assertRaises(TypeError):
  49             self._tr.create_stream(self._sc, name=22)
  50
  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)
  55
  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)
  60
  61     def test_create_invalid_user_attributes(self):
  62         with self.assertRaises(TypeError):
  63             self._tr.create_stream(self._sc, user_attributes=object())
  64
  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)
  68
  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)
  73
  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)
  77
  78     def test_trace(self):
  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)
  82
  83     def test_const_trace(self):
  84         stream = utils.get_const_stream_beginning_message().stream
  85         self.assertIs(type(stream.trace), bt2_trace._TraceConst)
  86
  87     def test_invalid_id(self):
  88         sc = self._tc.create_stream_class(assigns_automatic_stream_id=False)
  89
  90         with self.assertRaises(TypeError):
  91             self._tr.create_stream(sc, id='string')
  92
  93
  94 if __name__ == '__main__':
  95     unittest.main()