X-Git-Url: http://git.liburcu.org/?a=blobdiff_plain;f=extras%2Fbindings%2Fswig%2Fpython%2Ftests%2Ftests.py;h=7dd1d062472ce3e72e8b31914f231ff244b06470;hb=6a871bbe6177568f748d1a9f580fa981bc75171c;hp=a4be98147f641746bb0ef810bcdf2dbf3a831f9e;hpb=36907cb5a542b8eb01d95e1990894abd45e98bc0;p=lttng-tools.git diff --git a/extras/bindings/swig/python/tests/tests.py b/extras/bindings/swig/python/tests/tests.py index a4be98147..7dd1d0624 100644 --- a/extras/bindings/swig/python/tests/tests.py +++ b/extras/bindings/swig/python/tests/tests.py @@ -1,310 +1,335 @@ +# +# 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 +class TestLttngPythonModule(unittest.TestCase): - event = Event() - event.type = EVENT_TRACEPOINT - event.loglevel_type = EVENT_LOGLEVEL_ALL + def setUp(self): + self.tmpdir = tempfile.TemporaryDirectory() - han = Handle("test_kernel_all_ev", dom) + def tearDown(self): + self.tmpdir.cleanup() - r = create("test_kernel_all_ev","/lttng-traces/test") - self.assertGreaterEqual(r, 0, strerror(r)) + def test_kernel_all_events(self): + dom = Domain() + dom.type = DOMAIN_KERNEL + dom.buf_type = BUFFER_GLOBAL - r = enable_event(han, event, None) - self.assertGreaterEqual(r, 0, strerror(r)) + event = Event() + event.type = EVENT_TRACEPOINT + event.loglevel_type = EVENT_LOGLEVEL_ALL - r = start("test_kernel_all_ev") - self.assertGreaterEqual(r, 0, strerror(r)) - time.sleep(2) + han = Handle("test_kernel_all_ev", dom) - r = stop("test_kernel_all_ev") - self.assertGreaterEqual(r, 0, strerror(r)) + r = create("test_kernel_all_ev", self.tmpdir.name) + self.assertGreaterEqual(r, 0, strerror(r)) - r = destroy("test_kernel_all_ev") - 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) - def test_kernel_event(self): + r = stop("test_kernel_all_ev") + self.assertGreaterEqual(r, 0, strerror(r)) - dom = Domain() - dom.type = DOMAIN_KERNEL + r = destroy("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 + 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 + def test_ust_all_events(self): + dom = Domain() + dom.type = DOMAIN_UST + dom.buf_type = BUFFER_PER_UID - event = Event() - event.type = EVENT_TRACEPOINT - event.loglevel_type = EVENT_LOGLEVEL_ALL + event = Event() + event.type = EVENT_TRACEPOINT + event.loglevel_type = EVENT_LOGLEVEL_ALL - han = Handle("test_ust_all_ev", dom) + han = Handle("test_ust_all_ev", dom) - r = create("test_ust_all_ev","/lttng-traces/test") - self.assertGreaterEqual(r, 0, strerror(r)) + 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)) + r = enable_event(han, event, None) + self.assertGreaterEqual(r, 0, strerror(r)) - r = start("test_ust_all_ev") - self.assertGreaterEqual(r, 0, strerror(r)) - time.sleep(2) + 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)) + r = stop("test_ust_all_ev") + self.assertGreaterEqual(r, 0, strerror(r)) - r = destroy("test_ust_all_ev") - self.assertGreaterEqual(r, 0, strerror(r)) + r = destroy("test_ust_all_ev") + self.assertGreaterEqual(r, 0, strerror(r)) + def test_ust_event(self): - def test_ust_event(self): + dom = Domain() + dom.type = DOMAIN_UST + dom.buf_type = BUFFER_PER_UID - 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 - 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 + ev1 = Event() + ev1.name = "tp1" + ev1.type = EVENT_TRACEPOINT + ev1.loglevel_type = EVENT_LOGLEVEL_ALL - 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 - ev2 = Event() - ev2.name = "ev2" - ev2.type = EVENT_TRACEPOINT - ev2.loglevel_type = EVENT_LOGLEVEL_ALL + ev3 = Event() + ev3.name = "ev3" + ev3.type = EVENT_TRACEPOINT + ev3.loglevel_type = EVENT_LOGLEVEL_ALL - ev3 = Event() - ev3.name = "ev3" - ev3.type = EVENT_TRACEPOINT - ev3.loglevel_type = EVENT_LOGLEVEL_ALL + han = Handle("test_ust_event", dom) - han = Handle("test_ust_event", dom) + # Create session test + r = create("test_ust_event", self.tmpdir.name) + self.assertGreaterEqual(r, 0, strerror(r)) - #Create session test - r = create("test_ust_event","/lttng-traces/test") - self.assertGreaterEqual(r, 0, strerror(r)) + # Enabling channel tests + r = enable_channel(han, channel) + self.assertGreaterEqual(r, 0, strerror(r)) - #Enabling channel tests - r = enable_channel(han, channel) - self.assertGreaterEqual(r, 0, strerror(r)) + # Enabling events tests + r = enable_event(han, ev1, channel.name) + self.assertGreaterEqual(r, 0, strerror(r)) - #Enabling 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, 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)) + # Disabling events tests + r = disable_event(han, ev1.name, 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 = disable_event(han, ev3.name, channel.name) + self.assertGreaterEqual(r, 0, strerror(r)) - r = disable_event(han, ev3.name, channel.name) - self.assertGreaterEqual(r, 0, strerror(r)) + # Renabling 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, 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) - #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)) + # Restart/restop + r = start("test_ust_event") + self.assertGreaterEqual(r, 0, strerror(r)) + time.sleep(2) - #Restart/restop - 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)) - #Disabling channel and destroy - r=disable_channel(han, channel.name) - self.assertGreaterEqual(r, 0, strerror(r)) + r = destroy("test_ust_event") + self.assertGreaterEqual(r, 0, strerror(r)) - r=destroy("test_ust_event") - self.assertGreaterEqual(r, 0, strerror(r)) + def test_other_functions(self): + dom = Domain() + dom.type = DOMAIN_KERNEL + dom.buf_type = BUFFER_GLOBAL + event = Event() + event.type = EVENT_TRACEPOINT + event.loglevel_type = EVENT_LOGLEVEL_ALL - def test_other_functions(self): - dom = Domain() - dom.type=DOMAIN_KERNEL + ctx = EventContext() + ctx.type = EVENT_CONTEXT_PID - event=Event() - event.type=EVENT_TRACEPOINT - event.loglevel_type=EVENT_LOGLEVEL_ALL + 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 - calib = Calibrate() - calib.type = CALIBRATE_FUNCTION + han = Handle("test_otherf", dom) - ctx = EventContext() - ctx.type=EVENT_CONTEXT_PID + r = create("test_otherf", self.tmpdir.name) + 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","/lttng-traces/test") - 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)) - #Calibrate test - r = calibrate(han , calib) - self.assertGreaterEqual(r, 0, strerror(r)) + # Setting trace group + r = set_tracing_group("testing") + 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 - r = destroy("test_otherf") - self.assertGreaterEqual(r, 0, strerror(r)) +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__": - # 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") - - unittest.main() - else: - print('Script must be run as root') + 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())