Move to kernel style SPDX license identifiers
[lttng-tools.git] / extras / bindings / swig / python / tests / tests.py
1 #
2 # Copyright (C) 2012 Danny Serres <danny.serres@efficios.com>
3 #
4 # SPDX-License-Identifier: GPL-2.0-only
5 #
6
7 import unittest
8 import os
9 import time
10 import tempfile
11 from lttng import *
12
13 class TestLttngPythonModule (unittest.TestCase):
14
15 def setUp(self):
16 self.tmpdir = tempfile.TemporaryDirectory()
17
18 def tearDown(self):
19 self.tmpdir.cleanup()
20
21 def test_kernel_all_events(self):
22 dom = Domain()
23 dom.type = DOMAIN_KERNEL
24 dom.buf_type = BUFFER_GLOBAL
25
26 event = Event()
27 event.type = EVENT_TRACEPOINT
28 event.loglevel_type = EVENT_LOGLEVEL_ALL
29
30 han = Handle("test_kernel_all_ev", dom)
31
32 r = create("test_kernel_all_ev", self.tmpdir.name)
33 self.assertGreaterEqual(r, 0, strerror(r))
34
35 r = enable_event(han, event, None)
36 self.assertGreaterEqual(r, 0, strerror(r))
37
38 r = start("test_kernel_all_ev")
39 self.assertGreaterEqual(r, 0, strerror(r))
40 time.sleep(2)
41
42 r = stop("test_kernel_all_ev")
43 self.assertGreaterEqual(r, 0, strerror(r))
44
45 r = destroy("test_kernel_all_ev")
46 self.assertGreaterEqual(r, 0, strerror(r))
47
48
49 def test_kernel_event(self):
50
51 dom = Domain()
52 dom.type = DOMAIN_KERNEL
53 dom.buf_type = BUFFER_GLOBAL
54
55 channel = Channel()
56 channel.name="mychan"
57 channel.attr.overwrite = 0
58 channel.attr.subbuf_size = 4096
59 channel.attr.num_subbuf = 8
60 channel.attr.switch_timer_interval = 0
61 channel.attr.read_timer_interval = 200
62 channel.attr.output = EVENT_SPLICE
63
64 sched_switch = Event()
65 sched_switch.name = "sched_switch"
66 sched_switch.type = EVENT_TRACEPOINT
67 sched_switch.loglevel_type = EVENT_LOGLEVEL_ALL
68
69 sched_process_exit = Event()
70 sched_process_exit.name = "sched_process_exit"
71 sched_process_exit.type = EVENT_TRACEPOINT
72 sched_process_exit.loglevel_type = EVENT_LOGLEVEL_ALL
73
74 sched_process_free = Event()
75 sched_process_free.name = "sched_process_free"
76 sched_process_free.type = EVENT_TRACEPOINT
77 sched_process_free.loglevel_type = EVENT_LOGLEVEL_ALL
78
79 han = Handle("test_kernel_event", dom)
80
81 #Create session test
82 r = create("test_kernel_event", self.tmpdir.name)
83 self.assertGreaterEqual(r, 0, strerror(r))
84
85 #Enabling channel tests
86 r = enable_channel(han, channel)
87 self.assertGreaterEqual(r, 0, strerror(r))
88
89 #Enabling events tests
90 r = enable_event(han, sched_switch, channel.name)
91 self.assertGreaterEqual(r, 0, strerror(r))
92
93 r = enable_event(han, sched_process_exit, channel.name)
94 self.assertGreaterEqual(r, 0, strerror(r))
95
96 r = enable_event(han, sched_process_free, channel.name)
97 self.assertGreaterEqual(r, 0, strerror(r))
98
99 #Disabling events tests
100 r = disable_event(han, sched_switch.name, channel.name)
101 self.assertGreaterEqual(r, 0, strerror(r))
102
103 r = disable_event(han, sched_process_free.name, channel.name)
104 self.assertGreaterEqual(r, 0, strerror(r))
105
106 #Renabling events tests
107 r = enable_event(han, sched_switch, channel.name)
108 self.assertGreaterEqual(r, 0, strerror(r))
109
110 r = enable_event(han, sched_process_free, channel.name)
111 self.assertGreaterEqual(r, 0, strerror(r))
112
113 #Start, stop, destroy
114 r = start("test_kernel_event")
115 self.assertGreaterEqual(r, 0, strerror(r))
116 time.sleep(2)
117
118 r = stop("test_kernel_event")
119 self.assertGreaterEqual(r, 0, strerror(r))
120
121 r=disable_channel(han, channel.name)
122 self.assertGreaterEqual(r, 0, strerror(r))
123
124 r=destroy("test_kernel_event")
125 self.assertGreaterEqual(r, 0, strerror(r))
126
127
128
129 def test_ust_all_events(self):
130 dom = Domain()
131 dom.type = DOMAIN_UST
132 dom.buf_type = BUFFER_PER_UID
133
134 event = Event()
135 event.type = EVENT_TRACEPOINT
136 event.loglevel_type = EVENT_LOGLEVEL_ALL
137
138 han = Handle("test_ust_all_ev", dom)
139
140 r = create("test_ust_all_ev", self.tmpdir.name)
141 self.assertGreaterEqual(r, 0, strerror(r))
142
143 r = enable_event(han, event, None)
144 self.assertGreaterEqual(r, 0, strerror(r))
145
146 r = start("test_ust_all_ev")
147 self.assertGreaterEqual(r, 0, strerror(r))
148 time.sleep(2)
149
150 r = stop("test_ust_all_ev")
151 self.assertGreaterEqual(r, 0, strerror(r))
152
153 r = destroy("test_ust_all_ev")
154 self.assertGreaterEqual(r, 0, strerror(r))
155
156
157 def test_ust_event(self):
158
159 dom = Domain()
160 dom.type = DOMAIN_UST
161 dom.buf_type = BUFFER_PER_UID
162
163 channel = Channel()
164 channel.name="mychan"
165 channel.attr.overwrite = 0
166 channel.attr.subbuf_size = 4096
167 channel.attr.num_subbuf = 8
168 channel.attr.switch_timer_interval = 0
169 channel.attr.read_timer_interval = 200
170 channel.attr.output = EVENT_MMAP
171
172 ev1 = Event()
173 ev1.name = "tp1"
174 ev1.type = EVENT_TRACEPOINT
175 ev1.loglevel_type = EVENT_LOGLEVEL_ALL
176
177 ev2 = Event()
178 ev2.name = "ev2"
179 ev2.type = EVENT_TRACEPOINT
180 ev2.loglevel_type = EVENT_LOGLEVEL_ALL
181
182 ev3 = Event()
183 ev3.name = "ev3"
184 ev3.type = EVENT_TRACEPOINT
185 ev3.loglevel_type = EVENT_LOGLEVEL_ALL
186
187 han = Handle("test_ust_event", dom)
188
189 #Create session test
190 r = create("test_ust_event", self.tmpdir.name)
191 self.assertGreaterEqual(r, 0, strerror(r))
192
193 #Enabling channel tests
194 r = enable_channel(han, channel)
195 self.assertGreaterEqual(r, 0, strerror(r))
196
197 #Enabling events tests
198 r = enable_event(han, ev1, channel.name)
199 self.assertGreaterEqual(r, 0, strerror(r))
200
201 r = enable_event(han, ev2, channel.name)
202 self.assertGreaterEqual(r, 0, strerror(r))
203
204 r = enable_event(han, ev3, channel.name)
205 self.assertGreaterEqual(r, 0, strerror(r))
206
207 #Disabling events tests
208 r = disable_event(han, ev1.name, channel.name)
209 self.assertGreaterEqual(r, 0, strerror(r))
210
211 r = disable_event(han, ev3.name, channel.name)
212 self.assertGreaterEqual(r, 0, strerror(r))
213
214 #Renabling events tests
215 r = enable_event(han, ev1, channel.name)
216 self.assertGreaterEqual(r, 0, strerror(r))
217
218 r = enable_event(han, ev3, channel.name)
219 self.assertGreaterEqual(r, 0, strerror(r))
220
221 #Start, stop
222 r = start("test_ust_event")
223 self.assertGreaterEqual(r, 0, strerror(r))
224 time.sleep(2)
225
226 r = stop("test_ust_event")
227 self.assertGreaterEqual(r, 0, strerror(r))
228
229 #Restart/restop
230 r = start("test_ust_event")
231 self.assertGreaterEqual(r, 0, strerror(r))
232 time.sleep(2)
233
234 r = stop("test_ust_event")
235 self.assertGreaterEqual(r, 0, strerror(r))
236
237 #Disabling channel and destroy
238 r=disable_channel(han, channel.name)
239 self.assertGreaterEqual(r, 0, strerror(r))
240
241 r=destroy("test_ust_event")
242 self.assertGreaterEqual(r, 0, strerror(r))
243
244
245 def test_other_functions(self):
246 dom = Domain()
247 dom.type=DOMAIN_KERNEL
248 dom.buf_type = BUFFER_GLOBAL
249
250 event=Event()
251 event.type=EVENT_TRACEPOINT
252 event.loglevel_type=EVENT_LOGLEVEL_ALL
253
254 ctx = EventContext()
255 ctx.type=EVENT_CONTEXT_PID
256
257 chattr = ChannelAttr()
258 chattr.overwrite = 0
259 chattr.subbuf_size = 4096
260 chattr.num_subbuf = 8
261 chattr.switch_timer_interval = 0
262 chattr.read_timer_interval = 200
263 chattr.output = EVENT_SPLICE
264
265 han = Handle("test_otherf" , dom)
266
267 r = create("test_otherf", self.tmpdir.name)
268 self.assertGreaterEqual(r, 0, strerror(r))
269
270 r = enable_event(han, event, None)
271 self.assertGreaterEqual(r, 0, strerror(r))
272
273 #Context test
274 r = add_context(han, ctx, "sched_switch", "channel0")
275 self.assertGreaterEqual(r, 0, strerror(r))
276 #Any channel
277 r = add_context(han, ctx, "sched_wakeup", None)
278 self.assertGreaterEqual(r, 0, strerror(r))
279 #All events
280 r = add_context(han, ctx, None, None)
281 self.assertGreaterEqual(r, 0, strerror(r))
282
283 #Def. channel attr
284 channel_set_default_attr(dom, chattr)
285 channel_set_default_attr(None, None)
286
287 #Ses Daemon alive
288 r = session_daemon_alive()
289 self.assertTrue(r == 1 or r == 0, strerror(r))
290
291 #Setting trace group
292 r = set_tracing_group("testing")
293 self.assertGreaterEqual(r, 0, strerror(r))
294
295
296 r = start("test_otherf")
297 self.assertGreaterEqual(r, 0, strerror(r))
298 time.sleep(2)
299
300 r = stop("test_otherf")
301 self.assertGreaterEqual(r, 0, strerror(r))
302
303 domains = list_domains("test_otherf")
304 self.assertTrue(domains[0].type == DOMAIN_KERNEL)
305 self.assertTrue(domains[0].buf_type == BUFFER_GLOBAL)
306
307 del han
308
309 r = destroy("test_otherf")
310 self.assertGreaterEqual(r, 0, strerror(r))
311
312
313 def ust_suite():
314 suite = unittest.TestSuite()
315 suite.addTest(TestLttngPythonModule("test_ust_event"))
316 suite.addTest(TestLttngPythonModule("test_ust_all_events"))
317 return suite
318
319 def kernel_suite():
320 suite = unittest.TestSuite()
321 suite.addTest(TestLttngPythonModule("test_kernel_event"))
322 suite.addTest(TestLttngPythonModule("test_kernel_all_events"))
323 suite.addTest(TestLttngPythonModule("test_other_functions"))
324 return suite
325
326 if __name__ == '__main__':
327 destroy("test_kernel_event")
328 destroy("test_kernel_all_events")
329 destroy("test_ust_all_events")
330 destroy("test_ust_event")
331 destroy("test_otherf")
332
333 runner = unittest.TextTestRunner(verbosity=2)
334
335 if os.geteuid() == 0:
336 runner.run(kernel_suite())
337
338 runner.run(ust_suite())
This page took 0.056357 seconds and 5 git commands to generate.