X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=extras%2Fbindings%2Fswig%2Fpython%2Ftests%2Ftests.py;h=fcb3789a1aca3a237c1919c7cde7ef97aeaaf1dc;hp=edfb5e20563ec8286f876c37e1b1c5887eaa5b48;hb=68f8c317b696fb913a63298fa09ab0198995d292;hpb=dd5f427a5998f826204ac3bfc1659cd7177dc747 diff --git a/extras/bindings/swig/python/tests/tests.py b/extras/bindings/swig/python/tests/tests.py index edfb5e205..fcb3789a1 100644 --- a/extras/bindings/swig/python/tests/tests.py +++ b/extras/bindings/swig/python/tests/tests.py @@ -1,303 +1,332 @@ import unittest import os import time +import tempfile from lttng import * class TestLttngPythonModule (unittest.TestCase): - def test_kernel_all_events(self): - dom = Domain() - dom.type = DOMAIN_KERNEL + def setUp(self): + self.tmpdir = tempfile.TemporaryDirectory() - event = Event() - event.type = EVENT_TRACEPOINT - event.loglevel_type = EVENT_LOGLEVEL_ALL + def tearDown(self): + self.tmpdir.cleanup() - han = Handle("test_kernel_all_ev", dom) + def test_kernel_all_events(self): + dom = Domain() + dom.type = DOMAIN_KERNEL + dom.buf_type = BUFFER_GLOBAL - r = create("test_kernel_all_ev","/lttng-traces/test") - self.assertGreaterEqual(r, 0, strerror(r)) + event = Event() + event.type = EVENT_TRACEPOINT + event.loglevel_type = EVENT_LOGLEVEL_ALL - r = enable_event(han, event, None) - self.assertGreaterEqual(r, 0, strerror(r)) + han = Handle("test_kernel_all_ev", dom) - r = start("test_kernel_all_ev") - self.assertGreaterEqual(r, 0, strerror(r)) - time.sleep(2) + r = create("test_kernel_all_ev", self.tmpdir.name) + self.assertGreaterEqual(r, 0, strerror(r)) - r = stop("test_kernel_all_ev") - self.assertGreaterEqual(r, 0, strerror(r)) + r = enable_event(han, event, None) + self.assertGreaterEqual(r, 0, strerror(r)) - r = destroy("test_kernel_all_ev") - self.assertGreaterEqual(r, 0, strerror(r)) + r = start("test_kernel_all_ev") + self.assertGreaterEqual(r, 0, strerror(r)) + time.sleep(2) + r = stop("test_kernel_all_ev") + self.assertGreaterEqual(r, 0, strerror(r)) - def test_kernel_event(self): + r = destroy("test_kernel_all_ev") + self.assertGreaterEqual(r, 0, strerror(r)) - dom = Domain() - dom.type = DOMAIN_KERNEL - channel = Channel() - channel.name="mychan" - channel.attr.overwrite = 0 - channel.attr.subbuf_size = 4096 - channel.attr.num_subbuf = 8 - channel.attr.switch_timer_interval = 0 - channel.attr.read_timer_interval = 200 - channel.attr.output = EVENT_SPLICE + def test_kernel_event(self): - sched_switch = Event() - sched_switch.name = "sched_switch" - sched_switch.type = EVENT_TRACEPOINT - sched_switch.loglevel_type = EVENT_LOGLEVEL_ALL + dom = Domain() + dom.type = DOMAIN_KERNEL + dom.buf_type = BUFFER_GLOBAL - sched_process_exit = Event() - sched_process_exit.name = "sched_process_exit" - sched_process_exit.type = EVENT_TRACEPOINT - sched_process_exit.loglevel_type = EVENT_LOGLEVEL_ALL + channel = Channel() + channel.name="mychan" + channel.attr.overwrite = 0 + channel.attr.subbuf_size = 4096 + channel.attr.num_subbuf = 8 + channel.attr.switch_timer_interval = 0 + channel.attr.read_timer_interval = 200 + channel.attr.output = EVENT_SPLICE - sched_process_free = Event() - sched_process_free.name = "sched_process_free" - sched_process_free.type = EVENT_TRACEPOINT - sched_process_free.loglevel_type = EVENT_LOGLEVEL_ALL + sched_switch = Event() + sched_switch.name = "sched_switch" + sched_switch.type = EVENT_TRACEPOINT + sched_switch.loglevel_type = EVENT_LOGLEVEL_ALL - han = Handle("test_kernel_event", dom) + sched_process_exit = Event() + sched_process_exit.name = "sched_process_exit" + sched_process_exit.type = EVENT_TRACEPOINT + sched_process_exit.loglevel_type = EVENT_LOGLEVEL_ALL - #Create session test - r = create("test_kernel_event","/lttng-traces/test") - self.assertGreaterEqual(r, 0, strerror(r)) + sched_process_free = Event() + sched_process_free.name = "sched_process_free" + sched_process_free.type = EVENT_TRACEPOINT + sched_process_free.loglevel_type = EVENT_LOGLEVEL_ALL - #Enabling channel tests - r = enable_channel(han, channel) - self.assertGreaterEqual(r, 0, strerror(r)) + han = Handle("test_kernel_event", dom) - #Enabling events tests - r = enable_event(han, sched_switch, channel.name) - self.assertGreaterEqual(r, 0, strerror(r)) + #Create session test + r = create("test_kernel_event", self.tmpdir.name) + self.assertGreaterEqual(r, 0, strerror(r)) - r = enable_event(han, sched_process_exit, channel.name) - self.assertGreaterEqual(r, 0, strerror(r)) + #Enabling channel tests + r = enable_channel(han, channel) + self.assertGreaterEqual(r, 0, strerror(r)) - r = enable_event(han, sched_process_free, channel.name) - self.assertGreaterEqual(r, 0, strerror(r)) + #Enabling events tests + r = enable_event(han, sched_switch, channel.name) + self.assertGreaterEqual(r, 0, strerror(r)) - #Disabling events tests - r = disable_event(han, sched_switch.name, channel.name) - self.assertGreaterEqual(r, 0, strerror(r)) + r = enable_event(han, sched_process_exit, channel.name) + self.assertGreaterEqual(r, 0, strerror(r)) - r = disable_event(han, sched_process_free.name, channel.name) - self.assertGreaterEqual(r, 0, strerror(r)) + r = enable_event(han, sched_process_free, channel.name) + self.assertGreaterEqual(r, 0, strerror(r)) - #Renabling events tests - r = enable_event(han, sched_switch, channel.name) - self.assertGreaterEqual(r, 0, strerror(r)) + #Disabling events tests + r = disable_event(han, sched_switch.name, channel.name) + self.assertGreaterEqual(r, 0, strerror(r)) - r = enable_event(han, sched_process_free, channel.name) - self.assertGreaterEqual(r, 0, strerror(r)) + r = disable_event(han, sched_process_free.name, channel.name) + self.assertGreaterEqual(r, 0, strerror(r)) - #Start, stop, destroy - r = start("test_kernel_event") - self.assertGreaterEqual(r, 0, strerror(r)) - time.sleep(2) + #Renabling events tests + r = enable_event(han, sched_switch, channel.name) + self.assertGreaterEqual(r, 0, strerror(r)) - r = stop("test_kernel_event") - self.assertGreaterEqual(r, 0, strerror(r)) + r = enable_event(han, sched_process_free, channel.name) + self.assertGreaterEqual(r, 0, strerror(r)) - r=disable_channel(han, channel.name) - self.assertGreaterEqual(r, 0, strerror(r)) + #Start, stop, destroy + r = start("test_kernel_event") + self.assertGreaterEqual(r, 0, strerror(r)) + time.sleep(2) - r=destroy("test_kernel_event") - self.assertGreaterEqual(r, 0, strerror(r)) + r = stop("test_kernel_event") + self.assertGreaterEqual(r, 0, strerror(r)) + r=disable_channel(han, channel.name) + self.assertGreaterEqual(r, 0, strerror(r)) + r=destroy("test_kernel_event") + self.assertGreaterEqual(r, 0, strerror(r)) - def test_ust_all_events(self): - dom = Domain() - dom.type = DOMAIN_UST - event = Event() - event.type = EVENT_TRACEPOINT - event.loglevel_type = EVENT_LOGLEVEL_ALL - han = Handle("test_ust_all_ev", dom) + def test_ust_all_events(self): + dom = Domain() + dom.type = DOMAIN_UST + dom.buf_type = BUFFER_PER_UID - r = create("test_ust_all_ev","/lttng-traces/test") - self.assertGreaterEqual(r, 0, strerror(r)) + event = Event() + event.type = EVENT_TRACEPOINT + event.loglevel_type = EVENT_LOGLEVEL_ALL - r = enable_event(han, event, None) - self.assertGreaterEqual(r, 0, strerror(r)) + han = Handle("test_ust_all_ev", dom) - r = start("test_ust_all_ev") - self.assertGreaterEqual(r, 0, strerror(r)) - time.sleep(2) + r = create("test_ust_all_ev", self.tmpdir.name) + self.assertGreaterEqual(r, 0, strerror(r)) - r = stop("test_ust_all_ev") - self.assertGreaterEqual(r, 0, strerror(r)) + r = enable_event(han, event, None) + self.assertGreaterEqual(r, 0, strerror(r)) - r = destroy("test_ust_all_ev") - self.assertGreaterEqual(r, 0, strerror(r)) + r = start("test_ust_all_ev") + self.assertGreaterEqual(r, 0, strerror(r)) + time.sleep(2) + r = stop("test_ust_all_ev") + self.assertGreaterEqual(r, 0, strerror(r)) - def test_ust_event(self): + r = destroy("test_ust_all_ev") + self.assertGreaterEqual(r, 0, strerror(r)) - dom = Domain() - dom.type = DOMAIN_UST - channel = Channel() - channel.name="mychan" - channel.attr.overwrite = 0 - channel.attr.subbuf_size = 4096 - channel.attr.num_subbuf = 8 - channel.attr.switch_timer_interval = 0 - channel.attr.read_timer_interval = 200 - channel.attr.output = EVENT_MMAP + def test_ust_event(self): - ev1 = Event() - ev1.name = "tp1" - ev1.type = EVENT_TRACEPOINT - ev1.loglevel_type = EVENT_LOGLEVEL_ALL + dom = Domain() + dom.type = DOMAIN_UST + dom.buf_type = BUFFER_PER_UID - ev2 = Event() - ev2.name = "ev2" - ev2.type = EVENT_TRACEPOINT - ev2.loglevel_type = EVENT_LOGLEVEL_ALL + channel = Channel() + channel.name="mychan" + channel.attr.overwrite = 0 + channel.attr.subbuf_size = 4096 + channel.attr.num_subbuf = 8 + channel.attr.switch_timer_interval = 0 + channel.attr.read_timer_interval = 200 + channel.attr.output = EVENT_MMAP - ev3 = Event() - ev3.name = "ev3" - ev3.type = EVENT_TRACEPOINT - ev3.loglevel_type = EVENT_LOGLEVEL_ALL + ev1 = Event() + ev1.name = "tp1" + ev1.type = EVENT_TRACEPOINT + ev1.loglevel_type = EVENT_LOGLEVEL_ALL - han = Handle("test_ust_event", dom) + ev2 = Event() + ev2.name = "ev2" + ev2.type = EVENT_TRACEPOINT + ev2.loglevel_type = EVENT_LOGLEVEL_ALL - #Create session test - r = create("test_ust_event","/lttng-traces/test") - self.assertGreaterEqual(r, 0, strerror(r)) + ev3 = Event() + ev3.name = "ev3" + ev3.type = EVENT_TRACEPOINT + ev3.loglevel_type = EVENT_LOGLEVEL_ALL - #Enabling channel tests - r = enable_channel(han, channel) - self.assertGreaterEqual(r, 0, strerror(r)) + han = Handle("test_ust_event", dom) - #Enabling events tests - r = enable_event(han, ev1, channel.name) - self.assertGreaterEqual(r, 0, strerror(r)) + #Create session test + r = create("test_ust_event", self.tmpdir.name) + self.assertGreaterEqual(r, 0, strerror(r)) - r = enable_event(han, ev2, channel.name) - self.assertGreaterEqual(r, 0, strerror(r)) + #Enabling channel tests + r = enable_channel(han, channel) + self.assertGreaterEqual(r, 0, strerror(r)) - r = enable_event(han, ev3, channel.name) - self.assertGreaterEqual(r, 0, strerror(r)) + #Enabling events tests + r = enable_event(han, ev1, channel.name) + self.assertGreaterEqual(r, 0, strerror(r)) - #Disabling events tests - r = disable_event(han, ev1.name, channel.name) - self.assertGreaterEqual(r, 0, strerror(r)) + r = enable_event(han, ev2, channel.name) + self.assertGreaterEqual(r, 0, strerror(r)) - r = disable_event(han, ev3.name, channel.name) - self.assertGreaterEqual(r, 0, strerror(r)) + r = enable_event(han, ev3, channel.name) + self.assertGreaterEqual(r, 0, strerror(r)) - #Renabling events tests - r = enable_event(han, ev1, channel.name) - self.assertGreaterEqual(r, 0, strerror(r)) + #Disabling events tests + r = disable_event(han, ev1.name, channel.name) + self.assertGreaterEqual(r, 0, strerror(r)) - r = enable_event(han, ev3, channel.name) - self.assertGreaterEqual(r, 0, strerror(r)) + r = disable_event(han, ev3.name, channel.name) + self.assertGreaterEqual(r, 0, strerror(r)) - #Start, stop - r = start("test_ust_event") - self.assertGreaterEqual(r, 0, strerror(r)) - time.sleep(2) + #Renabling events tests + r = enable_event(han, ev1, channel.name) + self.assertGreaterEqual(r, 0, strerror(r)) - r = stop("test_ust_event") - self.assertGreaterEqual(r, 0, strerror(r)) + r = enable_event(han, ev3, channel.name) + self.assertGreaterEqual(r, 0, strerror(r)) - #Restart/restop - r = start("test_ust_event") - self.assertGreaterEqual(r, 0, strerror(r)) - time.sleep(2) + #Start, stop + r = start("test_ust_event") + self.assertGreaterEqual(r, 0, strerror(r)) + time.sleep(2) - r = stop("test_ust_event") - self.assertGreaterEqual(r, 0, strerror(r)) + r = stop("test_ust_event") + self.assertGreaterEqual(r, 0, strerror(r)) - #Disabling channel and destroy - r=disable_channel(han, channel.name) - self.assertGreaterEqual(r, 0, strerror(r)) + #Restart/restop + r = start("test_ust_event") + self.assertGreaterEqual(r, 0, strerror(r)) + time.sleep(2) - r=destroy("test_ust_event") - self.assertGreaterEqual(r, 0, strerror(r)) + r = stop("test_ust_event") + self.assertGreaterEqual(r, 0, strerror(r)) + #Disabling channel and destroy + r=disable_channel(han, channel.name) + self.assertGreaterEqual(r, 0, strerror(r)) - def test_other_functions(self): - dom = Domain() - dom.type=DOMAIN_KERNEL + r=destroy("test_ust_event") + self.assertGreaterEqual(r, 0, strerror(r)) - event=Event() - event.type=EVENT_TRACEPOINT - event.loglevel_type=EVENT_LOGLEVEL_ALL - ctx = EventContext() - ctx.type=EVENT_CONTEXT_PID + def test_other_functions(self): + dom = Domain() + dom.type=DOMAIN_KERNEL + dom.buf_type = BUFFER_GLOBAL - chattr = ChannelAttr() - chattr.overwrite = 0 - chattr.subbuf_size = 4096 - chattr.num_subbuf = 8 - chattr.switch_timer_interval = 0 - chattr.read_timer_interval = 200 - chattr.output = EVENT_SPLICE + event=Event() + event.type=EVENT_TRACEPOINT + event.loglevel_type=EVENT_LOGLEVEL_ALL - han = Handle("test_otherf" , dom) + ctx = EventContext() + ctx.type=EVENT_CONTEXT_PID - r = create("test_otherf","/lttng-traces/test") - self.assertGreaterEqual(r, 0, strerror(r)) + chattr = ChannelAttr() + chattr.overwrite = 0 + chattr.subbuf_size = 4096 + chattr.num_subbuf = 8 + chattr.switch_timer_interval = 0 + chattr.read_timer_interval = 200 + chattr.output = EVENT_SPLICE - r = enable_event(han, event, None) - self.assertGreaterEqual(r, 0, strerror(r)) + han = Handle("test_otherf" , dom) - #Context test - r = add_context(han, ctx, "sched_switch", "channel0") - self.assertGreaterEqual(r, 0, strerror(r)) - #Any channel - r = add_context(han, ctx, "sched_wakeup", None) - self.assertGreaterEqual(r, 0, strerror(r)) - #All events - r = add_context(han, ctx, None, None) - self.assertGreaterEqual(r, 0, strerror(r)) + r = create("test_otherf", self.tmpdir.name) + self.assertGreaterEqual(r, 0, strerror(r)) - #Def. channel attr - channel_set_default_attr(dom, chattr) - channel_set_default_attr(None, None) + r = enable_event(han, event, None) + self.assertGreaterEqual(r, 0, strerror(r)) - #Ses Daemon alive - r = session_daemon_alive() - self.assertTrue(r == 1 or r == 0, strerror(r)) + #Context test + r = add_context(han, ctx, "sched_switch", "channel0") + self.assertGreaterEqual(r, 0, strerror(r)) + #Any channel + r = add_context(han, ctx, "sched_wakeup", None) + self.assertGreaterEqual(r, 0, strerror(r)) + #All events + r = add_context(han, ctx, None, None) + self.assertGreaterEqual(r, 0, strerror(r)) - #Setting trace group - r = set_tracing_group("testing") - self.assertGreaterEqual(r, 0, strerror(r)) + #Def. channel attr + channel_set_default_attr(dom, chattr) + channel_set_default_attr(None, None) + #Ses Daemon alive + r = session_daemon_alive() + self.assertTrue(r == 1 or r == 0, strerror(r)) - r = start("test_otherf") - self.assertGreaterEqual(r, 0, strerror(r)) - time.sleep(2) + #Setting trace group + r = set_tracing_group("testing") + self.assertGreaterEqual(r, 0, strerror(r)) - r = stop("test_otherf") - self.assertGreaterEqual(r, 0, strerror(r)) - del han + r = start("test_otherf") + self.assertGreaterEqual(r, 0, strerror(r)) + time.sleep(2) - r = destroy("test_otherf") - self.assertGreaterEqual(r, 0, strerror(r)) + r = stop("test_otherf") + self.assertGreaterEqual(r, 0, strerror(r)) + domains = list_domains("test_otherf") + self.assertTrue(domains[0].type == DOMAIN_KERNEL) + self.assertTrue(domains[0].buf_type == BUFFER_GLOBAL) -if __name__ == "__main__": - # CHECK IF ROOT - if os.geteuid() == 0: - #Make sure session names don't already exist: - destroy("test_kernel_event") - destroy("test_kernel_all_events") - destroy("test_ust_all_events") - destroy("test_ust_event") - destroy("test_otherf") + del han - unittest.main() - else: - print('Script must be run as root') + r = destroy("test_otherf") + self.assertGreaterEqual(r, 0, strerror(r)) + + +def ust_suite(): + suite = unittest.TestSuite() + suite.addTest(TestLttngPythonModule("test_ust_event")) + suite.addTest(TestLttngPythonModule("test_ust_all_events")) + return suite + +def kernel_suite(): + suite = unittest.TestSuite() + suite.addTest(TestLttngPythonModule("test_kernel_event")) + suite.addTest(TestLttngPythonModule("test_kernel_all_events")) + suite.addTest(TestLttngPythonModule("test_other_functions")) + return suite + +if __name__ == '__main__': + destroy("test_kernel_event") + destroy("test_kernel_all_events") + destroy("test_ust_all_events") + destroy("test_ust_event") + destroy("test_otherf") + + runner = unittest.TextTestRunner(verbosity=2) + + if os.geteuid() == 0: + runner.run(kernel_suite()) + + runner.run(ust_suite())