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
   8 import utils
   9 from bt2 import trace as bt2_trace
  10 from bt2 import value as bt2_value
  11 from bt2 import stream as bt2_stream
  12 from bt2 import stream_class as bt2_stream_class
  13 from utils import run_in_component_init
  14
  15
  16 class StreamTestCase(unittest.TestCase):
  17     def setUp(self):
  18         def f(comp_self):
  19             return comp_self._create_trace_class()
  20
  21         self._tc = run_in_component_init(f)
  22         self._sc = self._tc.create_stream_class(assigns_automatic_stream_id=True)
  23         self._tr = self._tc()
  24
  25     def test_create_default(self):
  26         stream = self._tr.create_stream(self._sc)
  27         self.assertIsNone(stream.name)
  28         self.assertIs(type(stream), bt2_stream._Stream)
  29         self.assertEqual(len(stream.user_attributes), 0)
  30
  31     def test_name(self):
  32         stream = self._tr.create_stream(self._sc, name="équidistant")
  33         self.assertEqual(stream.name, "équidistant")
  34
  35     def test_invalid_name(self):
  36         with self.assertRaises(TypeError):
  37             self._tr.create_stream(self._sc, name=22)
  38
  39     def test_create_user_attributes(self):
  40         stream = self._tr.create_stream(self._sc, user_attributes={"salut": 23})
  41         self.assertEqual(stream.user_attributes, {"salut": 23})
  42         self.assertIs(type(stream.user_attributes), bt2_value.MapValue)
  43
  44     def test_const_user_attributes(self):
  45         stream = utils.get_const_stream_beginning_message().stream
  46         self.assertEqual(stream.user_attributes, {"salut": 23})
  47         self.assertIs(type(stream.user_attributes), bt2_value._MapValueConst)
  48
  49     def test_create_invalid_user_attributes(self):
  50         with self.assertRaises(TypeError):
  51             self._tr.create_stream(self._sc, user_attributes=object())
  52
  53     def test_create_invalid_user_attributes_value_type(self):
  54         with self.assertRaises(TypeError):
  55             self._tr.create_stream(self._sc, user_attributes=23)
  56
  57     def test_stream_class(self):
  58         stream = self._tr.create_stream(self._sc)
  59         self.assertEqual(stream.cls, self._sc)
  60         self.assertIs(type(stream.cls), bt2_stream_class._StreamClass)
  61
  62     def test_const_stream_class(self):
  63         stream = utils.get_const_stream_beginning_message().stream
  64         self.assertIs(type(stream.cls), bt2_stream_class._StreamClassConst)
  65
  66     def test_trace(self):
  67         stream = self._tr.create_stream(self._sc)
  68         self.assertEqual(stream.trace.addr, self._tr.addr)
  69         self.assertIs(type(stream.trace), bt2_trace._Trace)
  70
  71     def test_const_trace(self):
  72         stream = utils.get_const_stream_beginning_message().stream
  73         self.assertIs(type(stream.trace), bt2_trace._TraceConst)
  74
  75     def test_invalid_id(self):
  76         sc = self._tc.create_stream_class(assigns_automatic_stream_id=False)
  77
  78         with self.assertRaises(TypeError):
  79             self._tr.create_stream(sc, id="string")
  80
  81
  82 if __name__ == "__main__":
  83     unittest.main()