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