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