2 # Copyright (C) 2012 Danny Serres <danny.serres@efficios.com>
4 # SPDX-License-Identifier: GPL-2.0-only
13 class TestLttngPythonModule (unittest
.TestCase
):
16 self
.tmpdir
= tempfile
.TemporaryDirectory()
21 def test_kernel_all_events(self
):
23 dom
.type = DOMAIN_KERNEL
24 dom
.buf_type
= BUFFER_GLOBAL
27 event
.type = EVENT_TRACEPOINT
28 event
.loglevel_type
= EVENT_LOGLEVEL_ALL
30 han
= Handle("test_kernel_all_ev", dom
)
32 r
= create("test_kernel_all_ev", self
.tmpdir
.name
)
33 self
.assertGreaterEqual(r
, 0, strerror(r
))
35 r
= enable_event(han
, event
, None)
36 self
.assertGreaterEqual(r
, 0, strerror(r
))
38 r
= start("test_kernel_all_ev")
39 self
.assertGreaterEqual(r
, 0, strerror(r
))
42 r
= stop("test_kernel_all_ev")
43 self
.assertGreaterEqual(r
, 0, strerror(r
))
45 r
= destroy("test_kernel_all_ev")
46 self
.assertGreaterEqual(r
, 0, strerror(r
))
49 def test_kernel_event(self
):
52 dom
.type = DOMAIN_KERNEL
53 dom
.buf_type
= BUFFER_GLOBAL
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
64 sched_switch
= Event()
65 sched_switch
.name
= "sched_switch"
66 sched_switch
.type = EVENT_TRACEPOINT
67 sched_switch
.loglevel_type
= EVENT_LOGLEVEL_ALL
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
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
79 han
= Handle("test_kernel_event", dom
)
82 r
= create("test_kernel_event", self
.tmpdir
.name
)
83 self
.assertGreaterEqual(r
, 0, strerror(r
))
85 #Enabling channel tests
86 r
= enable_channel(han
, channel
)
87 self
.assertGreaterEqual(r
, 0, strerror(r
))
89 #Enabling events tests
90 r
= enable_event(han
, sched_switch
, channel
.name
)
91 self
.assertGreaterEqual(r
, 0, strerror(r
))
93 r
= enable_event(han
, sched_process_exit
, channel
.name
)
94 self
.assertGreaterEqual(r
, 0, strerror(r
))
96 r
= enable_event(han
, sched_process_free
, channel
.name
)
97 self
.assertGreaterEqual(r
, 0, strerror(r
))
99 #Disabling events tests
100 r
= disable_event(han
, sched_switch
.name
, channel
.name
)
101 self
.assertGreaterEqual(r
, 0, strerror(r
))
103 r
= disable_event(han
, sched_process_free
.name
, channel
.name
)
104 self
.assertGreaterEqual(r
, 0, strerror(r
))
106 #Renabling events tests
107 r
= enable_event(han
, sched_switch
, channel
.name
)
108 self
.assertGreaterEqual(r
, 0, strerror(r
))
110 r
= enable_event(han
, sched_process_free
, channel
.name
)
111 self
.assertGreaterEqual(r
, 0, strerror(r
))
113 #Start, stop, destroy
114 r
= start("test_kernel_event")
115 self
.assertGreaterEqual(r
, 0, strerror(r
))
118 r
= stop("test_kernel_event")
119 self
.assertGreaterEqual(r
, 0, strerror(r
))
121 r
=disable_channel(han
, channel
.name
)
122 self
.assertGreaterEqual(r
, 0, strerror(r
))
124 r
=destroy("test_kernel_event")
125 self
.assertGreaterEqual(r
, 0, strerror(r
))
129 def test_ust_all_events(self
):
131 dom
.type = DOMAIN_UST
132 dom
.buf_type
= BUFFER_PER_UID
135 event
.type = EVENT_TRACEPOINT
136 event
.loglevel_type
= EVENT_LOGLEVEL_ALL
138 han
= Handle("test_ust_all_ev", dom
)
140 r
= create("test_ust_all_ev", self
.tmpdir
.name
)
141 self
.assertGreaterEqual(r
, 0, strerror(r
))
143 r
= enable_event(han
, event
, None)
144 self
.assertGreaterEqual(r
, 0, strerror(r
))
146 r
= start("test_ust_all_ev")
147 self
.assertGreaterEqual(r
, 0, strerror(r
))
150 r
= stop("test_ust_all_ev")
151 self
.assertGreaterEqual(r
, 0, strerror(r
))
153 r
= destroy("test_ust_all_ev")
154 self
.assertGreaterEqual(r
, 0, strerror(r
))
157 def test_ust_event(self
):
160 dom
.type = DOMAIN_UST
161 dom
.buf_type
= BUFFER_PER_UID
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
174 ev1
.type = EVENT_TRACEPOINT
175 ev1
.loglevel_type
= EVENT_LOGLEVEL_ALL
179 ev2
.type = EVENT_TRACEPOINT
180 ev2
.loglevel_type
= EVENT_LOGLEVEL_ALL
184 ev3
.type = EVENT_TRACEPOINT
185 ev3
.loglevel_type
= EVENT_LOGLEVEL_ALL
187 han
= Handle("test_ust_event", dom
)
190 r
= create("test_ust_event", self
.tmpdir
.name
)
191 self
.assertGreaterEqual(r
, 0, strerror(r
))
193 #Enabling channel tests
194 r
= enable_channel(han
, channel
)
195 self
.assertGreaterEqual(r
, 0, strerror(r
))
197 #Enabling events tests
198 r
= enable_event(han
, ev1
, channel
.name
)
199 self
.assertGreaterEqual(r
, 0, strerror(r
))
201 r
= enable_event(han
, ev2
, channel
.name
)
202 self
.assertGreaterEqual(r
, 0, strerror(r
))
204 r
= enable_event(han
, ev3
, channel
.name
)
205 self
.assertGreaterEqual(r
, 0, strerror(r
))
207 #Disabling events tests
208 r
= disable_event(han
, ev1
.name
, channel
.name
)
209 self
.assertGreaterEqual(r
, 0, strerror(r
))
211 r
= disable_event(han
, ev3
.name
, channel
.name
)
212 self
.assertGreaterEqual(r
, 0, strerror(r
))
214 #Renabling events tests
215 r
= enable_event(han
, ev1
, channel
.name
)
216 self
.assertGreaterEqual(r
, 0, strerror(r
))
218 r
= enable_event(han
, ev3
, channel
.name
)
219 self
.assertGreaterEqual(r
, 0, strerror(r
))
222 r
= start("test_ust_event")
223 self
.assertGreaterEqual(r
, 0, strerror(r
))
226 r
= stop("test_ust_event")
227 self
.assertGreaterEqual(r
, 0, strerror(r
))
230 r
= start("test_ust_event")
231 self
.assertGreaterEqual(r
, 0, strerror(r
))
234 r
= stop("test_ust_event")
235 self
.assertGreaterEqual(r
, 0, strerror(r
))
237 #Disabling channel and destroy
238 r
=disable_channel(han
, channel
.name
)
239 self
.assertGreaterEqual(r
, 0, strerror(r
))
241 r
=destroy("test_ust_event")
242 self
.assertGreaterEqual(r
, 0, strerror(r
))
245 def test_other_functions(self
):
247 dom
.type=DOMAIN_KERNEL
248 dom
.buf_type
= BUFFER_GLOBAL
251 event
.type=EVENT_TRACEPOINT
252 event
.loglevel_type
=EVENT_LOGLEVEL_ALL
255 ctx
.type=EVENT_CONTEXT_PID
257 chattr
= ChannelAttr()
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
265 han
= Handle("test_otherf" , dom
)
267 r
= create("test_otherf", self
.tmpdir
.name
)
268 self
.assertGreaterEqual(r
, 0, strerror(r
))
270 r
= enable_event(han
, event
, None)
271 self
.assertGreaterEqual(r
, 0, strerror(r
))
274 r
= add_context(han
, ctx
, "sched_switch", "channel0")
275 self
.assertGreaterEqual(r
, 0, strerror(r
))
277 r
= add_context(han
, ctx
, "sched_wakeup", None)
278 self
.assertGreaterEqual(r
, 0, strerror(r
))
280 r
= add_context(han
, ctx
, None, None)
281 self
.assertGreaterEqual(r
, 0, strerror(r
))
284 channel_set_default_attr(dom
, chattr
)
285 channel_set_default_attr(None, None)
288 r
= session_daemon_alive()
289 self
.assertTrue(r
== 1 or r
== 0, strerror(r
))
292 r
= set_tracing_group("testing")
293 self
.assertGreaterEqual(r
, 0, strerror(r
))
296 r
= start("test_otherf")
297 self
.assertGreaterEqual(r
, 0, strerror(r
))
300 r
= stop("test_otherf")
301 self
.assertGreaterEqual(r
, 0, strerror(r
))
303 domains
= list_domains("test_otherf")
304 self
.assertTrue(domains
[0].type == DOMAIN_KERNEL
)
305 self
.assertTrue(domains
[0].buf_type
== BUFFER_GLOBAL
)
309 r
= destroy("test_otherf")
310 self
.assertGreaterEqual(r
, 0, strerror(r
))
314 suite
= unittest
.TestSuite()
315 suite
.addTest(TestLttngPythonModule("test_ust_event"))
316 suite
.addTest(TestLttngPythonModule("test_ust_all_events"))
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"))
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")
333 runner
= unittest
.TextTestRunner(verbosity
=2)
335 if os
.geteuid() == 0:
336 runner
.run(kernel_suite())
338 runner
.run(ust_suite())
This page took 0.036118 seconds and 4 git commands to generate.