3 # Copyright (C) 2013 Jérémie Galarneau <jeremie.galarneau@efficios.com>
5 # SPDX-License-Identifier: GPL-2.0-only
15 # Import lttng bindings generated in the current tree
16 lttng_bindings_path
= os
.path
.dirname(os
.path
.abspath(__file__
)) + "/"
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
)
25 BABELTRACE_BIN
="babeltrace2"
26 trace_format_mapping
= {
27 "ctf1" : lttng_trace_format_ctf_1_descriptor_create
,
28 "ctf2" : lttng_trace_format_ctf_2_descriptor_create
,
32 def __init__(self
, handle
, session_name
, tmp_directory
, channel_name
):
34 self
.name
= session_name
35 self
.tmp_directory
= tmp_directory
36 self
.trace_path
= tmp_directory
+ "/" + session_name
37 self
.channel_name
= channel_name
39 def bail(diag
, session_info
= None):
43 if session_info
is not None:
44 stop_session(session_info
, True)
46 if os
.path
.exists(session_info
.tmp_directory
):
47 shutil
.rmtree(session_info
.tmp_directory
)
50 def print_test_result(result
, number
, description
):
55 result_string
= "not ok"
57 result_string
+= " {0} - {1}".format(number
, description
)
60 def skip_test(number
, description
):
61 print('ok {} # skip {}'.format(number
, description
))
63 def enable_ust_tracepoint_event(session_info
, event_name
):
65 event
.name
= event_name
66 event
.type = EVENT_TRACEPOINT
67 event
.loglevel
= EVENT_LOGLEVEL_ALL
68 res
= enable_event(session_info
.handle
, event
, session_info
.channel_name
)
70 bail("Failed to enable userspace event " + event_name
, session_info
)
76 session_name
= str(uuid
.uuid1())
77 tmp_directory
= tempfile
.mkdtemp()
78 trace_path
= tmp_directory
+ "/" + session_name
80 if 'LTTNG_TEST_TRACE_FORMAT' in os
.environ
:
81 trace_format_env
= os
.getenv('LTTNG_TEST_TRACE_FORMAT')
83 trace_format_env
= 'ctf1'
85 trace_format_desc
= trace_format_mapping
[value
]()
87 session_descriptor
= lttng_session_descriptor_local_create(session_name
, trace_path
)
89 lttng_session_descriptor_set_trace_format_descriptor(session_descriptor
, trace_format_desc
)
91 res
= lttng_create_session_ext(session_descriptor
)
93 bail("Failed to create recording session.")
96 channel
.name
= "channel0"
97 channel_set_default_attr(dom
, channel
.attr
)
99 han
= Handle(session_name
, dom
)
100 res
= enable_channel(han
, channel
)
102 session_info
= SessionInfo(han
, session_name
, tmp_directory
, channel
.name
)
104 bail("Failed to enable channel " + channel
.name
, session_info
)
107 def start_session(session_info
):
108 start(session_info
.name
)
110 def stop_session(session_info
, bailing
= False):
111 # Workaround lttng-ctl outputing directly to stdout by spawning a subprocess.
112 lttng_binary_path
= os
.path
.dirname(os
.path
.abspath(__file__
)) + "/"
114 lttng_binary_path
= os
.path
.dirname(lttng_binary_path
)
115 lttng_binary_path
= lttng_binary_path
+ "/src/bin/lttng/lttng"
117 retcode
= subprocess
.call([lttng_binary_path
, "stop", session_info
.name
], stdout
=subprocess
.PIPE
, stderr
=subprocess
.PIPE
)
118 if retcode
!= 0 and not bailing
:
119 bail("Unable to stop session " + session_info
.name
, session_info
)
120 destroy(session_info
.name
)