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
  14 class TestLttngPythonModule(unittest.TestCase):
  15
  16     def setUp(self):
  17         self.tmpdir = tempfile.TemporaryDirectory()
  18
  19     def tearDown(self):
  20         self.tmpdir.cleanup()
  21
  22     def test_kernel_all_events(self):
  23         dom = Domain()
  24         dom.type = DOMAIN_KERNEL
  25         dom.buf_type = BUFFER_GLOBAL
  26
  27         event = Event()
  28         event.type = EVENT_TRACEPOINT
  29         event.loglevel_type = EVENT_LOGLEVEL_ALL
  30
  31         han = Handle("test_kernel_all_ev", dom)
  32
  33         r = create("test_kernel_all_ev", self.tmpdir.name)
  34         self.assertGreaterEqual(r, 0, strerror(r))
  35
  36         r = enable_event(han, event, None)
  37         self.assertGreaterEqual(r, 0, strerror(r))
  38
  39         r = start("test_kernel_all_ev")
  40         self.assertGreaterEqual(r, 0, strerror(r))
  41         time.sleep(2)
  42
  43         r = stop("test_kernel_all_ev")
  44         self.assertGreaterEqual(r, 0, strerror(r))
  45
  46         r = destroy("test_kernel_all_ev")
  47         self.assertGreaterEqual(r, 0, strerror(r))
  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     def test_ust_all_events(self):
 128         dom = Domain()
 129         dom.type = DOMAIN_UST
 130         dom.buf_type = BUFFER_PER_UID
 131
 132         event = Event()
 133         event.type = EVENT_TRACEPOINT
 134         event.loglevel_type = EVENT_LOGLEVEL_ALL
 135
 136         han = Handle("test_ust_all_ev", dom)
 137
 138         r = create("test_ust_all_ev", self.tmpdir.name)
 139         self.assertGreaterEqual(r, 0, strerror(r))
 140
 141         r = enable_event(han, event, None)
 142         self.assertGreaterEqual(r, 0, strerror(r))
 143
 144         r = start("test_ust_all_ev")
 145         self.assertGreaterEqual(r, 0, strerror(r))
 146         time.sleep(2)
 147
 148         r = stop("test_ust_all_ev")
 149         self.assertGreaterEqual(r, 0, strerror(r))
 150
 151         r = destroy("test_ust_all_ev")
 152         self.assertGreaterEqual(r, 0, strerror(r))
 153
 154     def test_ust_event(self):
 155
 156         dom = Domain()
 157         dom.type = DOMAIN_UST
 158         dom.buf_type = BUFFER_PER_UID
 159
 160         channel = Channel()
 161         channel.name = "mychan"
 162         channel.attr.overwrite = 0
 163         channel.attr.subbuf_size = 4096
 164         channel.attr.num_subbuf = 8
 165         channel.attr.switch_timer_interval = 0
 166         channel.attr.read_timer_interval = 200
 167         channel.attr.output = EVENT_MMAP
 168
 169         ev1 = Event()
 170         ev1.name = "tp1"
 171         ev1.type = EVENT_TRACEPOINT
 172         ev1.loglevel_type = EVENT_LOGLEVEL_ALL
 173
 174         ev2 = Event()
 175         ev2.name = "ev2"
 176         ev2.type = EVENT_TRACEPOINT
 177         ev2.loglevel_type = EVENT_LOGLEVEL_ALL
 178
 179         ev3 = Event()
 180         ev3.name = "ev3"
 181         ev3.type = EVENT_TRACEPOINT
 182         ev3.loglevel_type = EVENT_LOGLEVEL_ALL
 183
 184         han = Handle("test_ust_event", dom)
 185
 186         # Create session test
 187         r = create("test_ust_event", self.tmpdir.name)
 188         self.assertGreaterEqual(r, 0, strerror(r))
 189
 190         # Enabling channel tests
 191         r = enable_channel(han, channel)
 192         self.assertGreaterEqual(r, 0, strerror(r))
 193
 194         # Enabling events tests
 195         r = enable_event(han, ev1, channel.name)
 196         self.assertGreaterEqual(r, 0, strerror(r))
 197
 198         r = enable_event(han, ev2, channel.name)
 199         self.assertGreaterEqual(r, 0, strerror(r))
 200
 201         r = enable_event(han, ev3, channel.name)
 202         self.assertGreaterEqual(r, 0, strerror(r))
 203
 204         # Disabling events tests
 205         r = disable_event(han, ev1.name, channel.name)
 206         self.assertGreaterEqual(r, 0, strerror(r))
 207
 208         r = disable_event(han, ev3.name, channel.name)
 209         self.assertGreaterEqual(r, 0, strerror(r))
 210
 211         # Renabling events tests
 212         r = enable_event(han, ev1, channel.name)
 213         self.assertGreaterEqual(r, 0, strerror(r))
 214
 215         r = enable_event(han, ev3, channel.name)
 216         self.assertGreaterEqual(r, 0, strerror(r))
 217
 218         # Start, stop
 219         r = start("test_ust_event")
 220         self.assertGreaterEqual(r, 0, strerror(r))
 221         time.sleep(2)
 222
 223         r = stop("test_ust_event")
 224         self.assertGreaterEqual(r, 0, strerror(r))
 225
 226         # Restart/restop
 227         r = start("test_ust_event")
 228         self.assertGreaterEqual(r, 0, strerror(r))
 229         time.sleep(2)
 230
 231         r = stop("test_ust_event")
 232         self.assertGreaterEqual(r, 0, strerror(r))
 233
 234         # Disabling channel and destroy
 235         r = disable_channel(han, channel.name)
 236         self.assertGreaterEqual(r, 0, strerror(r))
 237
 238         r = destroy("test_ust_event")
 239         self.assertGreaterEqual(r, 0, strerror(r))
 240
 241     def test_other_functions(self):
 242         dom = Domain()
 243         dom.type = DOMAIN_KERNEL
 244         dom.buf_type = BUFFER_GLOBAL
 245
 246         event = Event()
 247         event.type = EVENT_TRACEPOINT
 248         event.loglevel_type = EVENT_LOGLEVEL_ALL
 249
 250         ctx = EventContext()
 251         ctx.type = EVENT_CONTEXT_PID
 252
 253         chattr = ChannelAttr()
 254         chattr.overwrite = 0
 255         chattr.subbuf_size = 4096
 256         chattr.num_subbuf = 8
 257         chattr.switch_timer_interval = 0
 258         chattr.read_timer_interval = 200
 259         chattr.output = EVENT_SPLICE
 260
 261         han = Handle("test_otherf", dom)
 262
 263         r = create("test_otherf", self.tmpdir.name)
 264         self.assertGreaterEqual(r, 0, strerror(r))
 265
 266         r = enable_event(han, event, None)
 267         self.assertGreaterEqual(r, 0, strerror(r))
 268
 269         # Context test
 270         r = add_context(han, ctx, "sched_switch", "channel0")
 271         self.assertGreaterEqual(r, 0, strerror(r))
 272         # Any channel
 273         r = add_context(han, ctx, "sched_wakeup", None)
 274         self.assertGreaterEqual(r, 0, strerror(r))
 275         # All events
 276         r = add_context(han, ctx, None, None)
 277         self.assertGreaterEqual(r, 0, strerror(r))
 278
 279         # Def. channel attr
 280         channel_set_default_attr(dom, chattr)
 281         channel_set_default_attr(None, None)
 282
 283         # Ses Daemon alive
 284         r = session_daemon_alive()
 285         self.assertTrue(r == 1 or r == 0, strerror(r))
 286
 287         # Setting trace group
 288         r = set_tracing_group("testing")
 289         self.assertGreaterEqual(r, 0, strerror(r))
 290
 291         r = start("test_otherf")
 292         self.assertGreaterEqual(r, 0, strerror(r))
 293         time.sleep(2)
 294
 295         r = stop("test_otherf")
 296         self.assertGreaterEqual(r, 0, strerror(r))
 297
 298         domains = list_domains("test_otherf")
 299         self.assertTrue(domains[0].type == DOMAIN_KERNEL)
 300         self.assertTrue(domains[0].buf_type == BUFFER_GLOBAL)
 301
 302         del han
 303
 304         r = destroy("test_otherf")
 305         self.assertGreaterEqual(r, 0, strerror(r))
 306
 307
 308 def ust_suite():
 309     suite = unittest.TestSuite()
 310     suite.addTest(TestLttngPythonModule("test_ust_event"))
 311     suite.addTest(TestLttngPythonModule("test_ust_all_events"))
 312     return suite
 313
 314
 315 def kernel_suite():
 316     suite = unittest.TestSuite()
 317     suite.addTest(TestLttngPythonModule("test_kernel_event"))
 318     suite.addTest(TestLttngPythonModule("test_kernel_all_events"))
 319     suite.addTest(TestLttngPythonModule("test_other_functions"))
 320     return suite
 321
 322
 323 if __name__ == "__main__":
 324     destroy("test_kernel_event")
 325     destroy("test_kernel_all_events")
 326     destroy("test_ust_all_events")
 327     destroy("test_ust_event")
 328     destroy("test_otherf")
 329
 330     runner = unittest.TextTestRunner(verbosity=2)
 331
 332     if os.geteuid() == 0:
 333         runner.run(kernel_suite())
 334
 335     runner.run(ust_suite())