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