tests/bindings/python/bt2/test_stream_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 from utils import run_in_component_init
  21 from bt2 import stream_class as bt2_stream_class
  22 from bt2 import trace_class as bt2_trace_class
  23 from bt2 import clock_class as bt2_clock_class
  24 from bt2 import event_class as bt2_event_class
  25 from bt2 import field_class as bt2_field_class
  26
  27
  28 class StreamClassTestCase(unittest.TestCase):
  29     def setUp(self):
  30         def f(comp_self):
  31             tc = comp_self._create_trace_class(assigns_automatic_stream_class_id=True)
  32             cc = comp_self._create_clock_class()
  33             return tc, cc
  34
  35         self._tc, self._cc = run_in_component_init(f)
  36         self._trace = self._tc()
  37
  38     def test_create_default(self):
  39         sc = self._tc.create_stream_class()
  40
  41         self.assertIs(type(sc), bt2_stream_class._StreamClass)
  42         self.assertIsNone(sc.name)
  43         self.assertIsNone(sc.packet_context_field_class)
  44         self.assertIsNone(sc.event_common_context_field_class)
  45         self.assertIsNone(sc.default_clock_class)
  46         self.assertTrue(sc.assigns_automatic_event_class_id)
  47         self.assertTrue(sc.assigns_automatic_stream_id)
  48         self.assertFalse(sc.supports_packets)
  49         self.assertFalse(sc.packets_have_beginning_default_clock_snapshot)
  50         self.assertFalse(sc.packets_have_end_default_clock_snapshot)
  51         self.assertFalse(sc.supports_discarded_events)
  52         self.assertFalse(sc.discarded_events_have_default_clock_snapshots)
  53         self.assertFalse(sc.supports_discarded_packets)
  54         self.assertFalse(sc.discarded_packets_have_default_clock_snapshots)
  55         self.assertEqual(len(sc.user_attributes), 0)
  56
  57     def test_create_name(self):
  58         sc = self._tc.create_stream_class(name='bozo')
  59         self.assertEqual(sc.name, 'bozo')
  60
  61     def test_create_invalid_name(self):
  62         with self.assertRaisesRegex(TypeError, "'int' is not a 'str' object"):
  63             self._tc.create_stream_class(name=17)
  64
  65         self.assertEqual(len(self._tc), 0)
  66
  67     def test_create_packet_context_field_class(self):
  68         fc = self._tc.create_structure_field_class()
  69         sc = self._tc.create_stream_class(
  70             packet_context_field_class=fc, supports_packets=True
  71         )
  72         self.assertEqual(sc.packet_context_field_class, fc)
  73         self.assertIs(
  74             type(sc.packet_context_field_class), bt2_field_class._StructureFieldClass
  75         )
  76
  77     def test_create_invalid_packet_context_field_class(self):
  78         with self.assertRaisesRegex(
  79             TypeError,
  80             "'int' is not a '<class 'bt2.field_class._StructureFieldClass'>' object",
  81         ):
  82             self._tc.create_stream_class(
  83                 packet_context_field_class=22, supports_packets=True
  84             )
  85
  86         self.assertEqual(len(self._tc), 0)
  87
  88     def test_create_invalid_packet_context_field_class_no_packets(self):
  89         fc = self._tc.create_structure_field_class()
  90
  91         with self.assertRaisesRegex(
  92             ValueError,
  93             "cannot have a packet context field class without supporting packets",
  94         ):
  95             self._tc.create_stream_class(packet_context_field_class=fc)
  96
  97         self.assertEqual(len(self._tc), 0)
  98
  99     def test_create_event_common_context_field_class(self):
 100         fc = self._tc.create_structure_field_class()
 101         sc = self._tc.create_stream_class(event_common_context_field_class=fc)
 102         self.assertEqual(sc.event_common_context_field_class, fc)
 103         self.assertIs(
 104             type(sc.event_common_context_field_class),
 105             bt2_field_class._StructureFieldClass,
 106         )
 107
 108     def test_create_invalid_event_common_context_field_class(self):
 109         with self.assertRaisesRegex(
 110             TypeError,
 111             "'int' is not a '<class 'bt2.field_class._StructureFieldClass'>' object",
 112         ):
 113             self._tc.create_stream_class(event_common_context_field_class=22)
 114
 115         self.assertEqual(len(self._tc), 0)
 116
 117     def test_create_default_clock_class(self):
 118         sc = self._tc.create_stream_class(default_clock_class=self._cc)
 119         self.assertEqual(sc.default_clock_class.addr, self._cc.addr)
 120         self.assertIs(type(sc.default_clock_class), bt2_clock_class._ClockClass)
 121
 122     def test_create_invalid_default_clock_class(self):
 123         with self.assertRaisesRegex(
 124             TypeError, "'int' is not a '<class 'bt2.clock_class._ClockClass'>' object"
 125         ):
 126             self._tc.create_stream_class(default_clock_class=12)
 127
 128         self.assertEqual(len(self._tc), 0)
 129
 130     def test_create_user_attributes(self):
 131         sc = self._tc.create_stream_class(user_attributes={'salut': 23})
 132         self.assertEqual(sc.user_attributes, {'salut': 23})
 133
 134     def test_create_invalid_user_attributes(self):
 135         with self.assertRaisesRegex(
 136             TypeError, "cannot create value object from 'object' object"
 137         ):
 138             self._tc.create_stream_class(user_attributes=object())
 139
 140         self.assertEqual(len(self._tc), 0)
 141
 142     def test_create_invalid_user_attributes_value_type(self):
 143         with self.assertRaisesRegex(
 144             TypeError,
 145             "'SignedIntegerValue' is not a '<class 'bt2.value.MapValue'>' object",
 146         ):
 147             self._tc.create_stream_class(user_attributes=23)
 148
 149         self.assertEqual(len(self._tc), 0)
 150
 151     def test_automatic_stream_ids(self):
 152         sc = self._tc.create_stream_class(assigns_automatic_stream_id=True)
 153         self.assertTrue(sc.assigns_automatic_stream_id)
 154
 155         stream = self._trace.create_stream(sc)
 156         self.assertIsNotNone(stream.id)
 157
 158     def test_automatic_stream_ids_raises(self):
 159         sc = self._tc.create_stream_class(assigns_automatic_stream_id=True)
 160         self.assertTrue(sc.assigns_automatic_stream_id)
 161
 162         with self.assertRaisesRegex(
 163             ValueError, "id provided, but stream class assigns automatic stream ids"
 164         ):
 165             self._trace.create_stream(sc, id=123)
 166
 167         self.assertEqual(len(self._trace), 0)
 168
 169     def test_automatic_stream_ids_wrong_type(self):
 170         with self.assertRaisesRegex(TypeError, "str' is not a 'bool' object"):
 171             self._tc.create_stream_class(assigns_automatic_stream_id='True')
 172
 173         self.assertEqual(len(self._tc), 0)
 174
 175     def test_no_automatic_stream_ids(self):
 176         sc = self._tc.create_stream_class(assigns_automatic_stream_id=False)
 177         self.assertFalse(sc.assigns_automatic_stream_id)
 178
 179         stream = self._trace.create_stream(sc, id=333)
 180         self.assertEqual(stream.id, 333)
 181
 182     def test_no_automatic_stream_ids_raises(self):
 183         sc = self._tc.create_stream_class(assigns_automatic_stream_id=False)
 184         self.assertFalse(sc.assigns_automatic_stream_id)
 185
 186         with self.assertRaisesRegex(
 187             ValueError,
 188             "id not provided, but stream class does not assign automatic stream ids",
 189         ):
 190             self._trace.create_stream(sc)
 191
 192         self.assertEqual(len(self._trace), 0)
 193
 194     def test_automatic_event_class_ids(self):
 195         sc = self._tc.create_stream_class(assigns_automatic_event_class_id=True)
 196         self.assertTrue(sc.assigns_automatic_event_class_id)
 197
 198         ec = sc.create_event_class()
 199         self.assertIsNotNone(ec.id)
 200
 201     def test_automatic_event_class_ids_raises(self):
 202         sc = self._tc.create_stream_class(assigns_automatic_event_class_id=True)
 203         self.assertTrue(sc.assigns_automatic_event_class_id)
 204
 205         with self.assertRaisesRegex(
 206             ValueError,
 207             "id provided, but stream class assigns automatic event class ids",
 208         ):
 209             sc.create_event_class(id=123)
 210
 211         self.assertEqual(len(sc), 0)
 212
 213     def test_automatic_event_class_ids_wrong_type(self):
 214         with self.assertRaisesRegex(TypeError, "'str' is not a 'bool' object"):
 215             self._tc.create_stream_class(assigns_automatic_event_class_id='True')
 216
 217         self.assertEqual(len(self._tc), 0)
 218
 219     def test_no_automatic_event_class_ids(self):
 220         sc = self._tc.create_stream_class(assigns_automatic_event_class_id=False)
 221         self.assertFalse(sc.assigns_automatic_event_class_id)
 222
 223         ec = sc.create_event_class(id=333)
 224         self.assertEqual(ec.id, 333)
 225
 226     def test_no_automatic_event_class_ids_raises(self):
 227         sc = self._tc.create_stream_class(assigns_automatic_event_class_id=False)
 228         self.assertFalse(sc.assigns_automatic_event_class_id)
 229
 230         with self.assertRaisesRegex(
 231             ValueError,
 232             "id not provided, but stream class does not assign automatic event class ids",
 233         ):
 234             sc.create_event_class()
 235
 236         self.assertEqual(len(sc), 0)
 237
 238     def test_supports_packets_without_cs(self):
 239         sc = self._tc.create_stream_class(
 240             default_clock_class=self._cc, supports_packets=True
 241         )
 242         self.assertTrue(sc.supports_packets)
 243         self.assertFalse(sc.packets_have_beginning_default_clock_snapshot)
 244         self.assertFalse(sc.packets_have_end_default_clock_snapshot)
 245
 246     def test_supports_packets_with_begin_cs(self):
 247         sc = self._tc.create_stream_class(
 248             default_clock_class=self._cc,
 249             supports_packets=True,
 250             packets_have_beginning_default_clock_snapshot=True,
 251         )
 252         self.assertTrue(sc.supports_packets)
 253         self.assertTrue(sc.packets_have_beginning_default_clock_snapshot)
 254         self.assertFalse(sc.packets_have_end_default_clock_snapshot)
 255
 256     def test_supports_packets_with_end_cs(self):
 257         sc = self._tc.create_stream_class(
 258             default_clock_class=self._cc,
 259             supports_packets=True,
 260             packets_have_end_default_clock_snapshot=True,
 261         )
 262         self.assertTrue(sc.supports_packets)
 263         self.assertFalse(sc.packets_have_beginning_default_clock_snapshot)
 264         self.assertTrue(sc.packets_have_end_default_clock_snapshot)
 265
 266     def test_supports_packets_raises_type_error(self):
 267         with self.assertRaisesRegex(TypeError, "'int' is not a 'bool' object"):
 268             self._tc.create_stream_class(
 269                 default_clock_class=self._cc, supports_packets=23
 270             )
 271
 272         self.assertEqual(len(self._tc), 0)
 273
 274     def test_packets_have_begin_default_cs_raises_type_error(self):
 275         with self.assertRaisesRegex(TypeError, "'int' is not a 'bool' object"):
 276             self._tc.create_stream_class(
 277                 default_clock_class=self._cc,
 278                 packets_have_beginning_default_clock_snapshot=23,
 279             )
 280
 281         self.assertEqual(len(self._tc), 0)
 282
 283     def test_packets_have_end_default_cs_raises_type_error(self):
 284         with self.assertRaisesRegex(TypeError, "'int' is not a 'bool' object"):
 285             self._tc.create_stream_class(
 286                 default_clock_class=self._cc, packets_have_end_default_clock_snapshot=23
 287             )
 288
 289         self.assertEqual(len(self._tc), 0)
 290
 291     def test_does_not_support_packets_raises_with_begin_cs(self):
 292         with self.assertRaisesRegex(
 293             ValueError,
 294             "cannot not support packets, but have packet beginning default clock snapshot",
 295         ):
 296             self._tc.create_stream_class(
 297                 default_clock_class=self._cc,
 298                 packets_have_beginning_default_clock_snapshot=True,
 299             )
 300
 301         self.assertEqual(len(self._tc), 0)
 302
 303     def test_does_not_support_packets_raises_with_end_cs(self):
 304         with self.assertRaisesRegex(
 305             ValueError,
 306             "cannot not support packets, but have packet end default clock snapshots",
 307         ):
 308             self._tc.create_stream_class(
 309                 default_clock_class=self._cc,
 310                 packets_have_end_default_clock_snapshot=True,
 311             )
 312
 313         self.assertEqual(len(self._tc), 0)
 314
 315     def test_supports_discarded_events_without_cs(self):
 316         sc = self._tc.create_stream_class(
 317             default_clock_class=self._cc, supports_discarded_events=True
 318         )
 319         self.assertTrue(sc.supports_discarded_events)
 320         self.assertFalse(sc.discarded_events_have_default_clock_snapshots)
 321
 322     def test_supports_discarded_events_with_cs(self):
 323         sc = self._tc.create_stream_class(
 324             default_clock_class=self._cc,
 325             supports_discarded_events=True,
 326             discarded_events_have_default_clock_snapshots=True,
 327         )
 328         self.assertTrue(sc.supports_discarded_events)
 329         self.assertTrue(sc.discarded_events_have_default_clock_snapshots)
 330
 331     def test_supports_discarded_events_raises_type_error(self):
 332         with self.assertRaisesRegex(TypeError, "'int' is not a 'bool' object"):
 333             self._tc.create_stream_class(
 334                 default_clock_class=self._cc, supports_discarded_events=23
 335             )
 336
 337         self.assertEqual(len(self._tc), 0)
 338
 339     def test_discarded_events_have_default_cs_raises_type_error(self):
 340         with self.assertRaisesRegex(TypeError, "'int' is not a 'bool' object"):
 341             self._tc.create_stream_class(
 342                 default_clock_class=self._cc,
 343                 discarded_events_have_default_clock_snapshots=23,
 344             )
 345
 346         self.assertEqual(len(self._tc), 0)
 347
 348     def test_does_not_support_discarded_events_raises_with_cs(self):
 349         with self.assertRaisesRegex(
 350             ValueError,
 351             "cannot not support discarded events, but have default clock snapshots for discarded event messages",
 352         ):
 353             self._tc.create_stream_class(
 354                 default_clock_class=self._cc,
 355                 discarded_events_have_default_clock_snapshots=True,
 356             )
 357
 358         self.assertEqual(len(self._tc), 0)
 359
 360     def test_supports_discarded_packets_without_cs(self):
 361         sc = self._tc.create_stream_class(
 362             default_clock_class=self._cc,
 363             supports_discarded_packets=True,
 364             supports_packets=True,
 365         )
 366         self.assertTrue(sc.supports_discarded_packets)
 367         self.assertFalse(sc.discarded_packets_have_default_clock_snapshots)
 368
 369     def test_supports_discarded_packets_with_cs(self):
 370         sc = self._tc.create_stream_class(
 371             default_clock_class=self._cc,
 372             supports_discarded_packets=True,
 373             discarded_packets_have_default_clock_snapshots=True,
 374             supports_packets=True,
 375         )
 376         self.assertTrue(sc.supports_discarded_packets)
 377         self.assertTrue(sc.discarded_packets_have_default_clock_snapshots)
 378
 379     def test_supports_discarded_packets_raises_without_packet_support(self):
 380         with self.assertRaisesRegex(
 381             ValueError, "cannot support discarded packets, but not support packets"
 382         ):
 383             self._tc.create_stream_class(
 384                 default_clock_class=self._cc, supports_discarded_packets=True
 385             )
 386
 387         self.assertEqual(len(self._tc), 0)
 388
 389     def test_supports_discarded_packets_raises_type_error(self):
 390         with self.assertRaisesRegex(TypeError, "'int' is not a 'bool' object"):
 391             self._tc.create_stream_class(
 392                 default_clock_class=self._cc,
 393                 supports_discarded_packets=23,
 394                 supports_packets=True,
 395             )
 396
 397         self.assertEqual(len(self._tc), 0)
 398
 399     def test_discarded_packets_have_default_cs_raises_type_error(self):
 400         with self.assertRaisesRegex(TypeError, "'int' is not a 'bool' object"):
 401             self._tc.create_stream_class(
 402                 default_clock_class=self._cc,
 403                 discarded_packets_have_default_clock_snapshots=23,
 404                 supports_packets=True,
 405             )
 406
 407         self.assertEqual(len(self._tc), 0)
 408
 409     def test_does_not_support_discarded_packets_raises_with_cs(self):
 410         with self.assertRaisesRegex(
 411             ValueError,
 412             "cannot not support discarded packets, but have default clock snapshots for discarded packet messages",
 413         ):
 414             self._tc.create_stream_class(
 415                 default_clock_class=self._cc,
 416                 discarded_packets_have_default_clock_snapshots=True,
 417                 supports_packets=True,
 418             )
 419
 420         self.assertEqual(len(self._tc), 0)
 421
 422     def test_trace_class(self):
 423         sc = self._tc.create_stream_class()
 424         self.assertEqual(sc.trace_class.addr, self._tc.addr)
 425         self.assertIs(type(sc.trace_class), bt2_trace_class._TraceClass)
 426
 427     def _create_stream_class_with_event_classes(self):
 428         sc = self._tc.create_stream_class(assigns_automatic_event_class_id=False)
 429         ec1 = sc.create_event_class(id=23)
 430         ec2 = sc.create_event_class(id=17)
 431         return sc, ec1, ec2
 432
 433     def test_getitem(self):
 434         sc, ec1, ec2 = self._create_stream_class_with_event_classes()
 435
 436         self.assertEqual(sc[23].addr, ec1.addr)
 437         self.assertEqual(type(sc[23]), bt2_event_class._EventClass)
 438         self.assertEqual(sc[17].addr, ec2.addr)
 439         self.assertEqual(type(sc[17]), bt2_event_class._EventClass)
 440
 441     def test_getitem_wrong_key_type(self):
 442         sc, _, _ = self._create_stream_class_with_event_classes()
 443
 444         with self.assertRaisesRegex(TypeError, "'str' is not an 'int' object"):
 445             sc['event23']
 446
 447     def test_getitem_wrong_key(self):
 448         sc, _, _ = self._create_stream_class_with_event_classes()
 449
 450         with self.assertRaisesRegex(KeyError, '19'):
 451             sc[19]
 452
 453     def test_len(self):
 454         sc, _, _ = self._create_stream_class_with_event_classes()
 455
 456         self.assertEqual(len(sc), 2)
 457
 458     def test_iter(self):
 459         sc, _, _ = self._create_stream_class_with_event_classes()
 460
 461         ec_ids = sorted(sc)
 462         self.assertEqual(ec_ids, [17, 23])
 463
 464
 465 if __name__ == '__main__':
 466     unittest.main()