2 # Copyright (C) 2012 Danny Serres <danny.serres@efficios.com>
4 # SPDX-License-Identifier: GPL-2.0-only
14 class TestLttngPythonModule(unittest
.TestCase
):
17 self
.tmpdir
= tempfile
.TemporaryDirectory()
22 def test_kernel_all_events(self
):
24 dom
.type = DOMAIN_KERNEL
25 dom
.buf_type
= BUFFER_GLOBAL
28 event
.type = EVENT_TRACEPOINT
29 event
.loglevel_type
= EVENT_LOGLEVEL_ALL
31 han
= Handle("test_kernel_all_ev", dom
)
33 r
= create("test_kernel_all_ev", self
.tmpdir
.name
)
34 self
.assertGreaterEqual(r
, 0, strerror(r
))
36 r
= enable_event(han
, event
, None)
37 self
.assertGreaterEqual(r
, 0, strerror(r
))
39 r
= start("test_kernel_all_ev")
40 self
.assertGreaterEqual(r
, 0, strerror(r
))
43 r
= stop("test_kernel_all_ev")
44 self
.assertGreaterEqual(r
, 0, strerror(r
))
46 r
= destroy("test_kernel_all_ev")
47 self
.assertGreaterEqual(r
, 0, strerror(r
))
49 def test_kernel_event(self
):
52 dom
.type = DOMAIN_KERNEL
53 dom
.buf_type
= BUFFER_GLOBAL
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
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
))
127 def test_ust_all_events(self
):
129 dom
.type = DOMAIN_UST
130 dom
.buf_type
= BUFFER_PER_UID
133 event
.type = EVENT_TRACEPOINT
134 event
.loglevel_type
= EVENT_LOGLEVEL_ALL
136 han
= Handle("test_ust_all_ev", dom
)
138 r
= create("test_ust_all_ev", self
.tmpdir
.name
)
139 self
.assertGreaterEqual(r
, 0, strerror(r
))
141 r
= enable_event(han
, event
, None)
142 self
.assertGreaterEqual(r
, 0, strerror(r
))
144 r
= start("test_ust_all_ev")
145 self
.assertGreaterEqual(r
, 0, strerror(r
))
148 r
= stop("test_ust_all_ev")
149 self
.assertGreaterEqual(r
, 0, strerror(r
))
151 r
= destroy("test_ust_all_ev")
152 self
.assertGreaterEqual(r
, 0, strerror(r
))
154 def test_ust_event(self
):
157 dom
.type = DOMAIN_UST
158 dom
.buf_type
= BUFFER_PER_UID
161 channel
.name
= "mychan"
162 channel
.attr
.overwrite
= 0
163 channel
.attr
.subbuf_size
= 4096
164 channel
.attr
.num_subbuf
= 8
165 channel
.attr
.switch_timer_interval
= 0
166 channel
.attr
.read_timer_interval
= 200
167 channel
.attr
.output
= EVENT_MMAP
171 ev1
.type = EVENT_TRACEPOINT
172 ev1
.loglevel_type
= EVENT_LOGLEVEL_ALL
176 ev2
.type = EVENT_TRACEPOINT
177 ev2
.loglevel_type
= EVENT_LOGLEVEL_ALL
181 ev3
.type = EVENT_TRACEPOINT
182 ev3
.loglevel_type
= EVENT_LOGLEVEL_ALL
184 han
= Handle("test_ust_event", dom
)
186 # Create session test
187 r
= create("test_ust_event", self
.tmpdir
.name
)
188 self
.assertGreaterEqual(r
, 0, strerror(r
))
190 # Enabling channel tests
191 r
= enable_channel(han
, channel
)
192 self
.assertGreaterEqual(r
, 0, strerror(r
))
194 # Enabling events tests
195 r
= enable_event(han
, ev1
, channel
.name
)
196 self
.assertGreaterEqual(r
, 0, strerror(r
))
198 r
= enable_event(han
, ev2
, channel
.name
)
199 self
.assertGreaterEqual(r
, 0, strerror(r
))
201 r
= enable_event(han
, ev3
, channel
.name
)
202 self
.assertGreaterEqual(r
, 0, strerror(r
))
204 # Disabling events tests
205 r
= disable_event(han
, ev1
.name
, channel
.name
)
206 self
.assertGreaterEqual(r
, 0, strerror(r
))
208 r
= disable_event(han
, ev3
.name
, channel
.name
)
209 self
.assertGreaterEqual(r
, 0, strerror(r
))
211 # Renabling events tests
212 r
= enable_event(han
, ev1
, channel
.name
)
213 self
.assertGreaterEqual(r
, 0, strerror(r
))
215 r
= enable_event(han
, ev3
, channel
.name
)
216 self
.assertGreaterEqual(r
, 0, strerror(r
))
219 r
= start("test_ust_event")
220 self
.assertGreaterEqual(r
, 0, strerror(r
))
223 r
= stop("test_ust_event")
224 self
.assertGreaterEqual(r
, 0, strerror(r
))
227 r
= start("test_ust_event")
228 self
.assertGreaterEqual(r
, 0, strerror(r
))
231 r
= stop("test_ust_event")
232 self
.assertGreaterEqual(r
, 0, strerror(r
))
234 # Disabling channel and destroy
235 r
= disable_channel(han
, channel
.name
)
236 self
.assertGreaterEqual(r
, 0, strerror(r
))
238 r
= destroy("test_ust_event")
239 self
.assertGreaterEqual(r
, 0, strerror(r
))
241 def test_other_functions(self
):
243 dom
.type = DOMAIN_KERNEL
244 dom
.buf_type
= BUFFER_GLOBAL
247 event
.type = EVENT_TRACEPOINT
248 event
.loglevel_type
= EVENT_LOGLEVEL_ALL
251 ctx
.type = EVENT_CONTEXT_PID
253 chattr
= ChannelAttr()
255 chattr
.subbuf_size
= 4096
256 chattr
.num_subbuf
= 8
257 chattr
.switch_timer_interval
= 0
258 chattr
.read_timer_interval
= 200
259 chattr
.output
= EVENT_SPLICE
261 han
= Handle("test_otherf", dom
)
263 r
= create("test_otherf", self
.tmpdir
.name
)
264 self
.assertGreaterEqual(r
, 0, strerror(r
))
266 r
= enable_event(han
, event
, None)
267 self
.assertGreaterEqual(r
, 0, strerror(r
))
270 r
= add_context(han
, ctx
, "sched_switch", "channel0")
271 self
.assertGreaterEqual(r
, 0, strerror(r
))
273 r
= add_context(han
, ctx
, "sched_wakeup", None)
274 self
.assertGreaterEqual(r
, 0, strerror(r
))
276 r
= add_context(han
, ctx
, None, None)
277 self
.assertGreaterEqual(r
, 0, strerror(r
))
280 channel_set_default_attr(dom
, chattr
)
281 channel_set_default_attr(None, None)
284 r
= session_daemon_alive()
285 self
.assertTrue(r
== 1 or r
== 0, strerror(r
))
287 # Setting trace group
288 r
= set_tracing_group("testing")
289 self
.assertGreaterEqual(r
, 0, strerror(r
))
291 r
= start("test_otherf")
292 self
.assertGreaterEqual(r
, 0, strerror(r
))
295 r
= stop("test_otherf")
296 self
.assertGreaterEqual(r
, 0, strerror(r
))
298 domains
= list_domains("test_otherf")
299 self
.assertTrue(domains
[0].type == DOMAIN_KERNEL
)
300 self
.assertTrue(domains
[0].buf_type
== BUFFER_GLOBAL
)
304 r
= destroy("test_otherf")
305 self
.assertGreaterEqual(r
, 0, strerror(r
))
309 suite
= unittest
.TestSuite()
310 suite
.addTest(TestLttngPythonModule("test_ust_event"))
311 suite
.addTest(TestLttngPythonModule("test_ust_all_events"))
316 suite
= unittest
.TestSuite()
317 suite
.addTest(TestLttngPythonModule("test_kernel_event"))
318 suite
.addTest(TestLttngPythonModule("test_kernel_all_events"))
319 suite
.addTest(TestLttngPythonModule("test_other_functions"))
323 if __name__
== "__main__":
324 destroy("test_kernel_event")
325 destroy("test_kernel_all_events")
326 destroy("test_ust_all_events")
327 destroy("test_ust_event")
328 destroy("test_otherf")
330 runner
= unittest
.TextTestRunner(verbosity
=2)
332 if os
.geteuid() == 0:
333 runner
.run(kernel_suite())
335 runner
.run(ust_suite())
This page took 0.046325 seconds and 4 git commands to generate.