extras/bindings/swig/python/tests/tests.py

   1 #
   2 # Copyright (C) 2012 Danny Serres <danny.serres@efficios.com>
   3 #
   4 # SPDX-License-Identifier: GPL-2.0-only
   5 #
   6
   7 import unittest
   8 import os
   9 import time
  10 import tempfile
  11 from lttng import *
  12
  13 class TestLttngPythonModule (unittest.TestCase):
  14
  15     def setUp(self):
  16         self.tmpdir = tempfile.TemporaryDirectory()
  17
  18     def tearDown(self):
  19         self.tmpdir.cleanup()
  20
  21     def test_kernel_all_events(self):
  22         dom = Domain()
  23         dom.type = DOMAIN_KERNEL
  24         dom.buf_type = BUFFER_GLOBAL
  25
  26         event = Event()
  27         event.type = EVENT_TRACEPOINT
  28         event.loglevel_type = EVENT_LOGLEVEL_ALL
  29
  30         han = Handle("test_kernel_all_ev", dom)
  31
  32         r = create("test_kernel_all_ev", self.tmpdir.name)
  33         self.assertGreaterEqual(r, 0, strerror(r))
  34
  35         r = enable_event(han, event, None)
  36         self.assertGreaterEqual(r, 0, strerror(r))
  37
  38         r = start("test_kernel_all_ev")
  39         self.assertGreaterEqual(r, 0, strerror(r))
  40         time.sleep(2)
  41
  42         r = stop("test_kernel_all_ev")
  43         self.assertGreaterEqual(r, 0, strerror(r))
  44
  45         r = destroy("test_kernel_all_ev")
  46         self.assertGreaterEqual(r, 0, strerror(r))
  47
  48
  49     def test_kernel_event(self):
  50
  51         dom = Domain()
  52         dom.type = DOMAIN_KERNEL
  53         dom.buf_type = BUFFER_GLOBAL
  54
  55         channel = Channel()
  56         channel.name="mychan"
  57         channel.attr.overwrite = 0
  58         channel.attr.subbuf_size = 4096
  59         channel.attr.num_subbuf = 8
  60         channel.attr.switch_timer_interval = 0
  61         channel.attr.read_timer_interval = 200
  62         channel.attr.output = EVENT_SPLICE
  63
  64         sched_switch = Event()
  65         sched_switch.name = "sched_switch"
  66         sched_switch.type = EVENT_TRACEPOINT
  67         sched_switch.loglevel_type = EVENT_LOGLEVEL_ALL
  68
  69         sched_process_exit = Event()
  70         sched_process_exit.name = "sched_process_exit"
  71         sched_process_exit.type = EVENT_TRACEPOINT
  72         sched_process_exit.loglevel_type = EVENT_LOGLEVEL_ALL
  73
  74         sched_process_free = Event()
  75         sched_process_free.name = "sched_process_free"
  76         sched_process_free.type = EVENT_TRACEPOINT
  77         sched_process_free.loglevel_type = EVENT_LOGLEVEL_ALL
  78
  79         han = Handle("test_kernel_event", dom)
  80
  81         #Create session test
  82         r = create("test_kernel_event", self.tmpdir.name)
  83         self.assertGreaterEqual(r, 0, strerror(r))
  84
  85         #Enabling channel tests
  86         r = enable_channel(han, channel)
  87         self.assertGreaterEqual(r, 0, strerror(r))
  88
  89         #Enabling events tests
  90         r = enable_event(han, sched_switch, channel.name)
  91         self.assertGreaterEqual(r, 0, strerror(r))
  92
  93         r = enable_event(han, sched_process_exit, channel.name)
  94         self.assertGreaterEqual(r, 0, strerror(r))
  95
  96         r = enable_event(han, sched_process_free, channel.name)
  97         self.assertGreaterEqual(r, 0, strerror(r))
  98
  99         #Disabling events tests
 100         r = disable_event(han, sched_switch.name, channel.name)
 101         self.assertGreaterEqual(r, 0, strerror(r))
 102
 103         r = disable_event(han, sched_process_free.name, channel.name)
 104         self.assertGreaterEqual(r, 0, strerror(r))
 105
 106         #Renabling events tests
 107         r = enable_event(han, sched_switch, channel.name)
 108         self.assertGreaterEqual(r, 0, strerror(r))
 109
 110         r = enable_event(han, sched_process_free, channel.name)
 111         self.assertGreaterEqual(r, 0, strerror(r))
 112
 113         #Start, stop, destroy
 114         r = start("test_kernel_event")
 115         self.assertGreaterEqual(r, 0, strerror(r))
 116         time.sleep(2)
 117
 118         r = stop("test_kernel_event")
 119         self.assertGreaterEqual(r, 0, strerror(r))
 120
 121         r=disable_channel(han, channel.name)
 122         self.assertGreaterEqual(r, 0, strerror(r))
 123
 124         r=destroy("test_kernel_event")
 125         self.assertGreaterEqual(r, 0, strerror(r))
 126
 127
 128
 129     def test_ust_all_events(self):
 130         dom = Domain()
 131         dom.type = DOMAIN_UST
 132         dom.buf_type = BUFFER_PER_UID
 133
 134         event = Event()
 135         event.type = EVENT_TRACEPOINT
 136         event.loglevel_type = EVENT_LOGLEVEL_ALL
 137
 138         han = Handle("test_ust_all_ev", dom)
 139
 140         r = create("test_ust_all_ev", self.tmpdir.name)
 141         self.assertGreaterEqual(r, 0, strerror(r))
 142
 143         r = enable_event(han, event, None)
 144         self.assertGreaterEqual(r, 0, strerror(r))
 145
 146         r = start("test_ust_all_ev")
 147         self.assertGreaterEqual(r, 0, strerror(r))
 148         time.sleep(2)
 149
 150         r = stop("test_ust_all_ev")
 151         self.assertGreaterEqual(r, 0, strerror(r))
 152
 153         r = destroy("test_ust_all_ev")
 154         self.assertGreaterEqual(r, 0, strerror(r))
 155
 156
 157     def test_ust_event(self):
 158
 159         dom = Domain()
 160         dom.type = DOMAIN_UST
 161         dom.buf_type = BUFFER_PER_UID
 162
 163         channel = Channel()
 164         channel.name="mychan"
 165         channel.attr.overwrite = 0
 166         channel.attr.subbuf_size = 4096
 167         channel.attr.num_subbuf = 8
 168         channel.attr.switch_timer_interval = 0
 169         channel.attr.read_timer_interval = 200
 170         channel.attr.output = EVENT_MMAP
 171
 172         ev1 = Event()
 173         ev1.name = "tp1"
 174         ev1.type = EVENT_TRACEPOINT
 175         ev1.loglevel_type = EVENT_LOGLEVEL_ALL
 176
 177         ev2 = Event()
 178         ev2.name = "ev2"
 179         ev2.type = EVENT_TRACEPOINT
 180         ev2.loglevel_type = EVENT_LOGLEVEL_ALL
 181
 182         ev3 = Event()
 183         ev3.name = "ev3"
 184         ev3.type = EVENT_TRACEPOINT
 185         ev3.loglevel_type = EVENT_LOGLEVEL_ALL
 186
 187         han = Handle("test_ust_event", dom)
 188
 189         #Create session test
 190         r = create("test_ust_event", self.tmpdir.name)
 191         self.assertGreaterEqual(r, 0, strerror(r))
 192
 193         #Enabling channel tests
 194         r = enable_channel(han, channel)
 195         self.assertGreaterEqual(r, 0, strerror(r))
 196
 197         #Enabling events tests
 198         r = enable_event(han, ev1, channel.name)
 199         self.assertGreaterEqual(r, 0, strerror(r))
 200
 201         r = enable_event(han, ev2, channel.name)
 202         self.assertGreaterEqual(r, 0, strerror(r))
 203
 204         r = enable_event(han, ev3, channel.name)
 205         self.assertGreaterEqual(r, 0, strerror(r))
 206
 207         #Disabling events tests
 208         r = disable_event(han, ev1.name, channel.name)
 209         self.assertGreaterEqual(r, 0, strerror(r))
 210
 211         r = disable_event(han, ev3.name, channel.name)
 212         self.assertGreaterEqual(r, 0, strerror(r))
 213
 214         #Renabling events tests
 215         r = enable_event(han, ev1, channel.name)
 216         self.assertGreaterEqual(r, 0, strerror(r))
 217
 218         r = enable_event(han, ev3, channel.name)
 219         self.assertGreaterEqual(r, 0, strerror(r))
 220
 221         #Start, stop
 222         r = start("test_ust_event")
 223         self.assertGreaterEqual(r, 0, strerror(r))
 224         time.sleep(2)
 225
 226         r = stop("test_ust_event")
 227         self.assertGreaterEqual(r, 0, strerror(r))
 228
 229         #Restart/restop
 230         r = start("test_ust_event")
 231         self.assertGreaterEqual(r, 0, strerror(r))
 232         time.sleep(2)
 233
 234         r = stop("test_ust_event")
 235         self.assertGreaterEqual(r, 0, strerror(r))
 236
 237         #Disabling channel and destroy
 238         r=disable_channel(han, channel.name)
 239         self.assertGreaterEqual(r, 0, strerror(r))
 240
 241         r=destroy("test_ust_event")
 242         self.assertGreaterEqual(r, 0, strerror(r))
 243
 244
 245     def test_other_functions(self):
 246         dom = Domain()
 247         dom.type=DOMAIN_KERNEL
 248         dom.buf_type = BUFFER_GLOBAL
 249
 250         event=Event()
 251         event.type=EVENT_TRACEPOINT
 252         event.loglevel_type=EVENT_LOGLEVEL_ALL
 253
 254         ctx = EventContext()
 255         ctx.type=EVENT_CONTEXT_PID
 256
 257         chattr = ChannelAttr()
 258         chattr.overwrite = 0
 259         chattr.subbuf_size = 4096
 260         chattr.num_subbuf = 8
 261         chattr.switch_timer_interval = 0
 262         chattr.read_timer_interval = 200
 263         chattr.output = EVENT_SPLICE
 264
 265         han = Handle("test_otherf" , dom)
 266
 267         r = create("test_otherf", self.tmpdir.name)
 268         self.assertGreaterEqual(r, 0, strerror(r))
 269
 270         r = enable_event(han, event, None)
 271         self.assertGreaterEqual(r, 0, strerror(r))
 272
 273         #Context test
 274         r = add_context(han, ctx, "sched_switch", "channel0")
 275         self.assertGreaterEqual(r, 0, strerror(r))
 276         #Any channel
 277         r = add_context(han, ctx, "sched_wakeup", None)
 278         self.assertGreaterEqual(r, 0, strerror(r))
 279         #All events
 280         r = add_context(han, ctx, None, None)
 281         self.assertGreaterEqual(r, 0, strerror(r))
 282
 283         #Def. channel attr
 284         channel_set_default_attr(dom, chattr)
 285         channel_set_default_attr(None, None)
 286
 287         #Ses Daemon alive
 288         r = session_daemon_alive()
 289         self.assertTrue(r == 1 or r == 0, strerror(r))
 290
 291         #Setting trace group
 292         r = set_tracing_group("testing")
 293         self.assertGreaterEqual(r, 0, strerror(r))
 294
 295
 296         r = start("test_otherf")
 297         self.assertGreaterEqual(r, 0, strerror(r))
 298         time.sleep(2)
 299
 300         r = stop("test_otherf")
 301         self.assertGreaterEqual(r, 0, strerror(r))
 302
 303         domains = list_domains("test_otherf")
 304         self.assertTrue(domains[0].type == DOMAIN_KERNEL)
 305         self.assertTrue(domains[0].buf_type == BUFFER_GLOBAL)
 306
 307         del han
 308
 309         r = destroy("test_otherf")
 310         self.assertGreaterEqual(r, 0, strerror(r))
 311
 312
 313 def ust_suite():
 314     suite = unittest.TestSuite()
 315     suite.addTest(TestLttngPythonModule("test_ust_event"))
 316     suite.addTest(TestLttngPythonModule("test_ust_all_events"))
 317     return suite
 318
 319 def kernel_suite():
 320     suite = unittest.TestSuite()
 321     suite.addTest(TestLttngPythonModule("test_kernel_event"))
 322     suite.addTest(TestLttngPythonModule("test_kernel_all_events"))
 323     suite.addTest(TestLttngPythonModule("test_other_functions"))
 324     return suite
 325
 326 if __name__ == '__main__':
 327     destroy("test_kernel_event")
 328     destroy("test_kernel_all_events")
 329     destroy("test_ust_all_events")
 330     destroy("test_ust_event")
 331     destroy("test_otherf")
 332
 333     runner = unittest.TextTestRunner(verbosity=2)
 334
 335     if os.geteuid() == 0:
 336         runner.run(kernel_suite())
 337
 338     runner.run(ust_suite())