Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions src/cwsandbox/cli/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,7 @@
from cwsandbox.cli.logs import logs
from cwsandbox.cli.shell import shell
from cwsandbox.cli.snapshots import snapshots
from cwsandbox.cli.stop import stop_sandbox
from cwsandbox.exceptions import CWSandboxError


Expand Down Expand Up @@ -62,3 +63,4 @@ def cli() -> None:
cli.add_command(logs, "logs")
cli.add_command(shell, "sh")
cli.add_command(snapshots, "snapshots")
cli.add_command(stop_sandbox, "stop")
90 changes: 90 additions & 0 deletions src/cwsandbox/cli/stop.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,90 @@
# SPDX-FileCopyrightText: 2025 CoreWeave, Inc.
# SPDX-License-Identifier: Apache-2.0
# SPDX-PackageName: cwsandbox-client

"""cwsandbox stop - stop a sandbox."""

from __future__ import annotations

import click

from cwsandbox import Sandbox
from cwsandbox._defaults import DEFAULT_GRACEFUL_SHUTDOWN_SECONDS
from cwsandbox.exceptions import SandboxNotFoundError


@click.command("stop")
@click.argument("sandbox_id")
@click.option(
"--missing-ok",
is_flag=True,
default=False,
help="Do not fail if the sandbox is already missing.",
)
@click.option(
"--snapshot-on-stop",
is_flag=True,
default=False,
help="Capture a file-system snapshot before stopping.",
)
@click.option(
"--wait-for-snapshot/--no-wait-for-snapshot",
default=True,
help="Wait for a snapshot-on-stop snapshot to reach ready or failed.",
)
@click.option(
"--request-id",
default=None,
help="Client-supplied request ID for snapshot-on-stop.",
)
@click.option(
"--graceful-shutdown-seconds",
type=click.FloatRange(min=0),
default=DEFAULT_GRACEFUL_SHUTDOWN_SECONDS,
show_default=True,
help="Seconds to wait for graceful shutdown.",
)
@click.option("--quiet", "-q", is_flag=True, default=False, help="Suppress success output.")
def stop_sandbox(
sandbox_id: str,
missing_ok: bool,
snapshot_on_stop: bool,
wait_for_snapshot: bool,
request_id: str | None,
graceful_shutdown_seconds: float,
quiet: bool,
) -> None:
"""Stop a sandbox.

SANDBOX_ID is the ID of the sandbox to stop.
"""
if not snapshot_on_stop and not wait_for_snapshot:
raise click.UsageError("--no-wait-for-snapshot requires --snapshot-on-stop.")
if request_id is not None and not snapshot_on_stop:
raise click.UsageError("--request-id requires --snapshot-on-stop.")

try:
sandbox = Sandbox.from_id(sandbox_id).result()
except SandboxNotFoundError:
if not missing_ok:
raise
if not quiet:
click.echo(f"Sandbox {sandbox_id} is already missing.")
return

sandbox.stop(
snapshot_on_stop=snapshot_on_stop,
graceful_shutdown_seconds=graceful_shutdown_seconds,
missing_ok=missing_ok,
wait_for_ready=wait_for_snapshot,
request_id=request_id,
).result()

if quiet:
return

snapshot_id = sandbox.file_system_snapshot_id
if snapshot_on_stop and snapshot_id:
click.echo(f"Stopped sandbox {sandbox_id}. Snapshot {snapshot_id}.")
else:
click.echo(f"Stopped sandbox {sandbox_id}.")
145 changes: 145 additions & 0 deletions tests/unit/cwsandbox/test_cli_stop.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,145 @@
# SPDX-FileCopyrightText: 2025 CoreWeave, Inc.
# SPDX-License-Identifier: Apache-2.0
# SPDX-PackageName: cwsandbox-client

"""Tests for cwsandbox stop CLI command."""

from __future__ import annotations

from unittest.mock import MagicMock, patch

from click.testing import CliRunner

from cwsandbox._defaults import DEFAULT_GRACEFUL_SHUTDOWN_SECONDS
from cwsandbox.cli import cli
from cwsandbox.exceptions import SandboxNotFoundError
from tests.unit.cwsandbox.conftest import make_operation_ref


def _patch_sandbox(sandbox: MagicMock):
"""Patch cwsandbox.cli.stop.Sandbox.from_id to return *sandbox*."""
return patch(
"cwsandbox.cli.stop.Sandbox",
**{"from_id.return_value": make_operation_ref(sandbox)},
)


class TestStopCommand:
"""Tests for the cwsandbox stop CLI command."""

def test_stop_registered(self) -> None:
"""Stop command is registered on the CLI group."""
runner = CliRunner()
result = runner.invoke(cli, ["stop", "--help"])
assert result.exit_code == 0
assert "SANDBOX_ID" in result.output

def test_stop_sandbox(self) -> None:
"""cwsandbox stop stops a sandbox and prints confirmation."""
mock_sandbox = MagicMock()
mock_sandbox.stop.return_value = make_operation_ref(None)

with _patch_sandbox(mock_sandbox) as mock_sandbox_cls:
runner = CliRunner()
result = runner.invoke(cli, ["stop", "abc-123"])

assert result.exit_code == 0
assert "Stopped sandbox abc-123." in result.output
mock_sandbox_cls.from_id.assert_called_once_with("abc-123")
mock_sandbox.stop.assert_called_once_with(
snapshot_on_stop=False,
graceful_shutdown_seconds=DEFAULT_GRACEFUL_SHUTDOWN_SECONDS,
missing_ok=False,
wait_for_ready=True,
request_id=None,
)

def test_stop_with_options(self) -> None:
"""cwsandbox stop passes stop options correctly."""
mock_sandbox = MagicMock()
mock_sandbox.stop.return_value = make_operation_ref(None)

with _patch_sandbox(mock_sandbox):
runner = CliRunner()
result = runner.invoke(
cli,
[
"stop",
"abc-123",
"--missing-ok",
"--snapshot-on-stop",
"--no-wait-for-snapshot",
"--request-id",
"idem-1",
"--graceful-shutdown-seconds",
"30",
"--quiet",
],
)

assert result.exit_code == 0
assert result.output == ""
mock_sandbox.stop.assert_called_once_with(
snapshot_on_stop=True,
graceful_shutdown_seconds=30.0,
missing_ok=True,
wait_for_ready=False,
request_id="idem-1",
)

def test_stop_snapshot_on_stop_prints_snapshot_id(self) -> None:
"""cwsandbox stop --snapshot-on-stop prints the resulting snapshot ID."""
mock_sandbox = MagicMock()
mock_sandbox.stop.return_value = make_operation_ref(None)
mock_sandbox.file_system_snapshot_id = "fss-123"

with _patch_sandbox(mock_sandbox):
runner = CliRunner()
result = runner.invoke(cli, ["stop", "abc-123", "--snapshot-on-stop"])

assert result.exit_code == 0
assert "Stopped sandbox abc-123. Snapshot fss-123." in result.output

def test_stop_missing_ok_suppresses_from_id_not_found(self) -> None:
"""cwsandbox stop --missing-ok succeeds when the sandbox is already missing."""
mock_op_ref = MagicMock()
mock_op_ref.result.side_effect = SandboxNotFoundError("not found", sandbox_id="bad-id")

with patch("cwsandbox.cli.stop.Sandbox") as mock_sandbox_cls:
mock_sandbox_cls.from_id.return_value = mock_op_ref

runner = CliRunner()
result = runner.invoke(cli, ["stop", "bad-id", "--missing-ok"])

assert result.exit_code == 0
assert "Sandbox bad-id is already missing." in result.output

def test_stop_sandbox_not_found(self) -> None:
"""cwsandbox stop shows clean error for SandboxNotFoundError."""
mock_op_ref = MagicMock()
mock_op_ref.result.side_effect = SandboxNotFoundError("not found", sandbox_id="bad-id")

with patch("cwsandbox.cli.stop.Sandbox") as mock_sandbox_cls:
mock_sandbox_cls.from_id.return_value = mock_op_ref

runner = CliRunner()
result = runner.invoke(cli, ["stop", "bad-id"])

assert result.exit_code == 1
assert "not found" in result.output

def test_stop_no_wait_for_snapshot_requires_snapshot_on_stop(self) -> None:
"""--no-wait-for-snapshot is only valid with --snapshot-on-stop."""
runner = CliRunner()
result = runner.invoke(cli, ["stop", "abc-123", "--no-wait-for-snapshot"])

assert result.exit_code == 2
assert "--no-wait-for-snapshot requires --snapshot-on-stop" in result.output

def test_stop_request_id_requires_snapshot_on_stop(self) -> None:
"""--request-id is only valid with --snapshot-on-stop."""
runner = CliRunner()
result = runner.invoke(cli, ["stop", "abc-123", "--request-id", "idem-1"])

assert result.exit_code == 2
assert "--request-id requires --snapshot-on-stop" in result.output
Loading