Update base test for binding
[lttng-tools.git] / extras / bindings / swig / python / tests / tests.py
CommitLineData
36907cb5
DS
1import unittest
2import os
3import time
68f8c317 4import tempfile
36907cb5
DS
5from lttng import *
6
7class TestLttngPythonModule (unittest.TestCase):
8
68f8c317
JR
9 def setUp(self):
10 self.tmpdir = tempfile.TemporaryDirectory()
36907cb5 11
68f8c317
JR
12 def tearDown(self):
13 self.tmpdir.cleanup()
36907cb5 14
68f8c317
JR
15 def test_kernel_all_events(self):
16 dom = Domain()
17 dom.type = DOMAIN_KERNEL
18 dom.buf_type = BUFFER_GLOBAL
36907cb5 19
68f8c317
JR
20 event = Event()
21 event.type = EVENT_TRACEPOINT
22 event.loglevel_type = EVENT_LOGLEVEL_ALL
36907cb5 23
68f8c317 24 han = Handle("test_kernel_all_ev", dom)
36907cb5 25
68f8c317
JR
26 r = create("test_kernel_all_ev", self.tmpdir.name)
27 self.assertGreaterEqual(r, 0, strerror(r))
36907cb5 28
68f8c317
JR
29 r = enable_event(han, event, None)
30 self.assertGreaterEqual(r, 0, strerror(r))
36907cb5 31
68f8c317
JR
32 r = start("test_kernel_all_ev")
33 self.assertGreaterEqual(r, 0, strerror(r))
34 time.sleep(2)
36907cb5 35
68f8c317
JR
36 r = stop("test_kernel_all_ev")
37 self.assertGreaterEqual(r, 0, strerror(r))
36907cb5 38
68f8c317
JR
39 r = destroy("test_kernel_all_ev")
40 self.assertGreaterEqual(r, 0, strerror(r))
36907cb5 41
36907cb5 42
68f8c317 43 def test_kernel_event(self):
36907cb5 44
68f8c317
JR
45 dom = Domain()
46 dom.type = DOMAIN_KERNEL
47 dom.buf_type = BUFFER_GLOBAL
36907cb5 48
68f8c317
JR
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
36907cb5 57
68f8c317
JR
58 sched_switch = Event()
59 sched_switch.name = "sched_switch"
60 sched_switch.type = EVENT_TRACEPOINT
61 sched_switch.loglevel_type = EVENT_LOGLEVEL_ALL
36907cb5 62
68f8c317
JR
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
36907cb5 67
68f8c317
JR
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
36907cb5 72
68f8c317 73 han = Handle("test_kernel_event", dom)
36907cb5 74
68f8c317
JR
75 #Create session test
76 r = create("test_kernel_event", self.tmpdir.name)
77 self.assertGreaterEqual(r, 0, strerror(r))
36907cb5 78
68f8c317
JR
79 #Enabling channel tests
80 r = enable_channel(han, channel)
81 self.assertGreaterEqual(r, 0, strerror(r))
36907cb5 82
68f8c317
JR
83 #Enabling events tests
84 r = enable_event(han, sched_switch, channel.name)
85 self.assertGreaterEqual(r, 0, strerror(r))
36907cb5 86
68f8c317
JR
87 r = enable_event(han, sched_process_exit, channel.name)
88 self.assertGreaterEqual(r, 0, strerror(r))
36907cb5 89
68f8c317
JR
90 r = enable_event(han, sched_process_free, channel.name)
91 self.assertGreaterEqual(r, 0, strerror(r))
36907cb5 92
68f8c317
JR
93 #Disabling events tests
94 r = disable_event(han, sched_switch.name, channel.name)
95 self.assertGreaterEqual(r, 0, strerror(r))
36907cb5 96
68f8c317
JR
97 r = disable_event(han, sched_process_free.name, channel.name)
98 self.assertGreaterEqual(r, 0, strerror(r))
36907cb5 99
68f8c317
JR
100 #Renabling events tests
101 r = enable_event(han, sched_switch, channel.name)
102 self.assertGreaterEqual(r, 0, strerror(r))
36907cb5 103
68f8c317
JR
104 r = enable_event(han, sched_process_free, channel.name)
105 self.assertGreaterEqual(r, 0, strerror(r))
36907cb5 106
68f8c317
JR
107 #Start, stop, destroy
108 r = start("test_kernel_event")
109 self.assertGreaterEqual(r, 0, strerror(r))
110 time.sleep(2)
36907cb5 111
68f8c317
JR
112 r = stop("test_kernel_event")
113 self.assertGreaterEqual(r, 0, strerror(r))
36907cb5 114
68f8c317
JR
115 r=disable_channel(han, channel.name)
116 self.assertGreaterEqual(r, 0, strerror(r))
36907cb5 117
68f8c317
JR
118 r=destroy("test_kernel_event")
119 self.assertGreaterEqual(r, 0, strerror(r))
36907cb5 120
36907cb5 121
36907cb5 122
68f8c317
JR
123 def test_ust_all_events(self):
124 dom = Domain()
125 dom.type = DOMAIN_UST
126 dom.buf_type = BUFFER_PER_UID
36907cb5 127
68f8c317
JR
128 event = Event()
129 event.type = EVENT_TRACEPOINT
130 event.loglevel_type = EVENT_LOGLEVEL_ALL
36907cb5 131
68f8c317 132 han = Handle("test_ust_all_ev", dom)
36907cb5 133
68f8c317
JR
134 r = create("test_ust_all_ev", self.tmpdir.name)
135 self.assertGreaterEqual(r, 0, strerror(r))
36907cb5 136
68f8c317
JR
137 r = enable_event(han, event, None)
138 self.assertGreaterEqual(r, 0, strerror(r))
36907cb5 139
68f8c317
JR
140 r = start("test_ust_all_ev")
141 self.assertGreaterEqual(r, 0, strerror(r))
142 time.sleep(2)
36907cb5 143
68f8c317
JR
144 r = stop("test_ust_all_ev")
145 self.assertGreaterEqual(r, 0, strerror(r))
36907cb5 146
68f8c317
JR
147 r = destroy("test_ust_all_ev")
148 self.assertGreaterEqual(r, 0, strerror(r))
36907cb5 149
36907cb5 150
68f8c317 151 def test_ust_event(self):
36907cb5 152
68f8c317
JR
153 dom = Domain()
154 dom.type = DOMAIN_UST
155 dom.buf_type = BUFFER_PER_UID
36907cb5 156
68f8c317
JR
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
36907cb5 165
68f8c317
JR
166 ev1 = Event()
167 ev1.name = "tp1"
168 ev1.type = EVENT_TRACEPOINT
169 ev1.loglevel_type = EVENT_LOGLEVEL_ALL
36907cb5 170
68f8c317
JR
171 ev2 = Event()
172 ev2.name = "ev2"
173 ev2.type = EVENT_TRACEPOINT
174 ev2.loglevel_type = EVENT_LOGLEVEL_ALL
36907cb5 175
68f8c317
JR
176 ev3 = Event()
177 ev3.name = "ev3"
178 ev3.type = EVENT_TRACEPOINT
179 ev3.loglevel_type = EVENT_LOGLEVEL_ALL
36907cb5 180
68f8c317 181 han = Handle("test_ust_event", dom)
36907cb5 182
68f8c317
JR
183 #Create session test
184 r = create("test_ust_event", self.tmpdir.name)
185 self.assertGreaterEqual(r, 0, strerror(r))
36907cb5 186
68f8c317
JR
187 #Enabling channel tests
188 r = enable_channel(han, channel)
189 self.assertGreaterEqual(r, 0, strerror(r))
36907cb5 190
68f8c317
JR
191 #Enabling events tests
192 r = enable_event(han, ev1, channel.name)
193 self.assertGreaterEqual(r, 0, strerror(r))
36907cb5 194
68f8c317
JR
195 r = enable_event(han, ev2, channel.name)
196 self.assertGreaterEqual(r, 0, strerror(r))
36907cb5 197
68f8c317
JR
198 r = enable_event(han, ev3, channel.name)
199 self.assertGreaterEqual(r, 0, strerror(r))
36907cb5 200
68f8c317
JR
201 #Disabling events tests
202 r = disable_event(han, ev1.name, channel.name)
203 self.assertGreaterEqual(r, 0, strerror(r))
36907cb5 204
68f8c317
JR
205 r = disable_event(han, ev3.name, channel.name)
206 self.assertGreaterEqual(r, 0, strerror(r))
36907cb5 207
68f8c317
JR
208 #Renabling events tests
209 r = enable_event(han, ev1, channel.name)
210 self.assertGreaterEqual(r, 0, strerror(r))
36907cb5 211
68f8c317
JR
212 r = enable_event(han, ev3, channel.name)
213 self.assertGreaterEqual(r, 0, strerror(r))
36907cb5 214
68f8c317
JR
215 #Start, stop
216 r = start("test_ust_event")
217 self.assertGreaterEqual(r, 0, strerror(r))
218 time.sleep(2)
36907cb5 219
68f8c317
JR
220 r = stop("test_ust_event")
221 self.assertGreaterEqual(r, 0, strerror(r))
36907cb5 222
68f8c317
JR
223 #Restart/restop
224 r = start("test_ust_event")
225 self.assertGreaterEqual(r, 0, strerror(r))
226 time.sleep(2)
36907cb5 227
68f8c317
JR
228 r = stop("test_ust_event")
229 self.assertGreaterEqual(r, 0, strerror(r))
36907cb5 230
68f8c317
JR
231 #Disabling channel and destroy
232 r=disable_channel(han, channel.name)
233 self.assertGreaterEqual(r, 0, strerror(r))
36907cb5 234
68f8c317
JR
235 r=destroy("test_ust_event")
236 self.assertGreaterEqual(r, 0, strerror(r))
36907cb5 237
36907cb5 238
68f8c317
JR
239 def test_other_functions(self):
240 dom = Domain()
241 dom.type=DOMAIN_KERNEL
242 dom.buf_type = BUFFER_GLOBAL
36907cb5 243
68f8c317
JR
244 event=Event()
245 event.type=EVENT_TRACEPOINT
246 event.loglevel_type=EVENT_LOGLEVEL_ALL
36907cb5 247
68f8c317
JR
248 ctx = EventContext()
249 ctx.type=EVENT_CONTEXT_PID
36907cb5 250
68f8c317
JR
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
36907cb5 258
68f8c317 259 han = Handle("test_otherf" , dom)
36907cb5 260
68f8c317
JR
261 r = create("test_otherf", self.tmpdir.name)
262 self.assertGreaterEqual(r, 0, strerror(r))
36907cb5 263
68f8c317
JR
264 r = enable_event(han, event, None)
265 self.assertGreaterEqual(r, 0, strerror(r))
36907cb5 266
68f8c317
JR
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))
36907cb5 276
68f8c317
JR
277 #Def. channel attr
278 channel_set_default_attr(dom, chattr)
279 channel_set_default_attr(None, None)
36907cb5 280
68f8c317
JR
281 #Ses Daemon alive
282 r = session_daemon_alive()
283 self.assertTrue(r == 1 or r == 0, strerror(r))
36907cb5 284
68f8c317
JR
285 #Setting trace group
286 r = set_tracing_group("testing")
287 self.assertGreaterEqual(r, 0, strerror(r))
36907cb5 288
36907cb5 289
68f8c317
JR
290 r = start("test_otherf")
291 self.assertGreaterEqual(r, 0, strerror(r))
292 time.sleep(2)
36907cb5 293
68f8c317
JR
294 r = stop("test_otherf")
295 self.assertGreaterEqual(r, 0, strerror(r))
36907cb5 296
68f8c317
JR
297 domains = list_domains("test_otherf")
298 self.assertTrue(domains[0].type == DOMAIN_KERNEL)
299 self.assertTrue(domains[0].buf_type == BUFFER_GLOBAL)
36907cb5 300
68f8c317 301 del han
36907cb5 302
68f8c317
JR
303 r = destroy("test_otherf")
304 self.assertGreaterEqual(r, 0, strerror(r))
305
306
307def 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
313def 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
320if __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.080778 seconds and 4 git commands to generate.