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