Commit | Line | Data |
---|---|---|
0bee8ea9 SM |
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 | ||
116779e3 | 48 | with self.assertRaises(ValueError): |
0bee8ea9 SM |
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. | |
116779e3 | 69 | with self.assertRaises(ValueError): |
0bee8ea9 SM |
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(): | |
78668ecd | 125 | self.assertIsInstance(stream_class, bt2.stream_class._StreamClass) |
0bee8ea9 SM |
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) |