X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=extras%2Fbindings%2Fswig%2Fpython%2Ftests%2Ftests.py;fp=extras%2Fbindings%2Fswig%2Fpython%2Ftests%2Ftests.py;h=a4be98147f641746bb0ef810bcdf2dbf3a831f9e;hp=0000000000000000000000000000000000000000;hb=36907cb5a542b8eb01d95e1990894abd45e98bc0;hpb=5168918c35a95e9d955fa4efb6533aaafe05b2e5

diff --git a/extras/bindings/swig/python/tests/tests.py b/extras/bindings/swig/python/tests/tests.py
new file mode 100644
index 000000000..a4be98147
--- /dev/null
+++ b/extras/bindings/swig/python/tests/tests.py
@@ -0,0 +1,310 @@
+import unittest
+import os
+import time
+from lttng import *
+
+class TestLttngPythonModule (unittest.TestCase):
+
+	def test_kernel_all_events(self):
+		dom = Domain()
+		dom.type = DOMAIN_KERNEL
+
+		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","/lttng-traces/test")
+		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)
+
+		r = stop("test_kernel_all_ev")
+		self.assertGreaterEqual(r, 0, strerror(r))
+
+		r = destroy("test_kernel_all_ev")
+		self.assertGreaterEqual(r, 0, strerror(r))
+
+
+	def test_kernel_event(self):
+
+		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
+
+		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
+
+		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
+
+		han = Handle("test_kernel_event", dom)
+
+		#Create session test
+		r = create("test_kernel_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 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))
+
+		#Disabling events tests
+		r = disable_event(han, sched_switch.name, channel.name)
+		self.assertGreaterEqual(r, 0, strerror(r))
+
+		r = disable_event(han, sched_process_free.name, 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_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))
+
+		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)
+
+		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))
+
+		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 = destroy("test_ust_all_ev")
+		self.assertGreaterEqual(r, 0, strerror(r))
+
+
+	def test_ust_event(self):
+
+		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
+
+		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
+
+		ev3 = Event()
+		ev3.name = "ev3"
+		ev3.type = EVENT_TRACEPOINT
+		ev3.loglevel_type = EVENT_LOGLEVEL_ALL
+
+		han = Handle("test_ust_event", dom)
+
+		#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 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))
+
+		#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))
+
+		#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))
+
+		#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))
+
+		#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))
+
+		#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))
+
+
+	def test_other_functions(self):
+		dom = Domain()
+		dom.type=DOMAIN_KERNEL
+
+		event=Event()
+		event.type=EVENT_TRACEPOINT
+		event.loglevel_type=EVENT_LOGLEVEL_ALL
+
+		calib = Calibrate()
+		calib.type = CALIBRATE_FUNCTION
+
+		ctx = EventContext()
+		ctx.type=EVENT_CONTEXT_PID
+
+		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)
+
+		r = create("test_otherf","/lttng-traces/test")
+		self.assertGreaterEqual(r, 0, strerror(r))
+
+		r = enable_event(han, event, None)
+		self.assertGreaterEqual(r, 0, strerror(r))
+
+		#Calibrate test
+		r = calibrate(han , calib)
+		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))
+
+		#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))
+
+		#Setting trace group
+		r = set_tracing_group("testing")
+		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))
+
+		del han
+
+		r = destroy("test_otherf")
+		self.assertGreaterEqual(r, 0, strerror(r))
+
+
+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')