tests/bindings/python/bt2/test_clock_class.py

   1 # SPDX-License-Identifier: GPL-2.0-only
   2 #
   3 # Copyright (C) 2019 EfficiOS Inc.
   4 #
   5
   6 import uuid
   7 import unittest
   8
   9 import bt2
  10 import utils
  11 from bt2 import value as bt2_value
  12 from bt2 import clock_class as bt2_clock_class
  13 from utils import TestOutputPortMessageIterator, run_in_component_init
  14
  15
  16 class ClockClassOffsetTestCase(unittest.TestCase):
  17     def test_create_default(self):
  18         cco = bt2.ClockClassOffset()
  19         self.assertEqual(cco.seconds, 0)
  20         self.assertEqual(cco.cycles, 0)
  21
  22     def test_create(self):
  23         cco = bt2.ClockClassOffset(23, 4871232)
  24         self.assertEqual(cco.seconds, 23)
  25         self.assertEqual(cco.cycles, 4871232)
  26
  27     def test_create_kwargs(self):
  28         cco = bt2.ClockClassOffset(seconds=23, cycles=4871232)
  29         self.assertEqual(cco.seconds, 23)
  30         self.assertEqual(cco.cycles, 4871232)
  31
  32     def test_create_invalid_seconds(self):
  33         with self.assertRaises(TypeError):
  34             bt2.ClockClassOffset("hello", 4871232)
  35
  36     def test_create_invalid_cycles(self):
  37         with self.assertRaises(TypeError):
  38             bt2.ClockClassOffset(23, "hello")
  39
  40     def test_eq(self):
  41         cco1 = bt2.ClockClassOffset(23, 42)
  42         cco2 = bt2.ClockClassOffset(23, 42)
  43         self.assertEqual(cco1, cco2)
  44
  45     def test_ne_seconds(self):
  46         cco1 = bt2.ClockClassOffset(23, 42)
  47         cco2 = bt2.ClockClassOffset(24, 42)
  48         self.assertNotEqual(cco1, cco2)
  49
  50     def test_ne_cycles(self):
  51         cco1 = bt2.ClockClassOffset(23, 42)
  52         cco2 = bt2.ClockClassOffset(23, 43)
  53         self.assertNotEqual(cco1, cco2)
  54
  55     def test_eq_invalid(self):
  56         self.assertFalse(bt2.ClockClassOffset() == 23)
  57
  58
  59 class ClockClassTestCase(unittest.TestCase):
  60     def assertRaisesInComponentInit(self, expected_exc_type, user_code):
  61         def f(comp_self):
  62             try:
  63                 user_code(comp_self)
  64             except Exception as exc:
  65                 return type(exc)
  66
  67         exc_type = run_in_component_init(f)
  68         self.assertIsNotNone(exc_type)
  69         self.assertEqual(exc_type, expected_exc_type)
  70
  71     def test_create_default(self):
  72         cc = run_in_component_init(lambda comp_self: comp_self._create_clock_class())
  73
  74         self.assertIsNone(cc.name)
  75         self.assertEqual(cc.frequency, 1000000000)
  76         self.assertIsNone(cc.description)
  77         self.assertEqual(cc.precision, 0)
  78         self.assertEqual(cc.offset, bt2.ClockClassOffset())
  79         self.assertTrue(cc.origin_is_unix_epoch)
  80         self.assertIsNone(cc.uuid)
  81         self.assertEqual(len(cc.user_attributes), 0)
  82
  83     def test_create_name(self):
  84         def f(comp_self):
  85             return comp_self._create_clock_class(name="the_clock")
  86
  87         cc = run_in_component_init(f)
  88         self.assertEqual(cc.name, "the_clock")
  89
  90     def test_create_invalid_name(self):
  91         def f(comp_self):
  92             comp_self._create_clock_class(name=23)
  93
  94         self.assertRaisesInComponentInit(TypeError, f)
  95
  96     def test_create_description(self):
  97         def f(comp_self):
  98             return comp_self._create_clock_class(description="hi people")
  99
 100         cc = run_in_component_init(f)
 101         self.assertEqual(cc.description, "hi people")
 102
 103     def test_create_invalid_description(self):
 104         def f(comp_self):
 105             return comp_self._create_clock_class(description=23)
 106
 107         self.assertRaisesInComponentInit(TypeError, f)
 108
 109     def test_create_frequency(self):
 110         def f(comp_self):
 111             return comp_self._create_clock_class(frequency=987654321)
 112
 113         cc = run_in_component_init(f)
 114         self.assertEqual(cc.frequency, 987654321)
 115
 116     def test_create_invalid_frequency(self):
 117         def f(comp_self):
 118             return comp_self._create_clock_class(frequency="lel")
 119
 120         self.assertRaisesInComponentInit(TypeError, f)
 121
 122     def test_create_precision(self):
 123         def f(comp_self):
 124             return comp_self._create_clock_class(precision=12)
 125
 126         cc = run_in_component_init(f)
 127         self.assertEqual(cc.precision, 12)
 128
 129     def test_create_invalid_precision(self):
 130         def f(comp_self):
 131             return comp_self._create_clock_class(precision="lel")
 132
 133         self.assertRaisesInComponentInit(TypeError, f)
 134
 135     def test_create_offset(self):
 136         def f(comp_self):
 137             return comp_self._create_clock_class(offset=bt2.ClockClassOffset(12, 56))
 138
 139         cc = run_in_component_init(f)
 140         self.assertEqual(cc.offset, bt2.ClockClassOffset(12, 56))
 141
 142     def test_create_invalid_offset(self):
 143         def f(comp_self):
 144             return comp_self._create_clock_class(offset=object())
 145
 146         self.assertRaisesInComponentInit(TypeError, f)
 147
 148     def test_create_origin_is_unix_epoch(self):
 149         def f(comp_self):
 150             return comp_self._create_clock_class(origin_is_unix_epoch=False)
 151
 152         cc = run_in_component_init(f)
 153         self.assertEqual(cc.origin_is_unix_epoch, False)
 154
 155     def test_create_invalid_origin_is_unix_epoch(self):
 156         def f(comp_self):
 157             return comp_self._create_clock_class(origin_is_unix_epoch=23)
 158
 159         self.assertRaisesInComponentInit(TypeError, f)
 160
 161     def test_cycles_to_ns_from_origin(self):
 162         def f(comp_self):
 163             return comp_self._create_clock_class(
 164                 frequency=10**8, origin_is_unix_epoch=True
 165             )
 166
 167         cc = run_in_component_init(f)
 168         self.assertEqual(cc.cycles_to_ns_from_origin(112), 1120)
 169
 170     def test_cycles_to_ns_from_origin_overflow(self):
 171         def f(comp_self):
 172             return comp_self._create_clock_class(frequency=1000)
 173
 174         cc = run_in_component_init(f)
 175         with self.assertRaises(bt2._OverflowError):
 176             cc.cycles_to_ns_from_origin(2**63)
 177
 178     def test_create_uuid(self):
 179         def f(comp_self):
 180             return comp_self._create_clock_class(
 181                 uuid=uuid.UUID("b43372c32ef0be28444dfc1c5cdafd33")
 182             )
 183
 184         cc = run_in_component_init(f)
 185         self.assertEqual(cc.uuid, uuid.UUID("b43372c32ef0be28444dfc1c5cdafd33"))
 186
 187     def test_create_invalid_uuid(self):
 188         def f(comp_self):
 189             return comp_self._create_clock_class(uuid=23)
 190
 191         self.assertRaisesInComponentInit(TypeError, f)
 192
 193     def test_create_user_attributes(self):
 194         def f(comp_self):
 195             return comp_self._create_clock_class(user_attributes={"salut": 23})
 196
 197         cc = run_in_component_init(f)
 198         self.assertEqual(cc.user_attributes, {"salut": 23})
 199         self.assertIs(type(cc.user_attributes), bt2_value.MapValue)
 200
 201     def test_create_invalid_user_attributes(self):
 202         def f(comp_self):
 203             return comp_self._create_clock_class(user_attributes=object())
 204
 205         self.assertRaisesInComponentInit(TypeError, f)
 206
 207     def test_create_invalid_user_attributes_value_type(self):
 208         def f(comp_self):
 209             return comp_self._create_clock_class(user_attributes=23)
 210
 211         self.assertRaisesInComponentInit(TypeError, f)
 212
 213     def test_const_user_attributes(self):
 214         cc = utils.get_const_event_message().default_clock_snapshot.clock_class
 215         self.assertIs(type(cc.user_attributes), bt2_value._MapValueConst)
 216
 217
 218 class ClockSnapshotTestCase(unittest.TestCase):
 219     def setUp(self):
 220         def f(comp_self):
 221             cc = comp_self._create_clock_class(
 222                 1000, "my_cc", offset=bt2.ClockClassOffset(45, 354)
 223             )
 224             tc = comp_self._create_trace_class()
 225
 226             return (cc, tc)
 227
 228         _cc, _tc = run_in_component_init(f)
 229         _trace = _tc()
 230         _sc = _tc.create_stream_class(default_clock_class=_cc)
 231         _ec = _sc.create_event_class(name="salut")
 232         _stream = _trace.create_stream(_sc)
 233         self._stream = _stream
 234         self._ec = _ec
 235         self._cc = _cc
 236
 237         class MyIter(bt2._UserMessageIterator):
 238             def __init__(self, config, self_port_output):
 239                 self._at = 0
 240
 241             def __next__(self):
 242                 if self._at == 0:
 243                     notif = self._create_stream_beginning_message(_stream)
 244                 elif self._at == 1:
 245                     notif = self._create_event_message(_ec, _stream, 123)
 246                 elif self._at == 2:
 247                     notif = self._create_event_message(_ec, _stream, 2**63)
 248                 elif self._at == 3:
 249                     notif = self._create_stream_end_message(_stream)
 250                 else:
 251                     raise bt2.Stop
 252
 253                 self._at += 1
 254                 return notif
 255
 256         class MySrc(bt2._UserSourceComponent, message_iterator_class=MyIter):
 257             def __init__(self, config, params, obj):
 258                 self._add_output_port("out")
 259
 260         self._graph = bt2.Graph()
 261         self._src_comp = self._graph.add_component(MySrc, "my_source")
 262         self._msg_iter = TestOutputPortMessageIterator(
 263             self._graph, self._src_comp.output_ports["out"]
 264         )
 265
 266         for i, msg in enumerate(self._msg_iter):
 267             if i == 1:
 268                 self._msg = msg
 269             elif i == 2:
 270                 self._msg_clock_overflow = msg
 271                 break
 272
 273     def tearDown(self):
 274         del self._cc
 275         del self._msg
 276
 277     def test_create_default(self):
 278         self.assertEqual(
 279             self._msg.default_clock_snapshot.clock_class.addr, self._cc.addr
 280         )
 281         self.assertEqual(self._msg.default_clock_snapshot.value, 123)
 282
 283     def test_clock_class(self):
 284         cc = self._msg.default_clock_snapshot.clock_class
 285         self.assertEqual(cc.addr, self._cc.addr)
 286         self.assertIs(type(cc), bt2_clock_class._ClockClassConst)
 287
 288     def test_ns_from_origin(self):
 289         s_from_origin = 45 + ((354 + 123) / 1000)
 290         ns_from_origin = int(s_from_origin * 1e9)
 291         self.assertEqual(
 292             self._msg.default_clock_snapshot.ns_from_origin, ns_from_origin
 293         )
 294
 295     def test_ns_from_origin_overflow(self):
 296         with self.assertRaises(bt2._OverflowError):
 297             self._msg_clock_overflow.default_clock_snapshot.ns_from_origin
 298
 299     def test_eq_int(self):
 300         self.assertEqual(self._msg.default_clock_snapshot, 123)
 301
 302     def test_eq_invalid(self):
 303         self.assertFalse(self._msg.default_clock_snapshot == 23)
 304
 305     def test_comparison(self):
 306         self.assertTrue(self._msg.default_clock_snapshot > 100)
 307         self.assertFalse(self._msg.default_clock_snapshot > 200)
 308
 309         self.assertTrue(self._msg.default_clock_snapshot >= 123)
 310         self.assertFalse(self._msg.default_clock_snapshot >= 200)
 311
 312         self.assertTrue(self._msg.default_clock_snapshot < 200)
 313         self.assertFalse(self._msg.default_clock_snapshot < 100)
 314
 315         self.assertTrue(self._msg.default_clock_snapshot <= 123)
 316         self.assertFalse(self._msg.default_clock_snapshot <= 100)
 317
 318
 319 if __name__ == "__main__":
 320     unittest.main()