tests/bindings/python/bt2/test_clock_class.py

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