tests/bindings/python/bt2/test_stream.py

   1 # SPDX-License-Identifier: GPL-2.0-only
   2 #
   3 # Copyright (C) 2019 EfficiOS Inc.
   4 #
   5
   6 import unittest
   7 import utils
   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
  13
  14
  15 class StreamTestCase(unittest.TestCase):
  16     def setUp(self):
  17         def f(comp_self):
  18             return comp_self._create_trace_class()
  19
  20         self._tc = run_in_component_init(f)
  21         self._sc = self._tc.create_stream_class(assigns_automatic_stream_id=True)
  22         self._tr = self._tc()
  23
  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)
  29
  30     def test_name(self):
  31         stream = self._tr.create_stream(self._sc, name='équidistant')
  32         self.assertEqual(stream.name, 'équidistant')
  33
  34     def test_invalid_name(self):
  35         with self.assertRaises(TypeError):
  36             self._tr.create_stream(self._sc, name=22)
  37
  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)
  42
  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)
  47
  48     def test_create_invalid_user_attributes(self):
  49         with self.assertRaises(TypeError):
  50             self._tr.create_stream(self._sc, user_attributes=object())
  51
  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)
  55
  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)
  60
  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)
  64
  65     def test_trace(self):
  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)
  69
  70     def test_const_trace(self):
  71         stream = utils.get_const_stream_beginning_message().stream
  72         self.assertIs(type(stream.trace), bt2_trace._TraceConst)
  73
  74     def test_invalid_id(self):
  75         sc = self._tc.create_stream_class(assigns_automatic_stream_id=False)
  76
  77         with self.assertRaises(TypeError):
  78             self._tr.create_stream(sc, id='string')
  79
  80
  81 if __name__ == '__main__':
  82     unittest.main()