tests/regression/tools/client/test_session_commands.py

   1 #!/usr/bin/env python3
   2 #
   3 # Copyright (C) 2023 Jérémie Galarneau <jeremie.galarneau@efficios.com>
   4 #
   5 # SPDX-License-Identifier: GPL-2.0-only
   6
   7 from cgi import test
   8 import pathlib
   9 import sys
  10 import os
  11 from typing import Any, Callable, Type, Dict, Iterator
  12 import random
  13 import string
  14 from collections.abc import Mapping
  15
  16 """
  17 Test the session commands of the `lttng` CLI client.
  18 """
  19
  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))
  23
  24 import lttngtest
  25 import bt2
  26
  27
  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")
  36                 ),
  37             )
  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
  41
  42     @staticmethod
  43     def _generate_session_name_from_prefix(prefix):
  44         # type: (str) -> str
  45         return (
  46             prefix
  47             + "_"
  48             + "".join(
  49                 random.choice(string.ascii_lowercase + string.digits) for _ in range(8)
  50             )
  51         )
  52
  53     def __getitem__(self, __key):
  54         # type: (str) -> lttngtest.Session
  55         return self._sessions[__key]
  56
  57     def __len__(self):
  58         # type: () -> int
  59         return len(self._sessions)
  60
  61     def __iter__(self):
  62         # type: () -> Iterator[str]
  63         return iter(self._sessions)
  64
  65
  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"]
  70
  71     client = lttngtest.LTTngClient(test_env, log=tap.diagnostic)
  72
  73     tap.diagnostic("Create a set of sessions to test globbing")
  74     sessions = None
  75     with tap.case(
  76         "Create sessions with prefixes [{}]".format(", ".join(name_prefixes))
  77     ) as test_case:
  78         sessions = SessionSet(client, name_prefixes)
  79
  80     tap.test(
  81         all(not session.is_active for prefix, session in sessions.items()),
  82         "All sessions created are in the inactive state",
  83     )
  84
  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)
  88
  89     tap.test(
  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",
  94     )
  95
  96     with tap.case(
  97         "Starting already started sessions with --glob={} doesn't produce an error".format(
  98             start_pattern
  99         )
 100     ) as test_case:
 101         client.start_session_by_glob_pattern(start_pattern)
 102
 103     start_pattern = "tintina*"
 104     with tap.case(
 105         "Starting with --glob={} that doesn't match any session doesn't produce an error".format(
 106             start_pattern
 107         )
 108     ) as test_case:
 109         client.start_session_by_glob_pattern(start_pattern)
 110
 111     for name, session in sessions.items():
 112         session.destroy()
 113
 114     with tap.case(
 115         "Starting with --glob={} when no sessions exist doesn't produce an error".format(
 116             start_pattern
 117         )
 118     ) as test_case:
 119         client.start_session_by_glob_pattern(start_pattern)
 120
 121
 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"]
 126
 127     client = lttngtest.LTTngClient(test_env, log=tap.diagnostic)
 128
 129     tap.diagnostic("Create a set of sessions to test single session start")
 130     sessions = None
 131     with tap.case(
 132         "Create sessions with prefixes [{}]".format(", ".join(name_prefixes))
 133     ) as test_case:
 134         sessions = SessionSet(client, name_prefixes)
 135
 136     tap.test(
 137         all(not session.is_active for prefix, session in sessions.items()),
 138         "All sessions created are in the inactive state",
 139     )
 140
 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)
 145
 146     tap.test(
 147         any(
 148             session.is_active and prefix != session_to_start_prefix
 149             for prefix, session in sessions.items()
 150         )
 151         is False,
 152         "Only session '{}' is active".format(session_to_start_prefix),
 153     )
 154
 155     with tap.case(
 156         "Starting already started session '{}' doesn't produce an error".format(
 157             session_to_start_prefix
 158         )
 159     ) as test_case:
 160         client.start_session_by_name(full_session_name)
 161
 162     for name, session in sessions.items():
 163         session.destroy()
 164
 165
 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"]
 170
 171     client = lttngtest.LTTngClient(test_env, log=tap.diagnostic)
 172
 173     tap.diagnostic("Create a set of sessions to test starting all sessions")
 174     sessions = None
 175     with tap.case(
 176         "Create sessions with prefixes [{}]".format(", ".join(name_prefixes))
 177     ) as test_case:
 178         sessions = SessionSet(client, name_prefixes)
 179
 180     tap.test(
 181         all(not session.is_active for prefix, session in sessions.items()),
 182         "All sessions created are in the inactive state",
 183     )
 184
 185     with tap.case("Start all sessions") as test_case:
 186         client.start_sessions_all()
 187
 188     tap.test(
 189         all(session.is_active for prefix, session in sessions.items()),
 190         "All sessions are active",
 191     )
 192
 193     with tap.case("Starting already started sessions") as test_case:
 194         client.start_sessions_all()
 195
 196     for name, session in sessions.items():
 197         session.destroy()
 198
 199     with tap.case(
 200         "Starting all sessions when none exist doesn't produce an error"
 201     ) as test_case:
 202         client.start_sessions_all()
 203
 204
 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"]
 209
 210     client = lttngtest.LTTngClient(test_env, log=tap.diagnostic)
 211
 212     tap.diagnostic("Create a set of sessions to test globbing")
 213     sessions = None
 214     with tap.case(
 215         "Create sessions with prefixes [{}]".format(", ".join(name_prefixes))
 216     ) as test_case:
 217         sessions = SessionSet(client, name_prefixes)
 218
 219     client.start_sessions_all()
 220     tap.test(
 221         all(session.is_active for prefix, session in sessions.items()),
 222         "All sessions are in the active state",
 223     )
 224
 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)
 228
 229     tap.test(
 230         (
 231             sessions["East Farnham"].is_active
 232             and sessions["Amos"].is_active
 233             and (not sessions["Amqui"].is_active)
 234         ),
 235         "Only session 'Amqui' is inactive",
 236     )
 237
 238     stop_pattern = "Am*"
 239     with tap.case(
 240         "Stopping more sessions, including a stopped session, with --glob={} doesn't produce an error".format(
 241             stop_pattern
 242         )
 243     ) as test_case:
 244         client.stop_session_by_glob_pattern(stop_pattern)
 245
 246     tap.test(
 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",
 251     )
 252
 253     stop_pattern = "Notre-Dame*"
 254     with tap.case(
 255         "Stopping with --glob={} that doesn't match any session doesn't produce an error".format(
 256             stop_pattern
 257         )
 258     ) as test_case:
 259         client.stop_session_by_glob_pattern(stop_pattern)
 260
 261     for name, session in sessions.items():
 262         session.destroy()
 263
 264     with tap.case(
 265         "Stopping with --glob={} when no sessions exist doesn't produce an error".format(
 266             stop_pattern
 267         )
 268     ) as test_case:
 269         client.stop_session_by_glob_pattern(stop_pattern)
 270
 271
 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"]
 276
 277     client = lttngtest.LTTngClient(test_env, log=tap.diagnostic)
 278
 279     tap.diagnostic("Create a set of sessions to test single session stop")
 280     sessions = None
 281     with tap.case(
 282         "Create sessions with prefixes [{}]".format(", ".join(name_prefixes))
 283     ) as test_case:
 284         sessions = SessionSet(client, name_prefixes)
 285
 286     client.start_sessions_all()
 287     tap.test(
 288         all(session.is_active for prefix, session in sessions.items()),
 289         "All sessions are in the active state",
 290     )
 291
 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)
 296
 297     inactive_session_prefixes = [
 298         prefix for prefix, session in sessions.items() if not session.is_active
 299     ]
 300     tap.test(
 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),
 304     )
 305
 306     with tap.case(
 307         "Stopping already stopped session '{}' doesn't produce an error".format(
 308             session_to_stop_prefix
 309         )
 310     ) as test_case:
 311         client.stop_session_by_name(full_session_name)
 312
 313     for name, session in sessions.items():
 314         session.destroy()
 315
 316
 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"]
 321
 322     client = lttngtest.LTTngClient(test_env, log=tap.diagnostic)
 323
 324     tap.diagnostic("Create a set of sessions to test stopping all sessions")
 325     sessions = None
 326     with tap.case(
 327         "Create sessions with prefixes [{}]".format(", ".join(name_prefixes))
 328     ) as test_case:
 329         sessions = SessionSet(client, name_prefixes)
 330
 331     client.start_sessions_all()
 332     tap.test(
 333         all(session.is_active for prefix, session in sessions.items()),
 334         "All sessions are in the active state",
 335     )
 336
 337     with tap.case("Stop all sessions") as test_case:
 338         client.stop_sessions_all()
 339
 340     tap.test(
 341         all(not session.is_active for prefix, session in sessions.items()),
 342         "All sessions are inactive",
 343     )
 344
 345     with tap.case("Stopping already stopped sessions") as test_case:
 346         client.stop_sessions_all()
 347
 348     for name, session in sessions.items():
 349         session.destroy()
 350
 351     with tap.case(
 352         "Stopping all sessions when none exist doesn't produce an error"
 353     ) as test_case:
 354         client.stop_sessions_all()
 355
 356
 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"]
 361
 362     client = lttngtest.LTTngClient(test_env, log=tap.diagnostic)
 363
 364     tap.diagnostic("Create a set of sessions to test globbing")
 365     sessions = None
 366     with tap.case(
 367         "Create sessions with prefixes [{}]".format(", ".join(name_prefixes))
 368     ) as test_case:
 369         sessions = SessionSet(client, name_prefixes)
 370
 371     destroy_pattern = "Mont*"
 372     with tap.case(
 373         "Destroy sessions with --glob={}".format(destroy_pattern)
 374     ) as test_case:
 375         client.destroy_session_by_glob_pattern(destroy_pattern)
 376
 377     listed_sessions = client.list_sessions()
 378     tap.test(
 379         len(listed_sessions) == 1
 380         and listed_sessions[0].name == sessions["Neuville"].name,
 381         "Neuville is the only remaining session",
 382     )
 383
 384     for session in listed_sessions:
 385         session.destroy()
 386
 387     with tap.case(
 388         "Destroying with --glob={} when no sessions exist doesn't produce an error".format(
 389             destroy_pattern
 390         )
 391     ) as test_case:
 392         client.destroy_session_by_glob_pattern(destroy_pattern)
 393
 394
 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"]
 399
 400     client = lttngtest.LTTngClient(test_env, log=tap.diagnostic)
 401
 402     tap.diagnostic("Create a set of sessions to test single session destruction")
 403     sessions = None
 404     with tap.case(
 405         "Create sessions with prefixes [{}]".format(", ".join(name_prefixes))
 406     ) as test_case:
 407         sessions = SessionSet(client, name_prefixes)
 408
 409     session_to_destroy_prefix = "Normetal"
 410     full_session_name = sessions[session_to_destroy_prefix].name
 411     with tap.case(
 412         "Destroy session '{}'".format(session_to_destroy_prefix)
 413     ) as test_case:
 414         client.destroy_session_by_name(full_session_name)
 415
 416     listed_sessions = client.list_sessions()
 417     tap.test(
 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),
 421     )
 422
 423     for session in listed_sessions:
 424         session.destroy()
 425
 426
 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"]
 431
 432     client = lttngtest.LTTngClient(test_env, log=tap.diagnostic)
 433
 434     tap.diagnostic("Create a set of sessions to test destroying all sessions")
 435     sessions = None
 436     with tap.case(
 437         "Create sessions with prefixes [{}]".format(", ".join(name_prefixes))
 438     ) as test_case:
 439         sessions = SessionSet(client, name_prefixes)
 440
 441     with tap.case("Destroy all sessions") as test_case:
 442         client.destroy_sessions_all()
 443
 444     tap.test(
 445         len(client.list_sessions()) == 0,
 446         "No sessions exist after destroying all sessions",
 447     )
 448
 449     with tap.case(
 450         "Destroy all sessions when none exist doesn't produce an error"
 451     ) as test_case:
 452         client.destroy_sessions_all()
 453
 454
 455 tap = lttngtest.TapGenerator(48)
 456 tap.diagnostic("Test client session command --glob and --all options")
 457
 458 with lttngtest.test_environment(with_sessiond=True, log=tap.diagnostic) as test_env:
 459     test_start_globbing(tap, test_env)
 460
 461 with lttngtest.test_environment(with_sessiond=True, log=tap.diagnostic) as test_env:
 462     test_start_single(tap, test_env)
 463
 464 with lttngtest.test_environment(with_sessiond=True, log=tap.diagnostic) as test_env:
 465     test_start_all(tap, test_env)
 466
 467 with lttngtest.test_environment(with_sessiond=True, log=tap.diagnostic) as test_env:
 468     test_stop_globbing(tap, test_env)
 469
 470 with lttngtest.test_environment(with_sessiond=True, log=tap.diagnostic) as test_env:
 471     test_stop_single(tap, test_env)
 472
 473 with lttngtest.test_environment(with_sessiond=True, log=tap.diagnostic) as test_env:
 474     test_stop_all(tap, test_env)
 475
 476 with lttngtest.test_environment(with_sessiond=True, log=tap.diagnostic) as test_env:
 477     test_destroy_globbing(tap, test_env)
 478
 479 with lttngtest.test_environment(with_sessiond=True, log=tap.diagnostic) as test_env:
 480     test_destroy_single(tap, test_env)
 481
 482 with lttngtest.test_environment(with_sessiond=True, log=tap.diagnostic) as test_env:
 483     test_destroy_all(tap, test_env)
 484
 485 sys.exit(0 if tap.is_successful else 1)