X-Git-Url: https://git.liburcu.org/?a=blobdiff_plain;f=extras%2Fbindings%2Fswig%2Fpython%2Ftests%2Ftests.py;h=fbbfdc48062e00a24a51cf5d5ee640ae4c9d519d;hb=ab5be9fa2eb5ba9600a82cd18fd3cfcbac69169a;hp=edfb5e20563ec8286f876c37e1b1c5887eaa5b48;hpb=a9885fc75d77e8a91416e83a076fa5381a954c50;p=lttng-tools.git diff --git a/extras/bindings/swig/python/tests/tests.py b/extras/bindings/swig/python/tests/tests.py index edfb5e205..fbbfdc480 100644 --- a/extras/bindings/swig/python/tests/tests.py +++ b/extras/bindings/swig/python/tests/tests.py @@ -1,303 +1,338 @@ +# +# Copyright (C) 2012 Danny Serres +# +# SPDX-License-Identifier: GPL-2.0-only +# + 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() + + def tearDown(self): + self.tmpdir.cleanup() + + def test_kernel_all_events(self): + dom = Domain() + dom.type = DOMAIN_KERNEL + dom.buf_type = BUFFER_GLOBAL + + event = Event() + event.type = EVENT_TRACEPOINT + event.loglevel_type = EVENT_LOGLEVEL_ALL + + han = Handle("test_kernel_all_ev", dom) + + r = create("test_kernel_all_ev", self.tmpdir.name) + self.assertGreaterEqual(r, 0, strerror(r)) + + r = enable_event(han, event, None) + self.assertGreaterEqual(r, 0, strerror(r)) + + r = start("test_kernel_all_ev") + self.assertGreaterEqual(r, 0, strerror(r)) + time.sleep(2) - event = Event() - event.type = EVENT_TRACEPOINT - event.loglevel_type = EVENT_LOGLEVEL_ALL + r = stop("test_kernel_all_ev") + self.assertGreaterEqual(r, 0, strerror(r)) - han = Handle("test_kernel_all_ev", dom) + r = destroy("test_kernel_all_ev") + self.assertGreaterEqual(r, 0, strerror(r)) - r = create("test_kernel_all_ev","/lttng-traces/test") - self.assertGreaterEqual(r, 0, strerror(r)) - r = enable_event(han, event, None) - self.assertGreaterEqual(r, 0, strerror(r)) + def test_kernel_event(self): - r = start("test_kernel_all_ev") - self.assertGreaterEqual(r, 0, strerror(r)) - time.sleep(2) + dom = Domain() + dom.type = DOMAIN_KERNEL + dom.buf_type = BUFFER_GLOBAL - r = stop("test_kernel_all_ev") - self.assertGreaterEqual(r, 0, strerror(r)) + 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 - r = destroy("test_kernel_all_ev") - self.assertGreaterEqual(r, 0, strerror(r)) + sched_switch = Event() + sched_switch.name = "sched_switch" + sched_switch.type = EVENT_TRACEPOINT + sched_switch.loglevel_type = EVENT_LOGLEVEL_ALL + 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 - def test_kernel_event(self): + 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 - dom = Domain() - dom.type = DOMAIN_KERNEL + han = Handle("test_kernel_event", dom) - 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 + #Create session test + r = create("test_kernel_event", self.tmpdir.name) + self.assertGreaterEqual(r, 0, strerror(r)) - sched_switch = Event() - sched_switch.name = "sched_switch" - sched_switch.type = EVENT_TRACEPOINT - sched_switch.loglevel_type = EVENT_LOGLEVEL_ALL + #Enabling channel tests + r = enable_channel(han, channel) + self.assertGreaterEqual(r, 0, strerror(r)) - 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 + #Enabling events tests + r = enable_event(han, sched_switch, channel.name) + 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 + r = enable_event(han, sched_process_exit, channel.name) + self.assertGreaterEqual(r, 0, strerror(r)) - han = Handle("test_kernel_event", dom) + r = enable_event(han, sched_process_free, channel.name) + self.assertGreaterEqual(r, 0, strerror(r)) - #Create session test - r = create("test_kernel_event","/lttng-traces/test") - self.assertGreaterEqual(r, 0, strerror(r)) + #Disabling events tests + r = disable_event(han, sched_switch.name, channel.name) + self.assertGreaterEqual(r, 0, strerror(r)) - #Enabling channel tests - r = enable_channel(han, channel) - self.assertGreaterEqual(r, 0, strerror(r)) + r = disable_event(han, sched_process_free.name, 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)) + #Renabling events tests + r = enable_event(han, sched_switch, channel.name) + self.assertGreaterEqual(r, 0, strerror(r)) - r = enable_event(han, sched_process_exit, channel.name) - self.assertGreaterEqual(r, 0, strerror(r)) + r = enable_event(han, sched_process_free, channel.name) + self.assertGreaterEqual(r, 0, strerror(r)) - r = enable_event(han, sched_process_free, 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) - #Disabling events tests - r = disable_event(han, sched_switch.name, channel.name) - self.assertGreaterEqual(r, 0, strerror(r)) + r = stop("test_kernel_event") + self.assertGreaterEqual(r, 0, strerror(r)) - r = disable_event(han, sched_process_free.name, channel.name) - self.assertGreaterEqual(r, 0, strerror(r)) + r=disable_channel(han, 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)) + r=destroy("test_kernel_event") + self.assertGreaterEqual(r, 0, strerror(r)) - r = enable_event(han, sched_process_free, 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 = stop("test_kernel_event") - self.assertGreaterEqual(r, 0, strerror(r)) + def test_ust_all_events(self): + dom = Domain() + dom.type = DOMAIN_UST + dom.buf_type = BUFFER_PER_UID - r=disable_channel(han, channel.name) - self.assertGreaterEqual(r, 0, strerror(r)) + event = Event() + event.type = EVENT_TRACEPOINT + event.loglevel_type = EVENT_LOGLEVEL_ALL - r=destroy("test_kernel_event") - self.assertGreaterEqual(r, 0, strerror(r)) + han = Handle("test_ust_all_ev", dom) + r = create("test_ust_all_ev", self.tmpdir.name) + self.assertGreaterEqual(r, 0, strerror(r)) + r = enable_event(han, event, None) + self.assertGreaterEqual(r, 0, strerror(r)) - def test_ust_all_events(self): - dom = Domain() - dom.type = DOMAIN_UST + r = start("test_ust_all_ev") + self.assertGreaterEqual(r, 0, strerror(r)) + time.sleep(2) - event = Event() - event.type = EVENT_TRACEPOINT - event.loglevel_type = EVENT_LOGLEVEL_ALL + r = stop("test_ust_all_ev") + self.assertGreaterEqual(r, 0, strerror(r)) - han = Handle("test_ust_all_ev", dom) + r = destroy("test_ust_all_ev") + self.assertGreaterEqual(r, 0, strerror(r)) - r = create("test_ust_all_ev","/lttng-traces/test") - self.assertGreaterEqual(r, 0, strerror(r)) - r = enable_event(han, event, None) - self.assertGreaterEqual(r, 0, strerror(r)) + def test_ust_event(self): - r = start("test_ust_all_ev") - self.assertGreaterEqual(r, 0, strerror(r)) - time.sleep(2) + dom = Domain() + dom.type = DOMAIN_UST + dom.buf_type = BUFFER_PER_UID - r = stop("test_ust_all_ev") - self.assertGreaterEqual(r, 0, strerror(r)) + 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 - r = destroy("test_ust_all_ev") - self.assertGreaterEqual(r, 0, strerror(r)) + ev1 = Event() + ev1.name = "tp1" + ev1.type = EVENT_TRACEPOINT + ev1.loglevel_type = EVENT_LOGLEVEL_ALL + ev2 = Event() + ev2.name = "ev2" + ev2.type = EVENT_TRACEPOINT + ev2.loglevel_type = EVENT_LOGLEVEL_ALL - def test_ust_event(self): + ev3 = Event() + ev3.name = "ev3" + ev3.type = EVENT_TRACEPOINT + ev3.loglevel_type = EVENT_LOGLEVEL_ALL - dom = Domain() - dom.type = DOMAIN_UST + han = Handle("test_ust_event", dom) - 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 + #Create session test + r = create("test_ust_event", self.tmpdir.name) + self.assertGreaterEqual(r, 0, strerror(r)) - ev1 = Event() - ev1.name = "tp1" - ev1.type = EVENT_TRACEPOINT - ev1.loglevel_type = EVENT_LOGLEVEL_ALL + #Enabling channel tests + r = enable_channel(han, channel) + self.assertGreaterEqual(r, 0, strerror(r)) - ev2 = Event() - ev2.name = "ev2" - ev2.type = EVENT_TRACEPOINT - ev2.loglevel_type = EVENT_LOGLEVEL_ALL + #Enabling events tests + r = enable_event(han, ev1, channel.name) + self.assertGreaterEqual(r, 0, strerror(r)) - ev3 = Event() - ev3.name = "ev3" - ev3.type = EVENT_TRACEPOINT - ev3.loglevel_type = EVENT_LOGLEVEL_ALL + r = enable_event(han, ev2, channel.name) + self.assertGreaterEqual(r, 0, strerror(r)) - han = Handle("test_ust_event", dom) + r = enable_event(han, ev3, channel.name) + self.assertGreaterEqual(r, 0, strerror(r)) - #Create session test - r = create("test_ust_event","/lttng-traces/test") - self.assertGreaterEqual(r, 0, strerror(r)) + #Disabling events tests + r = disable_event(han, ev1.name, channel.name) + self.assertGreaterEqual(r, 0, strerror(r)) - #Enabling channel tests - r = enable_channel(han, channel) - self.assertGreaterEqual(r, 0, strerror(r)) + r = disable_event(han, ev3.name, channel.name) + self.assertGreaterEqual(r, 0, strerror(r)) - #Enabling events tests - r = enable_event(han, ev1, channel.name) - self.assertGreaterEqual(r, 0, strerror(r)) + #Renabling events tests + r = enable_event(han, ev1, channel.name) + self.assertGreaterEqual(r, 0, strerror(r)) - r = enable_event(han, ev2, channel.name) - self.assertGreaterEqual(r, 0, strerror(r)) + r = enable_event(han, ev3, channel.name) + self.assertGreaterEqual(r, 0, strerror(r)) - r = enable_event(han, ev3, channel.name) - self.assertGreaterEqual(r, 0, strerror(r)) + #Start, stop + r = start("test_ust_event") + self.assertGreaterEqual(r, 0, strerror(r)) + time.sleep(2) - #Disabling events tests - r = disable_event(han, ev1.name, channel.name) - self.assertGreaterEqual(r, 0, strerror(r)) + r = stop("test_ust_event") + self.assertGreaterEqual(r, 0, strerror(r)) - r = disable_event(han, ev3.name, channel.name) - self.assertGreaterEqual(r, 0, strerror(r)) + #Restart/restop + 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)) + #Disabling channel and destroy + r=disable_channel(han, channel.name) + self.assertGreaterEqual(r, 0, strerror(r)) - #Start, stop - 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)) - #Restart/restop - r = start("test_ust_event") - self.assertGreaterEqual(r, 0, strerror(r)) - time.sleep(2) + def test_other_functions(self): + dom = Domain() + dom.type=DOMAIN_KERNEL + dom.buf_type = BUFFER_GLOBAL - r = stop("test_ust_event") - self.assertGreaterEqual(r, 0, strerror(r)) + event=Event() + event.type=EVENT_TRACEPOINT + event.loglevel_type=EVENT_LOGLEVEL_ALL - #Disabling channel and destroy - r=disable_channel(han, channel.name) - self.assertGreaterEqual(r, 0, strerror(r)) + ctx = EventContext() + ctx.type=EVENT_CONTEXT_PID - r=destroy("test_ust_event") - 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 + han = Handle("test_otherf" , dom) - def test_other_functions(self): - dom = Domain() - dom.type=DOMAIN_KERNEL + r = create("test_otherf", self.tmpdir.name) + 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)) - ctx = EventContext() - ctx.type=EVENT_CONTEXT_PID + #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)) - 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 + #Def. channel attr + channel_set_default_attr(dom, chattr) + channel_set_default_attr(None, None) - han = Handle("test_otherf" , dom) + #Ses Daemon alive + r = session_daemon_alive() + self.assertTrue(r == 1 or r == 0, strerror(r)) - r = create("test_otherf","/lttng-traces/test") - self.assertGreaterEqual(r, 0, strerror(r)) + #Setting trace group + r = set_tracing_group("testing") + self.assertGreaterEqual(r, 0, strerror(r)) - r = enable_event(han, event, None) - self.assertGreaterEqual(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)) + r = start("test_otherf") + self.assertGreaterEqual(r, 0, strerror(r)) + time.sleep(2) - #Def. channel attr - channel_set_default_attr(dom, chattr) - channel_set_default_attr(None, None) + r = stop("test_otherf") + self.assertGreaterEqual(r, 0, strerror(r)) - #Ses Daemon alive - r = session_daemon_alive() - self.assertTrue(r == 1 or r == 0, strerror(r)) + domains = list_domains("test_otherf") + self.assertTrue(domains[0].type == DOMAIN_KERNEL) + self.assertTrue(domains[0].buf_type == BUFFER_GLOBAL) - #Setting trace group - r = set_tracing_group("testing") - self.assertGreaterEqual(r, 0, strerror(r)) + del han + r = destroy("test_otherf") + self.assertGreaterEqual(r, 0, strerror(r)) - r = start("test_otherf") - self.assertGreaterEqual(r, 0, strerror(r)) - time.sleep(2) - r = stop("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 - del han +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 - r = destroy("test_otherf") - self.assertGreaterEqual(r, 0, strerror(r)) +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 __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") + if os.geteuid() == 0: + runner.run(kernel_suite()) - unittest.main() - else: - print('Script must be run as root') + runner.run(ust_suite())