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 # SPDX-License-Identifier: GPL-2.0-only
   6 #
   7
   8 import uuid
   9 import os
  10 import subprocess
  11 import shutil
  12 import sys
  13 import tempfile
  14
  15 # Import lttng bindings generated in the current tree
  16 lttng_bindings_path = os.path.dirname(os.path.abspath(__file__)) + "/"
  17 for i in range(3):
  18     lttng_bindings_path = os.path.dirname(lttng_bindings_path)
  19 lttng_bindings_path = lttng_bindings_path + "/extras/bindings/swig/python"
  20 lttng_bindings_libs_path = lttng_bindings_path + "/.libs"
  21 sys.path.append(lttng_bindings_path)
  22 sys.path.append(lttng_bindings_libs_path)
  23 from lttng import *
  24
  25
  26 class SessionInfo:
  27     def __init__(self, handle, session_name, tmp_directory, channel_name):
  28         self.handle = handle
  29         self.name = session_name
  30         self.tmp_directory = tmp_directory
  31         self.trace_path = tmp_directory + "/" + session_name
  32         self.channel_name = channel_name
  33
  34 def bail(diag, session_info = None):
  35     print("Bail out!")
  36     print("#", diag)
  37
  38     if session_info is not None:
  39         stop_session(session_info, True)
  40
  41         if os.path.exists(session_info.tmp_directory):
  42             shutil.rmtree(session_info.tmp_directory)
  43     exit(-1)
  44
  45 def print_test_result(result, number, description):
  46     result_string = None
  47     if result is True:
  48         result_string = "ok"
  49     else:
  50         result_string = "not ok"
  51
  52     result_string += " {0} - {1}".format(number, description)
  53     print(result_string)
  54
  55 def skip_test(number, description):
  56     print('ok {} # skip {}'.format(number, description))
  57
  58 def enable_ust_tracepoint_event(session_info, event_name):
  59     event = Event()
  60     event.name = event_name
  61     event.type = EVENT_TRACEPOINT
  62     event.loglevel = EVENT_LOGLEVEL_ALL
  63     res = enable_event(session_info.handle, event, session_info.channel_name)
  64     if res < 0:
  65         bail("Failed to enable userspace event " + event_name, session_info)
  66
  67 def create_session():
  68     dom = Domain()
  69     dom.type = DOMAIN_UST
  70
  71     session_name = str(uuid.uuid1())
  72     tmp_directory = tempfile.mkdtemp()
  73     trace_path = tmp_directory + "/" + session_name
  74
  75     res = create(session_name, trace_path)
  76     if res < 0:
  77         bail("Failed to create tracing session.")
  78
  79     channel = Channel()
  80     channel.name = "channel0"
  81     channel_set_default_attr(dom, channel.attr)
  82
  83     han = Handle(session_name, dom)
  84     res = enable_channel(han, channel)
  85
  86     session_info = SessionInfo(han, session_name, tmp_directory, channel.name)
  87     if res < 0:
  88         bail("Failed to enable channel " + channel.name, session_info)
  89     return session_info
  90
  91 def start_session(session_info):
  92     start(session_info.name)
  93
  94 def stop_session(session_info, bailing = False):
  95     # Workaround lttng-ctl outputing directly to stdout by spawning a subprocess.
  96     lttng_binary_path = os.path.dirname(os.path.abspath(__file__)) + "/"
  97     for i in range(3):
  98         lttng_binary_path = os.path.dirname(lttng_binary_path)
  99     lttng_binary_path = lttng_binary_path + "/src/bin/lttng/lttng"
 100
 101     retcode = subprocess.call([lttng_binary_path, "stop", session_info.name], stdout=subprocess.PIPE, stderr=subprocess.PIPE)
 102     if retcode != 0 and not bailing:
 103         bail("Unable to stop session " + session_info.name, session_info)
 104     destroy(session_info.name)