Source code for steplib.modules.cli.actions
"""Pure action functions for the cli module (shell command execution)."""
from __future__ import annotations
import re
import subprocess
from steplib.modules.cli.context import CLIContext
[docs]
def cli_run_command(cli_ctx: CLIContext, command: str) -> None:
"""Run a shell command and store the exit code, stdout, and stderr.
Args:
cli_ctx: The cli context.
command: The shell command to execute.
Raises:
subprocess.TimeoutExpired: If the command times out.
"""
result = subprocess.run(
command,
shell=True,
capture_output=True,
text=True,
timeout=30,
check=False,
)
cli_ctx.exit_code = result.returncode
cli_ctx.stdout = result.stdout
cli_ctx.stderr = result.stderr
[docs]
def cli_run_command_with_timeout(cli_ctx: CLIContext, command: str, seconds: int) -> None:
"""Run a shell command with a custom timeout.
Args:
cli_ctx: The cli context.
command: The shell command to execute.
seconds: Timeout in seconds.
Raises:
subprocess.TimeoutExpired: If the command times out.
"""
result = subprocess.run(
command,
shell=True,
capture_output=True,
text=True,
timeout=seconds,
check=False,
)
cli_ctx.exit_code = result.returncode
cli_ctx.stdout = result.stdout
cli_ctx.stderr = result.stderr
[docs]
def cli_assert_exit_code(cli_ctx: CLIContext, expected: int) -> None:
"""Assert that the last command's exit code equals the expected value.
Args:
cli_ctx: The cli context.
expected: Expected exit code.
Raises:
AssertionError: If the exit code does not match.
RuntimeError: If no command has been run.
"""
if cli_ctx.exit_code is None:
raise RuntimeError("No command has been executed.")
if cli_ctx.exit_code != expected:
raise AssertionError(f"Expected exit code {expected}, got {cli_ctx.exit_code}.")
[docs]
def cli_assert_output_contains(cli_ctx: CLIContext, text: str) -> None:
"""Assert that the last command's stdout contains the given text.
Args:
cli_ctx: The cli context.
text: Text to search for in stdout.
Raises:
AssertionError: If stdout does not contain the text.
RuntimeError: If no command has been run.
"""
if cli_ctx.exit_code is None:
raise RuntimeError("No command has been executed.")
if text not in cli_ctx.stdout:
raise AssertionError(f"Expected stdout to contain '{text}', got: {cli_ctx.stdout!r}")
[docs]
def cli_assert_output_equals(cli_ctx: CLIContext, text: str) -> None:
"""Assert that the last command's stdout equals the given text.
Args:
cli_ctx: The cli context.
text: Expected stdout content.
Raises:
AssertionError: If stdout does not match.
RuntimeError: If no command has been run.
"""
if cli_ctx.exit_code is None:
raise RuntimeError("No command has been executed.")
if cli_ctx.stdout != text:
raise AssertionError(f"Expected stdout to be '{text}', got: {cli_ctx.stdout!r}")
[docs]
def cli_assert_stderr_contains(cli_ctx: CLIContext, text: str) -> None:
"""Assert that the last command's stderr contains the given text.
Args:
cli_ctx: The cli context.
text: Text to search for in stderr.
Raises:
AssertionError: If stderr does not contain the text.
RuntimeError: If no command has been run.
"""
if cli_ctx.exit_code is None:
raise RuntimeError("No command has been executed.")
if text not in cli_ctx.stderr:
raise AssertionError(f"Expected stderr to contain '{text}', got: {cli_ctx.stderr!r}")
[docs]
def cli_store_output(cli_ctx: CLIContext, variable: str) -> None:
"""Store the last command's stdout into a variable.
Args:
cli_ctx: The cli context.
variable: Variable name to store the stdout.
Raises:
RuntimeError: If no command has been run.
"""
if cli_ctx.exit_code is None:
raise RuntimeError("No command has been executed.")
cli_ctx.variables[variable] = cli_ctx.stdout
[docs]
def cli_assert_output_not_contains(cli_ctx: CLIContext, text: str) -> None:
"""Assert that the last command's stdout does NOT contain the given text.
Args:
cli_ctx: The cli context.
text: Text to search for in stdout.
Raises:
AssertionError: If stdout contains the text.
RuntimeError: If no command has been run.
"""
if cli_ctx.exit_code is None:
raise RuntimeError("No command has been executed.")
if text in cli_ctx.stdout:
raise AssertionError(f"Expected stdout to NOT contain '{text}', got: {cli_ctx.stdout!r}")
[docs]
def cli_store_stderr(cli_ctx: CLIContext, variable: str) -> None:
"""Store the last command's stderr into a variable.
Args:
cli_ctx: The cli context.
variable: Variable name to store the stderr.
Raises:
RuntimeError: If no command has been run.
"""
if cli_ctx.exit_code is None:
raise RuntimeError("No command has been executed.")
cli_ctx.variables[variable] = cli_ctx.stderr
[docs]
def cli_assert_output_matches(cli_ctx: CLIContext, pattern: str) -> None:
"""Assert that the last command's stdout matches a regex pattern.
Args:
cli_ctx: The cli context.
pattern: Regex pattern to match against stdout.
Raises:
AssertionError: If stdout does not match the pattern.
RuntimeError: If no command has been run.
"""
if cli_ctx.exit_code is None:
raise RuntimeError("No command has been executed.")
try:
if not re.search(pattern, cli_ctx.stdout):
raise AssertionError(
f"Expected stdout to match pattern '{pattern}', got: {cli_ctx.stdout!r}"
)
except re.error as exc:
raise AssertionError(f"Invalid regex pattern '{pattern}': {exc}") from exc