From 5f2c438a1cb9591d017be2b88af13222a27410ec Mon Sep 17 00:00:00 2001
From: Jonathan Rajotte <jonathan.rajotte-julien@efficios.com>
Date: Thu, 23 May 2019 14:11:35 -0400
Subject: [PATCH lttng-tools 2/2] Update base test for binding

This test is not run for now as it is not part of the test suite.

Use a temporary directory to store trace.
Split in 2 test suite, one for ust and the other for kernel.
Partially fix formatting.

Signed-off-by: Jonathan Rajotte <jonathan.rajotte-julien@efficios.com>
---
 extras/bindings/swig/python/tests/tests.py | 459 +++++++++++----------
 1 file changed, 244 insertions(+), 215 deletions(-)

diff --git a/extras/bindings/swig/python/tests/tests.py b/extras/bindings/swig/python/tests/tests.py
index edfb5e205..fcb3789a1 100644
--- a/extras/bindings/swig/python/tests/tests.py
+++ b/extras/bindings/swig/python/tests/tests.py
@@ -1,303 +1,332 @@
 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.17.1

