7 class TestLttngPythonModule (unittest
.TestCase
):
10 self
.tmpdir
= tempfile
.TemporaryDirectory()
15 def test_kernel_all_events(self
):
17 dom
.type = DOMAIN_KERNEL
18 dom
.buf_type
= BUFFER_GLOBAL
21 event
.type = EVENT_TRACEPOINT
22 event
.loglevel_type
= EVENT_LOGLEVEL_ALL
24 han
= Handle("test_kernel_all_ev", dom
)
26 r
= create("test_kernel_all_ev", self
.tmpdir
.name
)
27 self
.assertGreaterEqual(r
, 0, strerror(r
))
29 r
= enable_event(han
, event
, None)
30 self
.assertGreaterEqual(r
, 0, strerror(r
))
32 r
= start("test_kernel_all_ev")
33 self
.assertGreaterEqual(r
, 0, strerror(r
))
36 r
= stop("test_kernel_all_ev")
37 self
.assertGreaterEqual(r
, 0, strerror(r
))
39 r
= destroy("test_kernel_all_ev")
40 self
.assertGreaterEqual(r
, 0, strerror(r
))
43 def test_kernel_event(self
):
46 dom
.type = DOMAIN_KERNEL
47 dom
.buf_type
= BUFFER_GLOBAL
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
58 sched_switch
= Event()
59 sched_switch
.name
= "sched_switch"
60 sched_switch
.type = EVENT_TRACEPOINT
61 sched_switch
.loglevel_type
= EVENT_LOGLEVEL_ALL
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
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
73 han
= Handle("test_kernel_event", dom
)
76 r
= create("test_kernel_event", self
.tmpdir
.name
)
77 self
.assertGreaterEqual(r
, 0, strerror(r
))
79 #Enabling channel tests
80 r
= enable_channel(han
, channel
)
81 self
.assertGreaterEqual(r
, 0, strerror(r
))
83 #Enabling events tests
84 r
= enable_event(han
, sched_switch
, channel
.name
)
85 self
.assertGreaterEqual(r
, 0, strerror(r
))
87 r
= enable_event(han
, sched_process_exit
, channel
.name
)
88 self
.assertGreaterEqual(r
, 0, strerror(r
))
90 r
= enable_event(han
, sched_process_free
, channel
.name
)
91 self
.assertGreaterEqual(r
, 0, strerror(r
))
93 #Disabling events tests
94 r
= disable_event(han
, sched_switch
.name
, channel
.name
)
95 self
.assertGreaterEqual(r
, 0, strerror(r
))
97 r
= disable_event(han
, sched_process_free
.name
, channel
.name
)
98 self
.assertGreaterEqual(r
, 0, strerror(r
))
100 #Renabling events tests
101 r
= enable_event(han
, sched_switch
, channel
.name
)
102 self
.assertGreaterEqual(r
, 0, strerror(r
))
104 r
= enable_event(han
, sched_process_free
, channel
.name
)
105 self
.assertGreaterEqual(r
, 0, strerror(r
))
107 #Start, stop, destroy
108 r
= start("test_kernel_event")
109 self
.assertGreaterEqual(r
, 0, strerror(r
))
112 r
= stop("test_kernel_event")
113 self
.assertGreaterEqual(r
, 0, strerror(r
))
115 r
=disable_channel(han
, channel
.name
)
116 self
.assertGreaterEqual(r
, 0, strerror(r
))
118 r
=destroy("test_kernel_event")
119 self
.assertGreaterEqual(r
, 0, strerror(r
))
123 def test_ust_all_events(self
):
125 dom
.type = DOMAIN_UST
126 dom
.buf_type
= BUFFER_PER_UID
129 event
.type = EVENT_TRACEPOINT
130 event
.loglevel_type
= EVENT_LOGLEVEL_ALL
132 han
= Handle("test_ust_all_ev", dom
)
134 r
= create("test_ust_all_ev", self
.tmpdir
.name
)
135 self
.assertGreaterEqual(r
, 0, strerror(r
))
137 r
= enable_event(han
, event
, None)
138 self
.assertGreaterEqual(r
, 0, strerror(r
))
140 r
= start("test_ust_all_ev")
141 self
.assertGreaterEqual(r
, 0, strerror(r
))
144 r
= stop("test_ust_all_ev")
145 self
.assertGreaterEqual(r
, 0, strerror(r
))
147 r
= destroy("test_ust_all_ev")
148 self
.assertGreaterEqual(r
, 0, strerror(r
))
151 def test_ust_event(self
):
154 dom
.type = DOMAIN_UST
155 dom
.buf_type
= BUFFER_PER_UID
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
168 ev1
.type = EVENT_TRACEPOINT
169 ev1
.loglevel_type
= EVENT_LOGLEVEL_ALL
173 ev2
.type = EVENT_TRACEPOINT
174 ev2
.loglevel_type
= EVENT_LOGLEVEL_ALL
178 ev3
.type = EVENT_TRACEPOINT
179 ev3
.loglevel_type
= EVENT_LOGLEVEL_ALL
181 han
= Handle("test_ust_event", dom
)
184 r
= create("test_ust_event", self
.tmpdir
.name
)
185 self
.assertGreaterEqual(r
, 0, strerror(r
))
187 #Enabling channel tests
188 r
= enable_channel(han
, channel
)
189 self
.assertGreaterEqual(r
, 0, strerror(r
))
191 #Enabling events tests
192 r
= enable_event(han
, ev1
, channel
.name
)
193 self
.assertGreaterEqual(r
, 0, strerror(r
))
195 r
= enable_event(han
, ev2
, channel
.name
)
196 self
.assertGreaterEqual(r
, 0, strerror(r
))
198 r
= enable_event(han
, ev3
, channel
.name
)
199 self
.assertGreaterEqual(r
, 0, strerror(r
))
201 #Disabling events tests
202 r
= disable_event(han
, ev1
.name
, channel
.name
)
203 self
.assertGreaterEqual(r
, 0, strerror(r
))
205 r
= disable_event(han
, ev3
.name
, channel
.name
)
206 self
.assertGreaterEqual(r
, 0, strerror(r
))
208 #Renabling events tests
209 r
= enable_event(han
, ev1
, channel
.name
)
210 self
.assertGreaterEqual(r
, 0, strerror(r
))
212 r
= enable_event(han
, ev3
, channel
.name
)
213 self
.assertGreaterEqual(r
, 0, strerror(r
))
216 r
= start("test_ust_event")
217 self
.assertGreaterEqual(r
, 0, strerror(r
))
220 r
= stop("test_ust_event")
221 self
.assertGreaterEqual(r
, 0, strerror(r
))
224 r
= start("test_ust_event")
225 self
.assertGreaterEqual(r
, 0, strerror(r
))
228 r
= stop("test_ust_event")
229 self
.assertGreaterEqual(r
, 0, strerror(r
))
231 #Disabling channel and destroy
232 r
=disable_channel(han
, channel
.name
)
233 self
.assertGreaterEqual(r
, 0, strerror(r
))
235 r
=destroy("test_ust_event")
236 self
.assertGreaterEqual(r
, 0, strerror(r
))
239 def test_other_functions(self
):
241 dom
.type=DOMAIN_KERNEL
242 dom
.buf_type
= BUFFER_GLOBAL
245 event
.type=EVENT_TRACEPOINT
246 event
.loglevel_type
=EVENT_LOGLEVEL_ALL
249 ctx
.type=EVENT_CONTEXT_PID
251 chattr
= ChannelAttr()
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
259 han
= Handle("test_otherf" , dom
)
261 r
= create("test_otherf", self
.tmpdir
.name
)
262 self
.assertGreaterEqual(r
, 0, strerror(r
))
264 r
= enable_event(han
, event
, None)
265 self
.assertGreaterEqual(r
, 0, strerror(r
))
268 r
= add_context(han
, ctx
, "sched_switch", "channel0")
269 self
.assertGreaterEqual(r
, 0, strerror(r
))
271 r
= add_context(han
, ctx
, "sched_wakeup", None)
272 self
.assertGreaterEqual(r
, 0, strerror(r
))
274 r
= add_context(han
, ctx
, None, None)
275 self
.assertGreaterEqual(r
, 0, strerror(r
))
278 channel_set_default_attr(dom
, chattr
)
279 channel_set_default_attr(None, None)
282 r
= session_daemon_alive()
283 self
.assertTrue(r
== 1 or r
== 0, strerror(r
))
286 r
= set_tracing_group("testing")
287 self
.assertGreaterEqual(r
, 0, strerror(r
))
290 r
= start("test_otherf")
291 self
.assertGreaterEqual(r
, 0, strerror(r
))
294 r
= stop("test_otherf")
295 self
.assertGreaterEqual(r
, 0, strerror(r
))
297 domains
= list_domains("test_otherf")
298 self
.assertTrue(domains
[0].type == DOMAIN_KERNEL
)
299 self
.assertTrue(domains
[0].buf_type
== BUFFER_GLOBAL
)
303 r
= destroy("test_otherf")
304 self
.assertGreaterEqual(r
, 0, strerror(r
))
308 suite
= unittest
.TestSuite()
309 suite
.addTest(TestLttngPythonModule("test_ust_event"))
310 suite
.addTest(TestLttngPythonModule("test_ust_all_events"))
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"))
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")
327 runner
= unittest
.TextTestRunner(verbosity
=2)
329 if os
.geteuid() == 0:
330 runner
.run(kernel_suite())
332 runner
.run(ust_suite())
This page took 0.043725 seconds and 5 git commands to generate.