tests/utils/test_utils.py

   1 #!/usr/bin/env python3
   2 #
   3 # Copyright (C) - 2013 Jérémie Galarneau <jeremie.galarneau@efficios.com>
   4 #
   5 # This program is free software; you can redistribute it and/or modify it
   6 # under the terms of the GNU General Public License, version 2 only, as
   7 # published by the Free Software Foundation.
   8 #
   9 # This program is distributed in the hope that it will be useful, but WITHOUT
  10 # ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
  11 # FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License for
  12 # more details.
  13 #
  14 # You should have received a copy of the GNU General Public License along with
  15 # this program; if not, write to the Free Software Foundation, Inc., 51
  16 # Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
  17
  18 import uuid
  19 import os
  20 import subprocess
  21 import shutil
  22 import sys
  23 import tempfile
  24
  25 # Import lttng bindings generated in the current tree
  26 lttng_bindings_path = os.path.dirname(os.path.abspath(__file__)) + "/"
  27 for i in range(3):
  28     lttng_bindings_path = os.path.dirname(lttng_bindings_path)
  29 lttng_bindings_path = lttng_bindings_path + "/extras/bindings/swig/python"
  30 lttng_bindings_libs_path = lttng_bindings_path + "/.libs"
  31 sys.path.append(lttng_bindings_path)
  32 sys.path.append(lttng_bindings_libs_path)
  33 from lttng import *
  34
  35
  36 class SessionInfo:
  37     def __init__(self, handle, session_name, tmp_directory, channel_name):
  38         self.handle = handle
  39         self.name = session_name
  40         self.tmp_directory = tmp_directory
  41         self.trace_path = tmp_directory + "/" + session_name
  42         self.channel_name = channel_name
  43
  44 def bail(diag, session_info = None):
  45     print("Bail out!")
  46     print("#", diag)
  47
  48     if session_info is not None:
  49         stop_session(session_info, True)
  50
  51         if os.path.exists(session_info.tmp_directory):
  52             shutil.rmtree(session_info.tmp_directory)
  53     exit(-1)
  54
  55 def print_test_result(result, number, description):
  56     result_string = None
  57     if result is True:
  58         result_string = "ok"
  59     else:
  60         result_string = "not ok"
  61
  62     result_string += " {0} - {1}".format(number, description)
  63     print(result_string)
  64
  65 def skip_test(number, description):
  66     print('ok {} # skip {}'.format(number, description))
  67
  68 def enable_ust_tracepoint_event(session_info, event_name):
  69     event = Event()
  70     event.name = event_name
  71     event.type = EVENT_TRACEPOINT
  72     event.loglevel = EVENT_LOGLEVEL_ALL
  73     res = enable_event(session_info.handle, event, session_info.channel_name)
  74     if res < 0:
  75         bail("Failed to enable userspace event " + event_name, session_info)
  76
  77 def create_session():
  78     dom = Domain()
  79     dom.type = DOMAIN_UST
  80
  81     session_name = str(uuid.uuid1())
  82     tmp_directory = tempfile.mkdtemp()
  83     trace_path = tmp_directory + "/" + session_name
  84
  85     res = create(session_name, trace_path)
  86     if res < 0:
  87         bail("Failed to create tracing session.")
  88
  89     channel = Channel()
  90     channel.name = "channel0"
  91     channel_set_default_attr(dom, channel.attr)
  92
  93     han = Handle(session_name, dom)
  94     res = enable_channel(han, channel)
  95
  96     session_info = SessionInfo(han, session_name, tmp_directory, channel.name)
  97     if res < 0:
  98         bail("Failed to enable channel " + channel.name, session_info)
  99     return session_info
 100
 101 def start_session(session_info):
 102     start(session_info.name)
 103
 104 def stop_session(session_info, bailing = False):
 105     # Workaround lttng-ctl outputing directly to stdout by spawning a subprocess.
 106     lttng_binary_path = os.path.dirname(os.path.abspath(__file__)) + "/"
 107     for i in range(3):
 108         lttng_binary_path = os.path.dirname(lttng_binary_path)
 109     lttng_binary_path = lttng_binary_path + "/src/bin/lttng/lttng"
 110
 111     retcode = subprocess.call([lttng_binary_path, "stop", session_info.name], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
 112     if retcode != 0 and not bailing:
 113         bail("Unable to stop session " + session_info.name, session_info)
 114     destroy(session_info.name)