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