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