| 1 | import uuid |
| 2 | import unittest |
| 3 | import bt2 |
| 4 | from utils import run_in_component_init, get_default_trace_class |
| 5 | |
| 6 | |
| 7 | class TraceClassTestCase(unittest.TestCase): |
| 8 | |
| 9 | def test_create_default(self): |
| 10 | def f(comp_self): |
| 11 | return comp_self._create_trace_class() |
| 12 | |
| 13 | tc = run_in_component_init(f) |
| 14 | |
| 15 | self.assertEqual(len(tc), 0) |
| 16 | self.assertEqual(len(tc.env), 0) |
| 17 | self.assertIsNone(tc.uuid) |
| 18 | self.assertTrue(tc.assigns_automatic_stream_class_id) |
| 19 | |
| 20 | def test_uuid(self): |
| 21 | def f(comp_self): |
| 22 | return comp_self._create_trace_class(uuid=uuid.UUID('da7d6b6f-3108-4706-89bd-ab554732611b')) |
| 23 | |
| 24 | tc = run_in_component_init(f) |
| 25 | |
| 26 | self.assertEqual(tc.uuid, uuid.UUID('da7d6b6f-3108-4706-89bd-ab554732611b')) |
| 27 | |
| 28 | def test_automatic_stream_class_id(self): |
| 29 | def f(comp_self): |
| 30 | return comp_self._create_trace_class(assigns_automatic_stream_class_id=True) |
| 31 | |
| 32 | tc = run_in_component_init(f) |
| 33 | self.assertTrue(tc.assigns_automatic_stream_class_id) |
| 34 | |
| 35 | # This should not throw. |
| 36 | sc1 = tc.create_stream_class() |
| 37 | sc2 = tc.create_stream_class() |
| 38 | |
| 39 | self.assertNotEqual(sc1.id, sc2.id) |
| 40 | |
| 41 | def test_automatic_stream_class_id_raises(self): |
| 42 | def f(comp_self): |
| 43 | return comp_self._create_trace_class(assigns_automatic_stream_class_id=True) |
| 44 | |
| 45 | tc = run_in_component_init(f) |
| 46 | self.assertTrue(tc.assigns_automatic_stream_class_id) |
| 47 | |
| 48 | with self.assertRaises(ValueError): |
| 49 | sc1 = tc.create_stream_class(23) |
| 50 | |
| 51 | def test_no_assigns_automatic_stream_class_id(self): |
| 52 | def f(comp_self): |
| 53 | return comp_self._create_trace_class(assigns_automatic_stream_class_id=False) |
| 54 | |
| 55 | tc = run_in_component_init(f) |
| 56 | self.assertFalse(tc.assigns_automatic_stream_class_id) |
| 57 | |
| 58 | sc = tc.create_stream_class(id=28) |
| 59 | self.assertEqual(sc.id, 28) |
| 60 | |
| 61 | def test_no_assigns_automatic_stream_class_id_raises(self): |
| 62 | def f(comp_self): |
| 63 | return comp_self._create_trace_class(assigns_automatic_stream_class_id=False) |
| 64 | |
| 65 | tc = run_in_component_init(f) |
| 66 | self.assertFalse(tc.assigns_automatic_stream_class_id) |
| 67 | |
| 68 | # In this mode, it is required to pass an explicit id. |
| 69 | with self.assertRaises(ValueError): |
| 70 | tc.create_stream_class() |
| 71 | |
| 72 | def test_env_get(self): |
| 73 | def f(comp_self): |
| 74 | return comp_self._create_trace_class(env={'hello': 'you', 'foo': -5}) |
| 75 | |
| 76 | tc = run_in_component_init(f) |
| 77 | |
| 78 | self.assertEqual(tc.env['hello'], 'you') |
| 79 | self.assertEqual(tc.env['foo'], -5) |
| 80 | |
| 81 | def test_env_get_non_existent(self): |
| 82 | def f(comp_self): |
| 83 | return comp_self._create_trace_class(env={'hello': 'you', 'foo': -5}) |
| 84 | |
| 85 | tc = run_in_component_init(f) |
| 86 | |
| 87 | with self.assertRaises(KeyError): |
| 88 | tc.env['lel'] |
| 89 | |
| 90 | @staticmethod |
| 91 | def _create_trace_class_with_some_stream_classes(): |
| 92 | def f(comp_self): |
| 93 | return comp_self._create_trace_class(assigns_automatic_stream_class_id=False) |
| 94 | |
| 95 | tc = run_in_component_init(f) |
| 96 | sc1 = tc.create_stream_class(id=12) |
| 97 | sc2 = tc.create_stream_class(id=54) |
| 98 | sc3 = tc.create_stream_class(id=2018) |
| 99 | return tc, sc1, sc2, sc3 |
| 100 | |
| 101 | def test_getitem(self): |
| 102 | tc, _, _, sc3 = self._create_trace_class_with_some_stream_classes() |
| 103 | self.assertEqual(tc[2018].addr, sc3.addr) |
| 104 | |
| 105 | def test_getitem_wrong_key_type(self): |
| 106 | tc, _, _, _ = self._create_trace_class_with_some_stream_classes() |
| 107 | with self.assertRaises(TypeError): |
| 108 | tc['hello'] |
| 109 | |
| 110 | def test_getitem_wrong_key(self): |
| 111 | tc, _, _, _ = self._create_trace_class_with_some_stream_classes() |
| 112 | with self.assertRaises(KeyError): |
| 113 | tc[4] |
| 114 | |
| 115 | def test_len(self): |
| 116 | tc = get_default_trace_class() |
| 117 | self.assertEqual(len(tc), 0) |
| 118 | tc.create_stream_class() |
| 119 | self.assertEqual(len(tc), 1) |
| 120 | |
| 121 | def test_iter(self): |
| 122 | tc, sc1, sc2, sc3 = self._create_trace_class_with_some_stream_classes() |
| 123 | |
| 124 | for sc_id, stream_class in tc.items(): |
| 125 | self.assertIsInstance(stream_class, bt2.stream_class._StreamClass) |
| 126 | |
| 127 | if sc_id == 12: |
| 128 | self.assertEqual(stream_class.addr, sc1.addr) |
| 129 | elif sc_id == 54: |
| 130 | self.assertEqual(stream_class.addr, sc2.addr) |
| 131 | elif sc_id == 2018: |
| 132 | self.assertEqual(stream_class.addr, sc3.addr) |
| 133 | |
| 134 | def test_destruction_listener(self): |
| 135 | def on_trace_class_destruction(trace_class): |
| 136 | nonlocal trace_class_destroyed |
| 137 | trace_class_destroyed = True |
| 138 | |
| 139 | trace_class_destroyed = False |
| 140 | |
| 141 | trace_class = get_default_trace_class() |
| 142 | trace_class.add_destruction_listener(on_trace_class_destruction) |
| 143 | |
| 144 | self.assertFalse(trace_class_destroyed) |
| 145 | |
| 146 | del trace_class |
| 147 | |
| 148 | self.assertTrue(trace_class_destroyed) |