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