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

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