Project

General

Profile

Bug #1185 » 0002-Update-base-test-for-binding.patch

Jonathan Rajotte Julien, 05/23/2019 04:42 PM

View differences:

extras/bindings/swig/python/tests/tests.py
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())
(2-2/2)