3 # Copyright (C) 2023 Jérémie Galarneau <jeremie.galarneau@efficios.com>
5 # SPDX-License-Identifier: GPL-2.0-only
10 from typing
import Any
, Callable
, Type
, Dict
, Iterator
13 from collections
.abc
import Mapping
16 Test the session commands of the `lttng` CLI client.
19 # Import in-tree test utils
20 test_utils_import_path
= pathlib
.Path(__file__
).absolute().parents
[3] / "utils"
21 sys
.path
.append(str(test_utils_import_path
))
27 class SessionSet(Mapping
):
28 def __init__(self
, client
, name_prefixes
):
29 self
._sessions
= {} # type dict[str, lttngtest.Session]
30 for prefix
in name_prefixes
:
31 new_session
= client
.create_session(
32 name
=self
._generate
_session
_name
_from
_prefix
(prefix
),
33 output
=lttngtest
.LocalSessionOutputLocation(
34 test_env
.create_temporary_directory("trace")
37 # Add a channel to all sessions to ensure the sessions can be started.
38 new_session
.add_channel(lttngtest
.TracingDomain
.User
)
39 self
._sessions
[prefix
] = new_session
42 def _generate_session_name_from_prefix(prefix
):
48 random
.choice(string
.ascii_lowercase
+ string
.digits
) for _
in range(8)
52 def __getitem__(self
, __key
):
53 # type: (str) -> lttngtest.Session
54 return self
._sessions
[__key
]
58 return len(self
._sessions
)
61 # type: () -> Iterator[str]
62 return iter(self
._sessions
)
65 def test_start_globbing(tap
, test_env
):
66 # type: (lttngtest.TapGenerator, lttngtest._Environment) -> None
67 tap
.diagnostic("Test --glob match of start command")
68 name_prefixes
= ["abba", "alakazou", "alakazam"]
70 client
= lttngtest
.LTTngClient(test_env
, log
=tap
.diagnostic
)
72 tap
.diagnostic("Create a set of sessions to test globbing")
75 "Create sessions with prefixes [{}]".format(", ".join(name_prefixes
))
77 sessions
= SessionSet(client
, name_prefixes
)
80 all(not session
.is_active
for prefix
, session
in sessions
.items()),
81 "All sessions created are in the inactive state",
84 start_pattern
= "alak*"
85 with tap
.case("Start sessions with --glob={}".format(start_pattern
)) as test_case
:
86 client
.start_session_by_glob_pattern(start_pattern
)
89 sessions
["alakazou"].is_active
90 and sessions
["alakazam"].is_active
91 and not sessions
["abba"].is_active
,
92 "Only sessions 'alakazou' and 'alakazam' are active",
96 "Starting already started sessions with --glob={} doesn't produce an error".format(
100 client
.start_session_by_glob_pattern(start_pattern
)
102 start_pattern
= "tintina*"
104 "Starting with --glob={} that doesn't match any session doesn't produce an error".format(
108 client
.start_session_by_glob_pattern(start_pattern
)
110 for name
, session
in sessions
.items():
114 "Starting with --glob={} when no sessions exist doesn't produce an error".format(
118 client
.start_session_by_glob_pattern(start_pattern
)
121 def test_start_single(tap
, test_env
):
122 # type: (lttngtest.TapGenerator, lttngtest._Environment) -> None
123 tap
.diagnostic("Test match of start command targeting a single session")
124 name_prefixes
= ["un", "deux", "patate", "pouel"]
126 client
= lttngtest
.LTTngClient(test_env
, log
=tap
.diagnostic
)
128 tap
.diagnostic("Create a set of sessions to test single session start")
131 "Create sessions with prefixes [{}]".format(", ".join(name_prefixes
))
133 sessions
= SessionSet(client
, name_prefixes
)
136 all(not session
.is_active
for prefix
, session
in sessions
.items()),
137 "All sessions created are in the inactive state",
140 session_to_start_prefix
= "patate"
141 full_session_name
= sessions
[session_to_start_prefix
].name
142 with tap
.case("Start session '{}'".format(session_to_start_prefix
)) as test_case
:
143 client
.start_session_by_name(full_session_name
)
147 session
.is_active
and prefix
!= session_to_start_prefix
148 for prefix
, session
in sessions
.items()
151 "Only session '{}' is active".format(session_to_start_prefix
),
155 "Starting already started session '{}' doesn't produce an error".format(
156 session_to_start_prefix
159 client
.start_session_by_name(full_session_name
)
161 for name
, session
in sessions
.items():
165 def test_start_all(tap
, test_env
):
166 # type: (lttngtest.TapGenerator, lttngtest._Environment) -> None
167 tap
.diagnostic("Test start command with the --all option")
168 name_prefixes
= ["a", "b", "c", "d"]
170 client
= lttngtest
.LTTngClient(test_env
, log
=tap
.diagnostic
)
172 tap
.diagnostic("Create a set of sessions to test starting all sessions")
175 "Create sessions with prefixes [{}]".format(", ".join(name_prefixes
))
177 sessions
= SessionSet(client
, name_prefixes
)
180 all(not session
.is_active
for prefix
, session
in sessions
.items()),
181 "All sessions created are in the inactive state",
184 with tap
.case("Start all sessions") as test_case
:
185 client
.start_sessions_all()
188 all(session
.is_active
for prefix
, session
in sessions
.items()),
189 "All sessions are active",
192 with tap
.case("Starting already started sessions") as test_case
:
193 client
.start_sessions_all()
195 for name
, session
in sessions
.items():
199 "Starting all sessions when none exist doesn't produce an error"
201 client
.start_sessions_all()
204 def test_stop_globbing(tap
, test_env
):
205 # type: (lttngtest.TapGenerator, lttngtest._Environment) -> None
206 tap
.diagnostic("Test --glob match of stop command")
207 name_prefixes
= ["East Farnham", "Amqui", "Amos"]
209 client
= lttngtest
.LTTngClient(test_env
, log
=tap
.diagnostic
)
211 tap
.diagnostic("Create a set of sessions to test globbing")
214 "Create sessions with prefixes [{}]".format(", ".join(name_prefixes
))
216 sessions
= SessionSet(client
, name_prefixes
)
218 client
.start_sessions_all()
220 all(session
.is_active
for prefix
, session
in sessions
.items()),
221 "All sessions are in the active state",
224 stop_pattern
= "Am??i*"
225 with tap
.case("Stop sessions with --glob={}".format(stop_pattern
)) as test_case
:
226 client
.stop_session_by_glob_pattern(stop_pattern
)
230 sessions
["East Farnham"].is_active
231 and sessions
["Amos"].is_active
232 and (not sessions
["Amqui"].is_active
)
234 "Only session 'Amqui' is inactive",
239 "Stopping more sessions, including a stopped session, with --glob={} doesn't produce an error".format(
243 client
.stop_session_by_glob_pattern(stop_pattern
)
246 sessions
["East Farnham"].is_active
247 and (not sessions
["Amqui"].is_active
)
248 and (not sessions
["Amos"].is_active
),
249 "Only session 'East Farnham' is active",
252 stop_pattern
= "Notre-Dame*"
254 "Stopping with --glob={} that doesn't match any session doesn't produce an error".format(
258 client
.stop_session_by_glob_pattern(stop_pattern
)
260 for name
, session
in sessions
.items():
264 "Stopping with --glob={} when no sessions exist doesn't produce an error".format(
268 client
.stop_session_by_glob_pattern(stop_pattern
)
271 def test_stop_single(tap
, test_env
):
272 # type: (lttngtest.TapGenerator, lttngtest._Environment) -> None
273 tap
.diagnostic("Test match of stop command targeting a single session")
274 name_prefixes
= ["Grosses-Roches", "Kazabazua", "Laval", "Magog"]
276 client
= lttngtest
.LTTngClient(test_env
, log
=tap
.diagnostic
)
278 tap
.diagnostic("Create a set of sessions to test single session stop")
281 "Create sessions with prefixes [{}]".format(", ".join(name_prefixes
))
283 sessions
= SessionSet(client
, name_prefixes
)
285 client
.start_sessions_all()
287 all(session
.is_active
for prefix
, session
in sessions
.items()),
288 "All sessions are in the active state",
291 session_to_stop_prefix
= "Kazabazua"
292 full_session_name
= sessions
[session_to_stop_prefix
].name
293 with tap
.case("Stop session '{}'".format(session_to_stop_prefix
)) as test_case
:
294 client
.stop_session_by_name(full_session_name
)
296 inactive_session_prefixes
= [
297 prefix
for prefix
, session
in sessions
.items() if not session
.is_active
300 len(inactive_session_prefixes
) == 1
301 and inactive_session_prefixes
[0] == session_to_stop_prefix
,
302 "Only session '{}' is inactive".format(session_to_stop_prefix
),
306 "Stopping already stopped session '{}' doesn't produce an error".format(
307 session_to_stop_prefix
310 client
.stop_session_by_name(full_session_name
)
312 for name
, session
in sessions
.items():
316 def test_stop_all(tap
, test_env
):
317 # type: (lttngtest.TapGenerator, lttngtest._Environment) -> None
318 tap
.diagnostic("Test stop command with the --all option")
319 name_prefixes
= ["a", "b", "c", "d"]
321 client
= lttngtest
.LTTngClient(test_env
, log
=tap
.diagnostic
)
323 tap
.diagnostic("Create a set of sessions to test stopping all sessions")
326 "Create sessions with prefixes [{}]".format(", ".join(name_prefixes
))
328 sessions
= SessionSet(client
, name_prefixes
)
330 client
.start_sessions_all()
332 all(session
.is_active
for prefix
, session
in sessions
.items()),
333 "All sessions are in the active state",
336 with tap
.case("Stop all sessions") as test_case
:
337 client
.stop_sessions_all()
340 all(not session
.is_active
for prefix
, session
in sessions
.items()),
341 "All sessions are inactive",
344 with tap
.case("Stopping already stopped sessions") as test_case
:
345 client
.stop_sessions_all()
347 for name
, session
in sessions
.items():
351 "Stopping all sessions when none exist doesn't produce an error"
353 client
.stop_sessions_all()
356 def test_destroy_globbing(tap
, test_env
):
357 # type: (lttngtest.TapGenerator, lttngtest._Environment) -> None
358 tap
.diagnostic("Test --glob match of destroy command")
359 name_prefixes
= ["Mont-Laurier", "Montreal", "Montmagny", "Neuville"]
361 client
= lttngtest
.LTTngClient(test_env
, log
=tap
.diagnostic
)
363 tap
.diagnostic("Create a set of sessions to test globbing")
366 "Create sessions with prefixes [{}]".format(", ".join(name_prefixes
))
368 sessions
= SessionSet(client
, name_prefixes
)
370 destroy_pattern
= "Mont*"
372 "Destroy sessions with --glob={}".format(destroy_pattern
)
374 client
.destroy_session_by_glob_pattern(destroy_pattern
)
376 listed_sessions
= client
.list_sessions()
378 len(listed_sessions
) == 1
379 and listed_sessions
[0].name
== sessions
["Neuville"].name
,
380 "Neuville is the only remaining session",
383 for session
in listed_sessions
:
387 "Destroying with --glob={} when no sessions exist doesn't produce an error".format(
391 client
.destroy_session_by_glob_pattern(destroy_pattern
)
394 def test_destroy_single(tap
, test_env
):
395 # type: (lttngtest.TapGenerator, lttngtest._Environment) -> None
396 tap
.diagnostic("Test match of destroy command targeting a single session")
397 name_prefixes
= ["Natashquan", "Normetal", "Notre-Dame-des-Sept-Douleurs"]
399 client
= lttngtest
.LTTngClient(test_env
, log
=tap
.diagnostic
)
401 tap
.diagnostic("Create a set of sessions to test single session destruction")
404 "Create sessions with prefixes [{}]".format(", ".join(name_prefixes
))
406 sessions
= SessionSet(client
, name_prefixes
)
408 session_to_destroy_prefix
= "Normetal"
409 full_session_name
= sessions
[session_to_destroy_prefix
].name
411 "Destroy session '{}'".format(session_to_destroy_prefix
)
413 client
.destroy_session_by_name(full_session_name
)
415 listed_sessions
= client
.list_sessions()
417 len(listed_sessions
) == 2
418 and full_session_name
not in [session
.name
for session
in listed_sessions
],
419 "Session '{}' no longer exists".format(session_to_destroy_prefix
),
422 for session
in listed_sessions
:
426 def test_destroy_all(tap
, test_env
):
427 # type: (lttngtest.TapGenerator, lttngtest._Environment) -> None
428 tap
.diagnostic("Test destroy command with the --all option")
429 name_prefixes
= ["a", "b", "c", "d"]
431 client
= lttngtest
.LTTngClient(test_env
, log
=tap
.diagnostic
)
433 tap
.diagnostic("Create a set of sessions to test destroying all sessions")
436 "Create sessions with prefixes [{}]".format(", ".join(name_prefixes
))
438 sessions
= SessionSet(client
, name_prefixes
)
440 with tap
.case("Destroy all sessions") as test_case
:
441 client
.destroy_sessions_all()
444 len(client
.list_sessions()) == 0,
445 "No sessions exist after destroying all sessions",
449 "Destroy all sessions when none exist doesn't produce an error"
451 client
.destroy_sessions_all()
454 tap
= lttngtest
.TapGenerator(48)
455 tap
.diagnostic("Test client session command --glob and --all options")
457 with lttngtest
.test_environment(with_sessiond
=True, log
=tap
.diagnostic
) as test_env
:
458 test_start_globbing(tap
, test_env
)
460 with lttngtest
.test_environment(with_sessiond
=True, log
=tap
.diagnostic
) as test_env
:
461 test_start_single(tap
, test_env
)
463 with lttngtest
.test_environment(with_sessiond
=True, log
=tap
.diagnostic
) as test_env
:
464 test_start_all(tap
, test_env
)
466 with lttngtest
.test_environment(with_sessiond
=True, log
=tap
.diagnostic
) as test_env
:
467 test_stop_globbing(tap
, test_env
)
469 with lttngtest
.test_environment(with_sessiond
=True, log
=tap
.diagnostic
) as test_env
:
470 test_stop_single(tap
, test_env
)
472 with lttngtest
.test_environment(with_sessiond
=True, log
=tap
.diagnostic
) as test_env
:
473 test_stop_all(tap
, test_env
)
475 with lttngtest
.test_environment(with_sessiond
=True, log
=tap
.diagnostic
) as test_env
:
476 test_destroy_globbing(tap
, test_env
)
478 with lttngtest
.test_environment(with_sessiond
=True, log
=tap
.diagnostic
) as test_env
:
479 test_destroy_single(tap
, test_env
)
481 with lttngtest
.test_environment(with_sessiond
=True, log
=tap
.diagnostic
) as test_env
:
482 test_destroy_all(tap
, test_env
)
484 sys
.exit(0 if tap
.is_successful
else 1)