Commit | Line | Data |
---|---|---|
36907cb5 DS |
1 | import unittest |
2 | import os | |
3 | import time | |
4 | from lttng import * | |
5 | ||
6 | class TestLttngPythonModule (unittest.TestCase): | |
7 | ||
8 | def test_kernel_all_events(self): | |
9 | dom = Domain() | |
10 | dom.type = DOMAIN_KERNEL | |
11 | ||
12 | event = Event() | |
13 | event.type = EVENT_TRACEPOINT | |
14 | event.loglevel_type = EVENT_LOGLEVEL_ALL | |
15 | ||
16 | han = Handle("test_kernel_all_ev", dom) | |
17 | ||
18 | r = create("test_kernel_all_ev","/lttng-traces/test") | |
19 | self.assertGreaterEqual(r, 0, strerror(r)) | |
20 | ||
21 | r = enable_event(han, event, None) | |
22 | self.assertGreaterEqual(r, 0, strerror(r)) | |
23 | ||
24 | r = start("test_kernel_all_ev") | |
25 | self.assertGreaterEqual(r, 0, strerror(r)) | |
26 | time.sleep(2) | |
27 | ||
28 | r = stop("test_kernel_all_ev") | |
29 | self.assertGreaterEqual(r, 0, strerror(r)) | |
30 | ||
31 | r = destroy("test_kernel_all_ev") | |
32 | self.assertGreaterEqual(r, 0, strerror(r)) | |
33 | ||
34 | ||
35 | def test_kernel_event(self): | |
36 | ||
37 | dom = Domain() | |
38 | dom.type = DOMAIN_KERNEL | |
39 | ||
40 | channel = Channel() | |
41 | channel.name="mychan" | |
42 | channel.attr.overwrite = 0 | |
43 | channel.attr.subbuf_size = 4096 | |
44 | channel.attr.num_subbuf = 8 | |
45 | channel.attr.switch_timer_interval = 0 | |
46 | channel.attr.read_timer_interval = 200 | |
47 | channel.attr.output = EVENT_SPLICE | |
48 | ||
49 | sched_switch = Event() | |
50 | sched_switch.name = "sched_switch" | |
51 | sched_switch.type = EVENT_TRACEPOINT | |
52 | sched_switch.loglevel_type = EVENT_LOGLEVEL_ALL | |
53 | ||
54 | sched_process_exit = Event() | |
55 | sched_process_exit.name = "sched_process_exit" | |
56 | sched_process_exit.type = EVENT_TRACEPOINT | |
57 | sched_process_exit.loglevel_type = EVENT_LOGLEVEL_ALL | |
58 | ||
59 | sched_process_free = Event() | |
60 | sched_process_free.name = "sched_process_free" | |
61 | sched_process_free.type = EVENT_TRACEPOINT | |
62 | sched_process_free.loglevel_type = EVENT_LOGLEVEL_ALL | |
63 | ||
64 | han = Handle("test_kernel_event", dom) | |
65 | ||
66 | #Create session test | |
67 | r = create("test_kernel_event","/lttng-traces/test") | |
68 | self.assertGreaterEqual(r, 0, strerror(r)) | |
69 | ||
70 | #Enabling channel tests | |
71 | r = enable_channel(han, channel) | |
72 | self.assertGreaterEqual(r, 0, strerror(r)) | |
73 | ||
74 | #Enabling events tests | |
75 | r = enable_event(han, sched_switch, channel.name) | |
76 | self.assertGreaterEqual(r, 0, strerror(r)) | |
77 | ||
78 | r = enable_event(han, sched_process_exit, channel.name) | |
79 | self.assertGreaterEqual(r, 0, strerror(r)) | |
80 | ||
81 | r = enable_event(han, sched_process_free, channel.name) | |
82 | self.assertGreaterEqual(r, 0, strerror(r)) | |
83 | ||
84 | #Disabling events tests | |
85 | r = disable_event(han, sched_switch.name, channel.name) | |
86 | self.assertGreaterEqual(r, 0, strerror(r)) | |
87 | ||
88 | r = disable_event(han, sched_process_free.name, channel.name) | |
89 | self.assertGreaterEqual(r, 0, strerror(r)) | |
90 | ||
91 | #Renabling events tests | |
92 | r = enable_event(han, sched_switch, channel.name) | |
93 | self.assertGreaterEqual(r, 0, strerror(r)) | |
94 | ||
95 | r = enable_event(han, sched_process_free, channel.name) | |
96 | self.assertGreaterEqual(r, 0, strerror(r)) | |
97 | ||
98 | #Start, stop, destroy | |
99 | r = start("test_kernel_event") | |
100 | self.assertGreaterEqual(r, 0, strerror(r)) | |
101 | time.sleep(2) | |
102 | ||
103 | r = stop("test_kernel_event") | |
104 | self.assertGreaterEqual(r, 0, strerror(r)) | |
105 | ||
106 | r=disable_channel(han, channel.name) | |
107 | self.assertGreaterEqual(r, 0, strerror(r)) | |
108 | ||
109 | r=destroy("test_kernel_event") | |
110 | self.assertGreaterEqual(r, 0, strerror(r)) | |
111 | ||
112 | ||
113 | ||
114 | def test_ust_all_events(self): | |
115 | dom = Domain() | |
116 | dom.type = DOMAIN_UST | |
117 | ||
118 | event = Event() | |
119 | event.type = EVENT_TRACEPOINT | |
120 | event.loglevel_type = EVENT_LOGLEVEL_ALL | |
121 | ||
122 | han = Handle("test_ust_all_ev", dom) | |
123 | ||
124 | r = create("test_ust_all_ev","/lttng-traces/test") | |
125 | self.assertGreaterEqual(r, 0, strerror(r)) | |
126 | ||
127 | r = enable_event(han, event, None) | |
128 | self.assertGreaterEqual(r, 0, strerror(r)) | |
129 | ||
130 | r = start("test_ust_all_ev") | |
131 | self.assertGreaterEqual(r, 0, strerror(r)) | |
132 | time.sleep(2) | |
133 | ||
134 | r = stop("test_ust_all_ev") | |
135 | self.assertGreaterEqual(r, 0, strerror(r)) | |
136 | ||
137 | r = destroy("test_ust_all_ev") | |
138 | self.assertGreaterEqual(r, 0, strerror(r)) | |
139 | ||
140 | ||
141 | def test_ust_event(self): | |
142 | ||
143 | dom = Domain() | |
144 | dom.type = DOMAIN_UST | |
145 | ||
146 | channel = Channel() | |
147 | channel.name="mychan" | |
148 | channel.attr.overwrite = 0 | |
149 | channel.attr.subbuf_size = 4096 | |
150 | channel.attr.num_subbuf = 8 | |
151 | channel.attr.switch_timer_interval = 0 | |
152 | channel.attr.read_timer_interval = 200 | |
153 | channel.attr.output = EVENT_MMAP | |
154 | ||
155 | ev1 = Event() | |
156 | ev1.name = "tp1" | |
157 | ev1.type = EVENT_TRACEPOINT | |
158 | ev1.loglevel_type = EVENT_LOGLEVEL_ALL | |
159 | ||
160 | ev2 = Event() | |
161 | ev2.name = "ev2" | |
162 | ev2.type = EVENT_TRACEPOINT | |
163 | ev2.loglevel_type = EVENT_LOGLEVEL_ALL | |
164 | ||
165 | ev3 = Event() | |
166 | ev3.name = "ev3" | |
167 | ev3.type = EVENT_TRACEPOINT | |
168 | ev3.loglevel_type = EVENT_LOGLEVEL_ALL | |
169 | ||
170 | han = Handle("test_ust_event", dom) | |
171 | ||
172 | #Create session test | |
173 | r = create("test_ust_event","/lttng-traces/test") | |
174 | self.assertGreaterEqual(r, 0, strerror(r)) | |
175 | ||
176 | #Enabling channel tests | |
177 | r = enable_channel(han, channel) | |
178 | self.assertGreaterEqual(r, 0, strerror(r)) | |
179 | ||
180 | #Enabling events tests | |
181 | r = enable_event(han, ev1, channel.name) | |
182 | self.assertGreaterEqual(r, 0, strerror(r)) | |
183 | ||
184 | r = enable_event(han, ev2, channel.name) | |
185 | self.assertGreaterEqual(r, 0, strerror(r)) | |
186 | ||
187 | r = enable_event(han, ev3, channel.name) | |
188 | self.assertGreaterEqual(r, 0, strerror(r)) | |
189 | ||
190 | #Disabling events tests | |
191 | r = disable_event(han, ev1.name, channel.name) | |
192 | self.assertGreaterEqual(r, 0, strerror(r)) | |
193 | ||
194 | r = disable_event(han, ev3.name, channel.name) | |
195 | self.assertGreaterEqual(r, 0, strerror(r)) | |
196 | ||
197 | #Renabling events tests | |
198 | r = enable_event(han, ev1, channel.name) | |
199 | self.assertGreaterEqual(r, 0, strerror(r)) | |
200 | ||
201 | r = enable_event(han, ev3, channel.name) | |
202 | self.assertGreaterEqual(r, 0, strerror(r)) | |
203 | ||
204 | #Start, stop | |
205 | r = start("test_ust_event") | |
206 | self.assertGreaterEqual(r, 0, strerror(r)) | |
207 | time.sleep(2) | |
208 | ||
209 | r = stop("test_ust_event") | |
210 | self.assertGreaterEqual(r, 0, strerror(r)) | |
211 | ||
212 | #Restart/restop | |
213 | r = start("test_ust_event") | |
214 | self.assertGreaterEqual(r, 0, strerror(r)) | |
215 | time.sleep(2) | |
216 | ||
217 | r = stop("test_ust_event") | |
218 | self.assertGreaterEqual(r, 0, strerror(r)) | |
219 | ||
220 | #Disabling channel and destroy | |
221 | r=disable_channel(han, channel.name) | |
222 | self.assertGreaterEqual(r, 0, strerror(r)) | |
223 | ||
224 | r=destroy("test_ust_event") | |
225 | self.assertGreaterEqual(r, 0, strerror(r)) | |
226 | ||
227 | ||
228 | def test_other_functions(self): | |
229 | dom = Domain() | |
230 | dom.type=DOMAIN_KERNEL | |
231 | ||
232 | event=Event() | |
233 | event.type=EVENT_TRACEPOINT | |
234 | event.loglevel_type=EVENT_LOGLEVEL_ALL | |
235 | ||
36907cb5 DS |
236 | ctx = EventContext() |
237 | ctx.type=EVENT_CONTEXT_PID | |
238 | ||
239 | chattr = ChannelAttr() | |
240 | chattr.overwrite = 0 | |
241 | chattr.subbuf_size = 4096 | |
242 | chattr.num_subbuf = 8 | |
243 | chattr.switch_timer_interval = 0 | |
244 | chattr.read_timer_interval = 200 | |
245 | chattr.output = EVENT_SPLICE | |
246 | ||
247 | han = Handle("test_otherf" , dom) | |
248 | ||
249 | r = create("test_otherf","/lttng-traces/test") | |
250 | self.assertGreaterEqual(r, 0, strerror(r)) | |
251 | ||
252 | r = enable_event(han, event, None) | |
253 | self.assertGreaterEqual(r, 0, strerror(r)) | |
254 | ||
36907cb5 DS |
255 | #Context test |
256 | r = add_context(han, ctx, "sched_switch", "channel0") | |
257 | self.assertGreaterEqual(r, 0, strerror(r)) | |
258 | #Any channel | |
259 | r = add_context(han, ctx, "sched_wakeup", None) | |
260 | self.assertGreaterEqual(r, 0, strerror(r)) | |
261 | #All events | |
262 | r = add_context(han, ctx, None, None) | |
263 | self.assertGreaterEqual(r, 0, strerror(r)) | |
264 | ||
265 | #Def. channel attr | |
266 | channel_set_default_attr(dom, chattr) | |
267 | channel_set_default_attr(None, None) | |
268 | ||
269 | #Ses Daemon alive | |
270 | r = session_daemon_alive() | |
271 | self.assertTrue(r == 1 or r == 0, strerror(r)) | |
272 | ||
273 | #Setting trace group | |
274 | r = set_tracing_group("testing") | |
275 | self.assertGreaterEqual(r, 0, strerror(r)) | |
276 | ||
277 | ||
278 | r = start("test_otherf") | |
279 | self.assertGreaterEqual(r, 0, strerror(r)) | |
280 | time.sleep(2) | |
281 | ||
282 | r = stop("test_otherf") | |
283 | self.assertGreaterEqual(r, 0, strerror(r)) | |
284 | ||
285 | del han | |
286 | ||
287 | r = destroy("test_otherf") | |
288 | self.assertGreaterEqual(r, 0, strerror(r)) | |
289 | ||
290 | ||
291 | if __name__ == "__main__": | |
292 | # CHECK IF ROOT | |
293 | if os.geteuid() == 0: | |
294 | #Make sure session names don't already exist: | |
295 | destroy("test_kernel_event") | |
296 | destroy("test_kernel_all_events") | |
297 | destroy("test_ust_all_events") | |
298 | destroy("test_ust_event") | |
299 | destroy("test_otherf") | |
300 | ||
301 | unittest.main() | |
302 | else: | |
303 | print('Script must be run as root') |