Add lttng-ctl SWIG python bindings
[lttng-tools.git] / extras / bindings / swig / python / tests / tests.py
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
236 calib = Calibrate()
237 calib.type = CALIBRATE_FUNCTION
238
239 ctx = EventContext()
240 ctx.type=EVENT_CONTEXT_PID
241
242 chattr = ChannelAttr()
243 chattr.overwrite = 0
244 chattr.subbuf_size = 4096
245 chattr.num_subbuf = 8
246 chattr.switch_timer_interval = 0
247 chattr.read_timer_interval = 200
248 chattr.output = EVENT_SPLICE
249
250 han = Handle("test_otherf" , dom)
251
252 r = create("test_otherf","/lttng-traces/test")
253 self.assertGreaterEqual(r, 0, strerror(r))
254
255 r = enable_event(han, event, None)
256 self.assertGreaterEqual(r, 0, strerror(r))
257
258 #Calibrate test
259 r = calibrate(han , calib)
260 self.assertGreaterEqual(r, 0, strerror(r))
261
262 #Context test
263 r = add_context(han, ctx, "sched_switch", "channel0")
264 self.assertGreaterEqual(r, 0, strerror(r))
265 #Any channel
266 r = add_context(han, ctx, "sched_wakeup", None)
267 self.assertGreaterEqual(r, 0, strerror(r))
268 #All events
269 r = add_context(han, ctx, None, None)
270 self.assertGreaterEqual(r, 0, strerror(r))
271
272 #Def. channel attr
273 channel_set_default_attr(dom, chattr)
274 channel_set_default_attr(None, None)
275
276 #Ses Daemon alive
277 r = session_daemon_alive()
278 self.assertTrue(r == 1 or r == 0, strerror(r))
279
280 #Setting trace group
281 r = set_tracing_group("testing")
282 self.assertGreaterEqual(r, 0, strerror(r))
283
284
285 r = start("test_otherf")
286 self.assertGreaterEqual(r, 0, strerror(r))
287 time.sleep(2)
288
289 r = stop("test_otherf")
290 self.assertGreaterEqual(r, 0, strerror(r))
291
292 del han
293
294 r = destroy("test_otherf")
295 self.assertGreaterEqual(r, 0, strerror(r))
296
297
298 if __name__ == "__main__":
299 # CHECK IF ROOT
300 if os.geteuid() == 0:
301 #Make sure session names don't already exist:
302 destroy("test_kernel_event")
303 destroy("test_kernel_all_events")
304 destroy("test_ust_all_events")
305 destroy("test_ust_event")
306 destroy("test_otherf")
307
308 unittest.main()
309 else:
310 print('Script must be run as root')
This page took 0.053314 seconds and 4 git commands to generate.