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