Bug #1185 » 0002-Update-base-test-for-binding.patch
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())
|
- « Previous
- 1
- 2
- Next »