3 # Copyright (C) 2023 Jérémie Galarneau <jeremie.galarneau@efficios.com>
5 # SPDX-License-Identifier: GPL-2.0-only
11 from typing
import Any
, Callable
, Type
, Dict
, Iterator
14 from collections
.abc
import Mapping
17 Test the session commands of the `lttng` CLI client.
20 # Import in-tree test utils
21 test_utils_import_path
= pathlib
.Path(__file__
).absolute().parents
[3] / "utils"
22 sys
.path
.append(str(test_utils_import_path
))
28 class SessionSet(Mapping
):
29 def __init__(self
, client
, name_prefixes
):
30 self
._sessions
= {} # type dict[str, lttngtest.Session]
31 for prefix
in name_prefixes
:
32 new_session
= client
.create_session(
33 name
=self
._generate
_session
_name
_from
_prefix
(prefix
),
34 output
=lttngtest
.LocalSessionOutputLocation(
35 test_env
.create_temporary_directory("trace")
38 # Add a channel to all sessions to ensure the sessions can be started.
39 new_session
.add_channel(lttngtest
.TracingDomain
.User
)
40 self
._sessions
[prefix
] = new_session
43 def _generate_session_name_from_prefix(prefix
):
49 random
.choice(string
.ascii_lowercase
+ string
.digits
) for _
in range(8)
53 def __getitem__(self
, __key
):
54 # type: (str) -> lttngtest.Session
55 return self
._sessions
[__key
]
59 return len(self
._sessions
)
62 # type: () -> Iterator[str]
63 return iter(self
._sessions
)
66 def test_start_globbing(tap
, test_env
):
67 # type: (lttngtest.TapGenerator, lttngtest._Environment) -> None
68 tap
.diagnostic("Test --glob match of start command")
69 name_prefixes
= ["abba", "alakazou", "alakazam"]
71 client
= lttngtest
.LTTngClient(test_env
, log
=tap
.diagnostic
)
73 tap
.diagnostic("Create a set of sessions to test globbing")
76 "Create sessions with prefixes [{}]".format(", ".join(name_prefixes
))
78 sessions
= SessionSet(client
, name_prefixes
)
81 all(not session
.is_active
for prefix
, session
in sessions
.items()),
82 "All sessions created are in the inactive state",
85 start_pattern
= "alak*"
86 with tap
.case("Start sessions with --glob={}".format(start_pattern
)) as test_case
:
87 client
.start_session_by_glob_pattern(start_pattern
)
90 sessions
["alakazou"].is_active
91 and sessions
["alakazam"].is_active
92 and not sessions
["abba"].is_active
,
93 "Only sessions 'alakazou' and 'alakazam' are active",
97 "Starting already started sessions with --glob={} doesn't produce an error".format(
101 client
.start_session_by_glob_pattern(start_pattern
)
103 start_pattern
= "tintina*"
105 "Starting with --glob={} that doesn't match any session doesn't produce an error".format(
109 client
.start_session_by_glob_pattern(start_pattern
)
111 for name
, session
in sessions
.items():
115 "Starting with --glob={} when no sessions exist doesn't produce an error".format(
119 client
.start_session_by_glob_pattern(start_pattern
)
122 def test_start_single(tap
, test_env
):
123 # type: (lttngtest.TapGenerator, lttngtest._Environment) -> None
124 tap
.diagnostic("Test match of start command targeting a single session")
125 name_prefixes
= ["un", "deux", "patate", "pouel"]
127 client
= lttngtest
.LTTngClient(test_env
, log
=tap
.diagnostic
)
129 tap
.diagnostic("Create a set of sessions to test single session start")
132 "Create sessions with prefixes [{}]".format(", ".join(name_prefixes
))
134 sessions
= SessionSet(client
, name_prefixes
)
137 all(not session
.is_active
for prefix
, session
in sessions
.items()),
138 "All sessions created are in the inactive state",
141 session_to_start_prefix
= "patate"
142 full_session_name
= sessions
[session_to_start_prefix
].name
143 with tap
.case("Start session '{}'".format(session_to_start_prefix
)) as test_case
:
144 client
.start_session_by_name(full_session_name
)
148 session
.is_active
and prefix
!= session_to_start_prefix
149 for prefix
, session
in sessions
.items()
152 "Only session '{}' is active".format(session_to_start_prefix
),
156 "Starting already started session '{}' doesn't produce an error".format(
157 session_to_start_prefix
160 client
.start_session_by_name(full_session_name
)
162 for name
, session
in sessions
.items():
166 def test_start_all(tap
, test_env
):
167 # type: (lttngtest.TapGenerator, lttngtest._Environment) -> None
168 tap
.diagnostic("Test start command with the --all option")
169 name_prefixes
= ["a", "b", "c", "d"]
171 client
= lttngtest
.LTTngClient(test_env
, log
=tap
.diagnostic
)
173 tap
.diagnostic("Create a set of sessions to test starting all sessions")
176 "Create sessions with prefixes [{}]".format(", ".join(name_prefixes
))
178 sessions
= SessionSet(client
, name_prefixes
)
181 all(not session
.is_active
for prefix
, session
in sessions
.items()),
182 "All sessions created are in the inactive state",
185 with tap
.case("Start all sessions") as test_case
:
186 client
.start_sessions_all()
189 all(session
.is_active
for prefix
, session
in sessions
.items()),
190 "All sessions are active",
193 with tap
.case("Starting already started sessions") as test_case
:
194 client
.start_sessions_all()
196 for name
, session
in sessions
.items():
200 "Starting all sessions when none exist doesn't produce an error"
202 client
.start_sessions_all()
205 def test_stop_globbing(tap
, test_env
):
206 # type: (lttngtest.TapGenerator, lttngtest._Environment) -> None
207 tap
.diagnostic("Test --glob match of stop command")
208 name_prefixes
= ["East Farnham", "Amqui", "Amos"]
210 client
= lttngtest
.LTTngClient(test_env
, log
=tap
.diagnostic
)
212 tap
.diagnostic("Create a set of sessions to test globbing")
215 "Create sessions with prefixes [{}]".format(", ".join(name_prefixes
))
217 sessions
= SessionSet(client
, name_prefixes
)
219 client
.start_sessions_all()
221 all(session
.is_active
for prefix
, session
in sessions
.items()),
222 "All sessions are in the active state",
225 stop_pattern
= "Am??i*"
226 with tap
.case("Stop sessions with --glob={}".format(stop_pattern
)) as test_case
:
227 client
.stop_session_by_glob_pattern(stop_pattern
)
231 sessions
["East Farnham"].is_active
232 and sessions
["Amos"].is_active
233 and (not sessions
["Amqui"].is_active
)
235 "Only session 'Amqui' is inactive",
240 "Stopping more sessions, including a stopped session, with --glob={} doesn't produce an error".format(
244 client
.stop_session_by_glob_pattern(stop_pattern
)
247 sessions
["East Farnham"].is_active
248 and (not sessions
["Amqui"].is_active
)
249 and (not sessions
["Amos"].is_active
),
250 "Only session 'East Farnham' is active",
253 stop_pattern
= "Notre-Dame*"
255 "Stopping with --glob={} that doesn't match any session doesn't produce an error".format(
259 client
.stop_session_by_glob_pattern(stop_pattern
)
261 for name
, session
in sessions
.items():
265 "Stopping with --glob={} when no sessions exist doesn't produce an error".format(
269 client
.stop_session_by_glob_pattern(stop_pattern
)
272 def test_stop_single(tap
, test_env
):
273 # type: (lttngtest.TapGenerator, lttngtest._Environment) -> None
274 tap
.diagnostic("Test match of stop command targeting a single session")
275 name_prefixes
= ["Grosses-Roches", "Kazabazua", "Laval", "Magog"]
277 client
= lttngtest
.LTTngClient(test_env
, log
=tap
.diagnostic
)
279 tap
.diagnostic("Create a set of sessions to test single session stop")
282 "Create sessions with prefixes [{}]".format(", ".join(name_prefixes
))
284 sessions
= SessionSet(client
, name_prefixes
)
286 client
.start_sessions_all()
288 all(session
.is_active
for prefix
, session
in sessions
.items()),
289 "All sessions are in the active state",
292 session_to_stop_prefix
= "Kazabazua"
293 full_session_name
= sessions
[session_to_stop_prefix
].name
294 with tap
.case("Stop session '{}'".format(session_to_stop_prefix
)) as test_case
:
295 client
.stop_session_by_name(full_session_name
)
297 inactive_session_prefixes
= [
298 prefix
for prefix
, session
in sessions
.items() if not session
.is_active
301 len(inactive_session_prefixes
) == 1
302 and inactive_session_prefixes
[0] == session_to_stop_prefix
,
303 "Only session '{}' is inactive".format(session_to_stop_prefix
),
307 "Stopping already stopped session '{}' doesn't produce an error".format(
308 session_to_stop_prefix
311 client
.stop_session_by_name(full_session_name
)
313 for name
, session
in sessions
.items():
317 def test_stop_all(tap
, test_env
):
318 # type: (lttngtest.TapGenerator, lttngtest._Environment) -> None
319 tap
.diagnostic("Test stop command with the --all option")
320 name_prefixes
= ["a", "b", "c", "d"]
322 client
= lttngtest
.LTTngClient(test_env
, log
=tap
.diagnostic
)
324 tap
.diagnostic("Create a set of sessions to test stopping all sessions")
327 "Create sessions with prefixes [{}]".format(", ".join(name_prefixes
))
329 sessions
= SessionSet(client
, name_prefixes
)
331 client
.start_sessions_all()
333 all(session
.is_active
for prefix
, session
in sessions
.items()),
334 "All sessions are in the active state",
337 with tap
.case("Stop all sessions") as test_case
:
338 client
.stop_sessions_all()
341 all(not session
.is_active
for prefix
, session
in sessions
.items()),
342 "All sessions are inactive",
345 with tap
.case("Stopping already stopped sessions") as test_case
:
346 client
.stop_sessions_all()
348 for name
, session
in sessions
.items():
352 "Stopping all sessions when none exist doesn't produce an error"
354 client
.stop_sessions_all()
357 def test_destroy_globbing(tap
, test_env
):
358 # type: (lttngtest.TapGenerator, lttngtest._Environment) -> None
359 tap
.diagnostic("Test --glob match of destroy command")
360 name_prefixes
= ["Mont-Laurier", "Montreal", "Montmagny", "Neuville"]
362 client
= lttngtest
.LTTngClient(test_env
, log
=tap
.diagnostic
)
364 tap
.diagnostic("Create a set of sessions to test globbing")
367 "Create sessions with prefixes [{}]".format(", ".join(name_prefixes
))
369 sessions
= SessionSet(client
, name_prefixes
)
371 destroy_pattern
= "Mont*"
373 "Destroy sessions with --glob={}".format(destroy_pattern
)
375 client
.destroy_session_by_glob_pattern(destroy_pattern
)
377 listed_sessions
= client
.list_sessions()
379 len(listed_sessions
) == 1
380 and listed_sessions
[0].name
== sessions
["Neuville"].name
,
381 "Neuville is the only remaining session",
384 for session
in listed_sessions
:
388 "Destroying with --glob={} when no sessions exist doesn't produce an error".format(
392 client
.destroy_session_by_glob_pattern(destroy_pattern
)
395 def test_destroy_single(tap
, test_env
):
396 # type: (lttngtest.TapGenerator, lttngtest._Environment) -> None
397 tap
.diagnostic("Test match of destroy command targeting a single session")
398 name_prefixes
= ["Natashquan", "Normetal", "Notre-Dame-des-Sept-Douleurs"]
400 client
= lttngtest
.LTTngClient(test_env
, log
=tap
.diagnostic
)
402 tap
.diagnostic("Create a set of sessions to test single session destruction")
405 "Create sessions with prefixes [{}]".format(", ".join(name_prefixes
))
407 sessions
= SessionSet(client
, name_prefixes
)
409 session_to_destroy_prefix
= "Normetal"
410 full_session_name
= sessions
[session_to_destroy_prefix
].name
412 "Destroy session '{}'".format(session_to_destroy_prefix
)
414 client
.destroy_session_by_name(full_session_name
)
416 listed_sessions
= client
.list_sessions()
418 len(listed_sessions
) == 2
419 and full_session_name
not in [session
.name
for session
in listed_sessions
],
420 "Session '{}' no longer exists".format(session_to_destroy_prefix
),
423 for session
in listed_sessions
:
427 def test_destroy_all(tap
, test_env
):
428 # type: (lttngtest.TapGenerator, lttngtest._Environment) -> None
429 tap
.diagnostic("Test destroy command with the --all option")
430 name_prefixes
= ["a", "b", "c", "d"]
432 client
= lttngtest
.LTTngClient(test_env
, log
=tap
.diagnostic
)
434 tap
.diagnostic("Create a set of sessions to test destroying all sessions")
437 "Create sessions with prefixes [{}]".format(", ".join(name_prefixes
))
439 sessions
= SessionSet(client
, name_prefixes
)
441 with tap
.case("Destroy all sessions") as test_case
:
442 client
.destroy_sessions_all()
445 len(client
.list_sessions()) == 0,
446 "No sessions exist after destroying all sessions",
450 "Destroy all sessions when none exist doesn't produce an error"
452 client
.destroy_sessions_all()
455 tap
= lttngtest
.TapGenerator(48)
456 tap
.diagnostic("Test client session command --glob and --all options")
458 with lttngtest
.test_environment(with_sessiond
=True, log
=tap
.diagnostic
) as test_env
:
459 test_start_globbing(tap
, test_env
)
461 with lttngtest
.test_environment(with_sessiond
=True, log
=tap
.diagnostic
) as test_env
:
462 test_start_single(tap
, test_env
)
464 with lttngtest
.test_environment(with_sessiond
=True, log
=tap
.diagnostic
) as test_env
:
465 test_start_all(tap
, test_env
)
467 with lttngtest
.test_environment(with_sessiond
=True, log
=tap
.diagnostic
) as test_env
:
468 test_stop_globbing(tap
, test_env
)
470 with lttngtest
.test_environment(with_sessiond
=True, log
=tap
.diagnostic
) as test_env
:
471 test_stop_single(tap
, test_env
)
473 with lttngtest
.test_environment(with_sessiond
=True, log
=tap
.diagnostic
) as test_env
:
474 test_stop_all(tap
, test_env
)
476 with lttngtest
.test_environment(with_sessiond
=True, log
=tap
.diagnostic
) as test_env
:
477 test_destroy_globbing(tap
, test_env
)
479 with lttngtest
.test_environment(with_sessiond
=True, log
=tap
.diagnostic
) as test_env
:
480 test_destroy_single(tap
, test_env
)
482 with lttngtest
.test_environment(with_sessiond
=True, log
=tap
.diagnostic
) as test_env
:
483 test_destroy_all(tap
, test_env
)
485 sys
.exit(0 if tap
.is_successful
else 1)