From 17f26d71693eada3220718e2543d58b804c3d60e Mon Sep 17 00:00:00 2001 From: zeke <40004347+KAJdev@users.noreply.github.com> Date: Mon, 19 Oct 2026 14:38:44 -0700 Subject: [PATCH 1/4] feat: add rp volume commands for network and global volumes --- runpod/rp_cli/main.py | 3 +- runpod/rp_cli/volume.py | 372 +++++++++++++++++++++ tests/test_cli/test_volume_command.py | 449 ++++++++++++++++++++++++++ 3 files changed, 823 insertions(+), 1 deletion(-) create mode 100644 runpod/rp_cli/volume.py create mode 100644 tests/test_cli/test_volume_command.py diff --git a/runpod/rp_cli/main.py b/runpod/rp_cli/main.py index d8be18330..9c6f15463 100644 --- a/runpod/rp_cli/main.py +++ b/runpod/rp_cli/main.py @@ -1039,8 +1039,9 @@ def _mount_groups() -> None: from runpod.cli.groups.pod.commands import pod_cli from runpod.cli.groups.ssh.commands import ssh_cli from runpod.rp_cli.catalog import catalog + from runpod.rp_cli.volume import volume - for command in (ssh_cli, pod_cli, catalog): + for command in (ssh_cli, pod_cli, catalog, volume): cli.add_command(command) diff --git a/runpod/rp_cli/volume.py b/runpod/rp_cli/volume.py new file mode 100644 index 000000000..43be7cc51 --- /dev/null +++ b/runpod/rp_cli/volume.py @@ -0,0 +1,372 @@ +"""`rp volume`: network and global volume management.""" + +import re +from typing import NamedTuple, Optional + +import click + +from runpod.api.rest import run_rest_request +from runpod.rp_cli import compute +from runpod.rp_cli.output import ( + api_errors, + completer, + confirm, + json_option, + pick, + print_fields, + print_json, + print_table, +) + +MIN_SIZE = 10 +MAX_SIZE = 4096 + + +class Kind(NamedTuple): + path: str + collection: str + noun: str + + +NETWORK = Kind("/v2/network-volumes", "networkVolumes", "network volume") +GLOBAL = Kind("/v2/global-volumes", "globalVolumes", "global volume") + + +def _kind(global_volume: bool) -> Kind: + return GLOBAL if global_volume else NETWORK + + +def _type_label(record: dict) -> Optional[str]: + return (record.get("type") or "").lower() or None + + +def _size_label(record: dict) -> Optional[str]: + return f"{record['size']} GB" if record.get("size") else None + + +def _columns(kind: Kind): + if kind is GLOBAL: + return [ + ("ID", lambda row: row["id"]), + ("NAME", lambda row: row.get("name")), + ("CREATED", lambda row: row.get("createdAt")), + ("CREATED BY", lambda row: row.get("createdByUserId")), + ] + return [ + ("ID", lambda row: row["id"]), + ("NAME", lambda row: row.get("name")), + ("SIZE", _size_label), + ("DATA CENTER", lambda row: row.get("dataCenter")), + ("TYPE", _type_label), + ] + + +def _list(kind: Kind) -> list[dict]: + return compute.list_all(kind.path, kind.collection) + + +def _get(kind: Kind, volume_id: str) -> dict: + return run_rest_request("GET", f"{kind.path}/{volume_id}") or {} + + +def _label(kind: Kind, record: dict) -> str: + if kind is GLOBAL: + return record.get("name") or "" + details = [_size_label(record), record.get("dataCenter"), _type_label(record)] + return " ".join([record.get("name") or ""] + [d for d in details if d]) + + +def _choices(kind: Kind) -> list[tuple[str, str]]: + return [(record["id"], _label(kind, record)) for record in _list(kind)] + + +def _complete_ids(ctx, param, incomplete): + kind = _kind(bool(ctx.params.get("global_volume"))) + return completer(lambda: _choices(kind))(ctx, param, incomplete) + + +def _show(kind: Kind, record: dict, as_json: bool) -> None: + if as_json: + print_json(record) + return + print_fields(record, [(header.lower(), fn) for header, fn in _columns(kind)]) + + +def _data_center( + regions: tuple[str, ...], data_center: Optional[str], volume_type: Optional[str] +) -> str: + """the single data center a network volume is created in. + + regions narrow to data centers offering the requested volume type + (or any type), then a person picks one; scripts get the candidates. + """ + if data_center: + return data_center + datacenters = compute.list_datacenters() + if regions: + allowed = set(compute.resolve_regions(regions, datacenters)) + datacenters = [dc for dc in datacenters if dc["id"] in allowed] + candidates = [ + dc + for dc in datacenters + if (volume_type in (dc.get("networkVolumeTypes") or [])) + or (not volume_type and dc.get("networkVolumeTypes")) + ] + scope = f"region {', '.join(regions)}" if regions else "the catalog" + offered = ( + f"{volume_type.lower()} network volumes" if volume_type else "network volumes" + ) + if not candidates: + raise compute.ResolutionError(f"no data center in {scope} offers {offered}") + if len(candidates) == 1: + return candidates[0]["id"] + ids = ", ".join(dc["id"] for dc in candidates) + return pick( + "data center", + [ + ( + dc["id"], + f"{dc['id']} {(dc.get('region') or '').lower().replace('_', ' ')} " + + ", ".join(t.lower() for t in dc["networkVolumeTypes"]), + ) + for dc in candidates + ], + f"several data centers in {scope} offer {offered} ({ids}). " + "pass one with --data-center", + ) + + +def _parse_growth(value: str) -> tuple[bool, int]: + """split an expand size into (relative, gigabytes): N or +N.""" + text = value.strip() + if not re.fullmatch(r"\+?\d+", text): + raise click.BadParameter( + f"'{value}' is not a size. use N for a total or +N to add N GB", + param_hint="--size", + ) + return text.startswith("+"), int(text.lstrip("+")) + + +def _check_growth(volume_id: str, current: int, size: int, strict: bool) -> None: + if size < current or (strict and size == current): + raise click.UsageError( + f"network volumes can only grow: {volume_id} is {current} GB, " + f"requested {size} GB" + ) + if size > MAX_SIZE: + raise click.UsageError( + f"network volumes hold at most {MAX_SIZE} GB, requested {size} GB" + ) + + +global_option = click.option( + "--global", + "global_volume", + is_flag=True, + help="Manage global volumes, which mount in any data center.", +) + + +@click.group() +def volume(): + """Create, list, resize, and delete network and global volumes.""" + + +@volume.command(name="list") +@global_option +@json_option +def volume_list(global_volume, as_json): + """List volumes with their size, data center, and type.""" + kind = _kind(global_volume) + with api_errors(): + rows = _list(kind) + if as_json: + print_json(rows) + return + print_table( + rows, _columns(kind), f"no {kind.noun}s. create one with rp volume create" + ) + + +@volume.command(name="get") +@click.argument("volume_id", required=False, shell_complete=_complete_ids) +@global_option +@json_option +def volume_get(volume_id, global_volume, as_json): + """Show one volume (pick one when VOLUME_ID is omitted).""" + kind = _kind(global_volume) + with api_errors(): + if volume_id is None: + volume_id = pick(kind.noun, _choices(kind), "pass VOLUME_ID") + record = _get(kind, volume_id) + _show(kind, record, as_json) + + +@volume.command(name="create") +@click.option("--name", required=True, help="Human-readable name.") +@click.option( + "--size", + type=click.IntRange(MIN_SIZE, MAX_SIZE), + default=None, + help=f"Storage in GB ({MIN_SIZE}-{MAX_SIZE}). Required for network volumes.", +) +@click.option( + "--region", + "regions", + multiple=True, + help="Region (eu, na), id prefix (us, eu-ro), or data center id. Repeatable.", +) +@click.option("--data-center", default=None, help="Exact data center id, like EU-RO-1.") +@click.option( + "--type", + "volume_type", + type=click.Choice(["standard", "high_performance"], case_sensitive=False), + default=None, + help="Storage tier, immutable after creation. Defaults to the data center's " + "primary tier.", +) +@global_option +@json_option +def volume_create( + name, size, regions, data_center, volume_type, global_volume, as_json +): # pylint: disable=too-many-arguments + """Create a network volume in one data center, or a global volume. + + A region matching several data centers narrows to those offering + the requested --type, then prompts for one. + + \b + rp volume create --name models --size 100 --region eu + rp volume create --name scratch --size 50 --data-center EU-RO-1 + rp volume create --name shared --global + """ + from runpod.rp_cli import console as ui + + kind = _kind(global_volume) + if global_volume: + given = [ + flag + for flag, value in ( + ("--size", size), + ("--region", regions), + ("--data-center", data_center), + ("--type", volume_type), + ) + if value + ] + if given: + raise click.UsageError(f"{', '.join(given)} do not apply to global volumes") + body = {"name": name} + else: + if size is None: + raise click.UsageError("pass --size in GB for a network volume") + if regions and data_center: + raise click.UsageError("use either --region or --data-center") + volume_type = volume_type.upper() if volume_type else None + with api_errors(): + body = { + "name": name, + "size": size, + "dataCenter": _data_center(regions, data_center, volume_type), + } + if volume_type: + body["type"] = volume_type + + with api_errors(): + record = run_rest_request("POST", kind.path, json=body) or {} + if as_json: + print_json(record) + return + ui.success(f"created {kind.noun} [white]{record.get('id')}[/white]") + _show(kind, record, as_json) + + +@volume.command(name="update") +@click.argument("volume_id", shell_complete=_complete_ids) +@click.option("--name", default=None, help="New name.") +@click.option( + "--size", + type=click.IntRange(MIN_SIZE, MAX_SIZE), + default=None, + help="New size in GB; network volumes can only grow.", +) +@global_option +@json_option +def volume_update(volume_id, name, size, global_volume, as_json): + """Rename and/or resize a volume.""" + from runpod.rp_cli import console as ui + + kind = _kind(global_volume) + if global_volume and size is not None: + raise click.UsageError("global volumes have no size to change") + if name is None and size is None: + raise click.UsageError("pass --name and/or --size") + body = { + key: value + for key, value in (("name", name), ("size", size)) + if value is not None + } + with api_errors(): + if size is not None: + current = _get(kind, volume_id).get("size") or 0 + _check_growth(volume_id, current, size, strict=False) + record = run_rest_request("PATCH", f"{kind.path}/{volume_id}", json=body) or {} + if as_json: + print_json(record) + return + ui.success(f"updated {kind.noun} [white]{volume_id}[/white]") + _show(kind, record, as_json) + + +@volume.command(name="expand") +@click.argument("volume_id", shell_complete=_complete_ids) +@click.option( + "--size", + required=True, + metavar="N|+N", + help="New total size in GB, or +N to add N GB.", +) +@json_option +def volume_expand(volume_id, size, as_json): + """Grow a network volume; storage can never shrink.""" + from runpod.rp_cli import console as ui + + relative, amount = _parse_growth(size) + with api_errors(): + current = _get(NETWORK, volume_id).get("size") or 0 + target = current + amount if relative else amount + _check_growth(volume_id, current, target, strict=True) + record = ( + run_rest_request( + "PATCH", f"{NETWORK.path}/{volume_id}", json={"size": target} + ) + or {} + ) + if as_json: + print_json(record) + return + ui.success( + f"expanded [white]{volume_id}[/white] from {current} GB to " + f"{record.get('size', target)} GB" + ) + + +@volume.command(name="delete") +@click.argument("volume_id", shell_complete=_complete_ids) +@click.option("--yes", "-y", is_flag=True, help="Skip the confirmation prompt.") +@global_option +def volume_delete(volume_id, yes, global_volume): + """Delete a volume and everything stored on it.""" + from runpod.rp_cli import console as ui + + kind = _kind(global_volume) + with api_errors(): + if not yes: + record = _get(kind, volume_id) + confirm( + f"delete {kind.noun} {volume_id} ({_label(kind, record)}) " + "and all its data?", + yes, + ) + run_rest_request("DELETE", f"{kind.path}/{volume_id}") + ui.success(f"deleted {kind.noun} [white]{volume_id}[/white]") diff --git a/tests/test_cli/test_volume_command.py b/tests/test_cli/test_volume_command.py new file mode 100644 index 000000000..0aa0c8ad2 --- /dev/null +++ b/tests/test_cli/test_volume_command.py @@ -0,0 +1,449 @@ +"""tests for the rp volume command group.""" + +import json +from unittest.mock import patch + +import pytest +from click.testing import CliRunner + +from runpod import error +from runpod.rp_cli.main import cli + +VOLUME = { + "id": "vol1", + "name": "models", + "size": 100, + "dataCenter": "EU-RO-1", + "type": "STANDARD", +} +GLOBAL_VOLUME = { + "id": "gsb_abc", + "name": "shared", + "createdAt": "2026-09-24T00:00:00Z", + "createdByUserId": "user_123", +} +DATACENTERS = [ + {"id": "EU-RO-1", "region": "EUROPE", "networkVolumeTypes": ["STANDARD"]}, + {"id": "EU-FR-1", "region": "EUROPE", "networkVolumeTypes": ["HIGH_PERFORMANCE"]}, + { + "id": "EU-CZ-1", + "region": "EUROPE", + "networkVolumeTypes": ["STANDARD", "HIGH_PERFORMANCE"], + }, + {"id": "EU-SE-1", "region": "EUROPE", "networkVolumeTypes": []}, + {"id": "US-TX-3", "region": "NORTH_AMERICA", "networkVolumeTypes": ["STANDARD"]}, +] + + +def invoke(*args): + return CliRunner().invoke(cli, ["volume", *args]) + + +@pytest.fixture +def rest(): + with patch("runpod.rp_cli.volume.run_rest_request") as mock: + yield mock + + +@pytest.fixture +def listing(): + with patch("runpod.rp_cli.compute.list_all") as mock: + yield mock + + +@pytest.fixture +def datacenters(): + with patch( + "runpod.rp_cli.compute.list_datacenters", return_value=DATACENTERS + ) as mock: + yield mock + + +def test_volume_is_mounted(): + assert "volume" in cli.commands + + +class TestList: + def test_network_table(self, listing): + listing.return_value = [VOLUME] + result = invoke("list") + assert result.exit_code == 0, result.output + listing.assert_called_once_with("/v2/network-volumes", "networkVolumes") + for text in ( + "ID", + "DATA CENTER", + "TYPE", + "vol1", + "models", + "100 GB", + "EU-RO-1", + "standard", + ): + assert text in result.output + + def test_global_json(self, listing): + listing.return_value = [GLOBAL_VOLUME] + result = invoke("list", "--global", "--json") + assert result.exit_code == 0, result.output + listing.assert_called_once_with("/v2/global-volumes", "globalVolumes") + assert json.loads(result.output) == [GLOBAL_VOLUME] + + def test_global_table(self, listing): + listing.return_value = [GLOBAL_VOLUME] + result = invoke("list", "--global") + assert "gsb_abc" in result.output and "user_123" in result.output + + def test_empty(self, listing): + listing.return_value = [] + result = invoke("list") + assert result.exit_code == 0 + assert "no network volumes" in result.output + + def test_api_error(self, listing): + listing.side_effect = error.AuthenticationError("bad key") + result = invoke("list") + assert result.exit_code == 1 + assert "bad key" in result.output + + +class TestGet: + def test_fields(self, rest): + rest.return_value = VOLUME + result = invoke("get", "vol1") + assert result.exit_code == 0, result.output + rest.assert_called_once_with("GET", "/v2/network-volumes/vol1") + assert "data center" in result.output and "EU-RO-1" in result.output + assert "standard" in result.output + + def test_global_json(self, rest): + rest.return_value = GLOBAL_VOLUME + result = invoke("get", "gsb_abc", "--global", "--json") + rest.assert_called_once_with("GET", "/v2/global-volumes/gsb_abc") + assert json.loads(result.output) == GLOBAL_VOLUME + + def test_picker(self, rest, listing): + listing.return_value = [VOLUME] + rest.return_value = VOLUME + with patch("runpod.rp_cli.volume.pick", return_value="vol1") as pick: + result = invoke("get") + assert result.exit_code == 0, result.output + choices = pick.call_args.args[1] + assert choices == [("vol1", "models 100 GB EU-RO-1 standard")] + rest.assert_called_once_with("GET", "/v2/network-volumes/vol1") + + def test_missing_id_without_tty(self, rest, listing): + listing.return_value = [VOLUME] + result = invoke("get") + assert result.exit_code == 2 + assert "pass VOLUME_ID" in result.output + rest.assert_not_called() + + def test_not_found(self, rest): + rest.side_effect = error.QueryError("network volume not found", status_code=404) + result = invoke("get", "nope") + assert result.exit_code == 1 + assert "not found" in result.output + + +class TestCreate: + def test_exact_data_center(self, rest, datacenters): + rest.return_value = VOLUME + result = invoke( + "create", + "--name", + "models", + "--size", + "100", + "--data-center", + "EU-RO-1", + "--type", + "standard", + ) + assert result.exit_code == 0, result.output + datacenters.assert_not_called() + rest.assert_called_once_with( + "POST", + "/v2/network-volumes", + json={ + "name": "models", + "size": 100, + "dataCenter": "EU-RO-1", + "type": "STANDARD", + }, + ) + assert "created network volume" in result.output and "vol1" in result.output + + def test_region_narrows_by_type(self, rest, datacenters): + rest.return_value = VOLUME + with patch("runpod.rp_cli.volume.pick", return_value="EU-CZ-1") as pick: + result = invoke( + "create", + "--name", + "m", + "--size", + "10", + "--region", + "eu", + "--type", + "high_performance", + "--json", + ) + assert result.exit_code == 0, result.output + assert [value for value, _ in pick.call_args.args[1]] == ["EU-FR-1", "EU-CZ-1"] + assert rest.call_args.kwargs["json"] == { + "name": "m", + "size": 10, + "dataCenter": "EU-CZ-1", + "type": "HIGH_PERFORMANCE", + } + assert json.loads(result.output) == VOLUME + + def test_region_with_one_candidate(self, rest, datacenters): + rest.return_value = VOLUME + result = invoke("create", "--name", "m", "--size", "10", "--region", "na") + assert result.exit_code == 0, result.output + assert rest.call_args.kwargs["json"] == { + "name": "m", + "size": 10, + "dataCenter": "US-TX-3", + } + + def test_ambiguous_region_without_tty_lists_candidates(self, rest, datacenters): + result = invoke("create", "--name", "m", "--size", "10", "--region", "eu") + assert result.exit_code == 2 + assert "EU-RO-1, EU-FR-1, EU-CZ-1" in result.output + assert "EU-SE-1" not in result.output + assert "--data-center" in result.output + rest.assert_not_called() + + def test_no_region_picks_from_every_data_center(self, rest, datacenters): + rest.return_value = VOLUME + with patch("runpod.rp_cli.volume.pick", return_value="US-TX-3") as pick: + result = invoke( + "create", "--name", "m", "--size", "10", "--type", "standard" + ) + assert result.exit_code == 0, result.output + labels = dict(pick.call_args.args[1]) + assert set(labels) == {"EU-RO-1", "EU-CZ-1", "US-TX-3"} + assert labels["EU-CZ-1"] == "EU-CZ-1 europe standard, high_performance" + + def test_region_without_type_support(self, rest, datacenters): + result = invoke( + "create", + "--name", + "m", + "--size", + "10", + "--region", + "na", + "--type", + "high_performance", + ) + assert result.exit_code == 1 + assert "no data center in region na offers high_performance" in result.output + rest.assert_not_called() + + def test_unknown_region(self, rest, datacenters): + result = invoke("create", "--name", "m", "--size", "10", "--region", "mars") + assert result.exit_code == 1 + assert "unknown region 'mars'" in result.output + + def test_requires_size(self, rest): + result = invoke("create", "--name", "m", "--data-center", "EU-RO-1") + assert result.exit_code == 2 + assert "--size" in result.output + rest.assert_not_called() + + def test_size_bounds(self, rest): + result = invoke( + "create", "--name", "m", "--size", "5000", "--data-center", "EU-RO-1" + ) + assert result.exit_code == 2 + rest.assert_not_called() + + def test_region_and_data_center_conflict(self, rest): + result = invoke( + "create", + "--name", + "m", + "--size", + "10", + "--region", + "eu", + "--data-center", + "EU-RO-1", + ) + assert result.exit_code == 2 + assert "either --region or --data-center" in result.output + + def test_global(self, rest): + rest.return_value = GLOBAL_VOLUME + result = invoke("create", "--name", "shared", "--global") + assert result.exit_code == 0, result.output + rest.assert_called_once_with( + "POST", "/v2/global-volumes", json={"name": "shared"} + ) + assert "created global volume" in result.output and "gsb_abc" in result.output + + def test_global_rejects_network_flags(self, rest): + result = invoke( + "create", "--name", "s", "--global", "--size", "10", "--region", "eu" + ) + assert result.exit_code == 2 + assert "--size, --region do not apply to global volumes" in result.output + rest.assert_not_called() + + +class TestUpdate: + def test_rename(self, rest): + rest.return_value = dict(VOLUME, name="renamed") + result = invoke("update", "vol1", "--name", "renamed") + assert result.exit_code == 0, result.output + rest.assert_called_once_with( + "PATCH", "/v2/network-volumes/vol1", json={"name": "renamed"} + ) + assert "renamed" in result.output + + def test_rename_and_resize(self, rest): + rest.side_effect = [VOLUME, dict(VOLUME, name="r", size=200)] + result = invoke("update", "vol1", "--name", "r", "--size", "200", "--json") + assert result.exit_code == 0, result.output + assert rest.call_args.args == ("PATCH", "/v2/network-volumes/vol1") + assert rest.call_args.kwargs["json"] == {"name": "r", "size": 200} + assert json.loads(result.output)["size"] == 200 + + def test_same_size_is_allowed(self, rest): + rest.side_effect = [VOLUME, VOLUME] + result = invoke("update", "vol1", "--size", "100") + assert result.exit_code == 0, result.output + + def test_rejects_shrinking(self, rest): + rest.return_value = VOLUME + result = invoke("update", "vol1", "--size", "50") + assert result.exit_code == 2 + assert "can only grow: vol1 is 100 GB, requested 50 GB" in result.output + rest.assert_called_once_with("GET", "/v2/network-volumes/vol1") + + def test_requires_a_change(self, rest): + result = invoke("update", "vol1") + assert result.exit_code == 2 + assert "--name and/or --size" in result.output + + def test_global_rename(self, rest): + rest.return_value = dict(GLOBAL_VOLUME, name="renamed") + result = invoke("update", "gsb_abc", "--global", "--name", "renamed") + assert result.exit_code == 0, result.output + rest.assert_called_once_with( + "PATCH", "/v2/global-volumes/gsb_abc", json={"name": "renamed"} + ) + + def test_global_rejects_size(self, rest): + result = invoke("update", "gsb_abc", "--global", "--size", "20") + assert result.exit_code == 2 + assert "global volumes have no size" in result.output + rest.assert_not_called() + + +class TestExpand: + def test_absolute(self, rest): + rest.side_effect = [VOLUME, dict(VOLUME, size=250)] + result = invoke("expand", "vol1", "--size", "250") + assert result.exit_code == 0, result.output + rest.assert_called_with("PATCH", "/v2/network-volumes/vol1", json={"size": 250}) + assert "from 100 GB to 250 GB" in result.output + + def test_relative(self, rest): + rest.side_effect = [VOLUME, dict(VOLUME, size=150)] + result = invoke("expand", "vol1", "--size", "+50", "--json") + assert result.exit_code == 0, result.output + rest.assert_called_with("PATCH", "/v2/network-volumes/vol1", json={"size": 150}) + assert json.loads(result.output)["size"] == 150 + + @pytest.mark.parametrize("size", ["50", "100", "+0"]) + def test_rejects_shrinking_or_no_growth(self, rest, size): + rest.return_value = VOLUME + result = invoke("expand", "vol1", "--size", size) + assert result.exit_code == 2 + assert "network volumes can only grow: vol1 is 100 GB" in result.output + assert rest.call_count == 1 + + def test_rejects_past_maximum(self, rest): + rest.return_value = VOLUME + result = invoke("expand", "vol1", "--size", "+4000") + assert result.exit_code == 2 + assert "at most 4096 GB, requested 4100 GB" in result.output + assert rest.call_count == 1 + + @pytest.mark.parametrize("size", ["abc", "-5", "1.5"]) + def test_rejects_malformed_size(self, rest, size): + result = invoke("expand", "vol1", "--size", size) + assert result.exit_code == 2 + assert "is not a size" in result.output + rest.assert_not_called() + + +class TestDelete: + def test_yes_skips_confirmation(self, rest): + result = invoke("delete", "vol1", "--yes") + assert result.exit_code == 0, result.output + rest.assert_called_once_with("DELETE", "/v2/network-volumes/vol1") + assert "deleted network volume" in result.output + + def test_confirms_interactively(self, rest): + rest.return_value = VOLUME + with patch("runpod.rp_cli.output.interactive", return_value=True): + result = CliRunner().invoke(cli, ["volume", "delete", "vol1"], input="y\n") + assert result.exit_code == 0, result.output + assert "models 100 GB EU-RO-1 standard" in result.output + rest.assert_called_with("DELETE", "/v2/network-volumes/vol1") + + def test_declined(self, rest): + rest.return_value = VOLUME + with patch("runpod.rp_cli.output.interactive", return_value=True): + result = CliRunner().invoke(cli, ["volume", "delete", "vol1"], input="n\n") + assert result.exit_code == 1 + assert all(call.args[0] == "GET" for call in rest.call_args_list) + + def test_requires_yes_without_tty(self, rest): + rest.return_value = VOLUME + result = invoke("delete", "vol1") + assert result.exit_code == 2 + assert "--yes" in result.output + assert all(call.args[0] == "GET" for call in rest.call_args_list) + + def test_global(self, rest): + result = invoke("delete", "gsb_abc", "--global", "--yes") + assert result.exit_code == 0, result.output + rest.assert_called_once_with("DELETE", "/v2/global-volumes/gsb_abc") + + def test_mounted_global_volume(self, rest): + rest.side_effect = error.QueryError("volume is still mounted", status_code=409) + result = invoke("delete", "gsb_abc", "--global", "--yes") + assert result.exit_code == 1 + assert "still mounted" in result.output + + +class TestCompletion: + def _complete(self, args, incomplete): + from click.shell_completion import ShellComplete + + return [ + item.value + for item in ShellComplete(cli, {}, "rp", "_RP_COMPLETE").get_completions( + args, incomplete + ) + ] + + def test_network_ids(self, listing): + listing.return_value = [VOLUME, dict(VOLUME, id="other")] + assert self._complete(["volume", "get"], "v") == ["vol1"] + listing.assert_called_with("/v2/network-volumes", "networkVolumes") + + def test_global_ids(self, listing): + listing.return_value = [GLOBAL_VOLUME] + assert self._complete(["volume", "delete", "--global"], "") == ["gsb_abc"] + listing.assert_called_with("/v2/global-volumes", "globalVolumes") + + def test_errors_complete_nothing(self, listing): + listing.side_effect = error.AuthenticationError("no key") + assert self._complete(["volume", "update"], "") == [] From 0e98c3bb4592d06c29db205928ee6dfea7d4c089 Mon Sep 17 00:00:00 2001 From: zeke <40004347+KAJdev@users.noreply.github.com> Date: Mon, 19 Oct 2026 15:01:01 -0700 Subject: [PATCH 2/4] fix: keep global volume ids whole in rp volume list --- runpod/rp_cli/volume.py | 11 ++++++++--- tests/test_cli/test_volume_command.py | 21 ++++++++++++++++++--- 2 files changed, 26 insertions(+), 6 deletions(-) diff --git a/runpod/rp_cli/volume.py b/runpod/rp_cli/volume.py index 43be7cc51..f28212e4d 100644 --- a/runpod/rp_cli/volume.py +++ b/runpod/rp_cli/volume.py @@ -49,8 +49,7 @@ def _columns(kind: Kind): return [ ("ID", lambda row: row["id"]), ("NAME", lambda row: row.get("name")), - ("CREATED", lambda row: row.get("createdAt")), - ("CREATED BY", lambda row: row.get("createdByUserId")), + ("CREATED", lambda row: (row.get("createdAt") or "")[:10] or None), ] return [ ("ID", lambda row: row["id"]), @@ -89,7 +88,13 @@ def _show(kind: Kind, record: dict, as_json: bool) -> None: if as_json: print_json(record) return - print_fields(record, [(header.lower(), fn) for header, fn in _columns(kind)]) + fields = [(header.lower(), fn) for header, fn in _columns(kind)] + if kind is GLOBAL: + fields[2:] = [ + ("created", lambda row: row.get("createdAt")), + ("created by", lambda row: row.get("createdByUserId")), + ] + print_fields(record, fields) def _data_center( diff --git a/tests/test_cli/test_volume_command.py b/tests/test_cli/test_volume_command.py index 0aa0c8ad2..f66441a4f 100644 --- a/tests/test_cli/test_volume_command.py +++ b/tests/test_cli/test_volume_command.py @@ -88,10 +88,18 @@ def test_global_json(self, listing): listing.assert_called_once_with("/v2/global-volumes", "globalVolumes") assert json.loads(result.output) == [GLOBAL_VOLUME] - def test_global_table(self, listing): - listing.return_value = [GLOBAL_VOLUME] + def test_global_table_keeps_long_ids_whole(self, listing): + listing.return_value = [ + dict( + GLOBAL_VOLUME, + id="cmvfsn771000007jp15tn8esm", + name="capy-test-global-renamed", + ) + ] result = invoke("list", "--global") - assert "gsb_abc" in result.output and "user_123" in result.output + assert "cmvfsn771000007jp15tn8esm" in result.output + assert "2026-09-24" in result.output + assert "user_123" not in result.output def test_empty(self, listing): listing.return_value = [] @@ -115,6 +123,13 @@ def test_fields(self, rest): assert "data center" in result.output and "EU-RO-1" in result.output assert "standard" in result.output + def test_global_fields(self, rest): + rest.return_value = GLOBAL_VOLUME + result = invoke("get", "gsb_abc", "--global") + assert result.exit_code == 0, result.output + assert "2026-09-24T00:00:00Z" in result.output + assert "created by" in result.output and "user_123" in result.output + def test_global_json(self, rest): rest.return_value = GLOBAL_VOLUME result = invoke("get", "gsb_abc", "--global", "--json") From 139713ab3cd3a38c66233e6be0d3f1aa2824521c Mon Sep 17 00:00:00 2001 From: zeke <40004347+KAJdev@users.noreply.github.com> Date: Mon, 19 Oct 2026 15:48:34 -0700 Subject: [PATCH 3/4] refactor: simplify volume commands and shared API integration --- runpod/rp_cli/volume.py | 330 +++++-------- tests/test_cli/test_volume_command.py | 687 +++++++++++--------------- 2 files changed, 409 insertions(+), 608 deletions(-) diff --git a/runpod/rp_cli/volume.py b/runpod/rp_cli/volume.py index f28212e4d..61898d79c 100644 --- a/runpod/rp_cli/volume.py +++ b/runpod/rp_cli/volume.py @@ -1,14 +1,13 @@ """`rp volume`: network and global volume management.""" import re -from typing import NamedTuple, Optional import click from runpod.api.rest import run_rest_request -from runpod.rp_cli import compute +from runpod.rp_cli import compute, console as ui +from runpod.rp_cli.api import api_errors, list_all from runpod.rp_cli.output import ( - api_errors, completer, confirm, json_option, @@ -18,104 +17,69 @@ print_table, ) -MIN_SIZE = 10 -MAX_SIZE = 4096 - - -class Kind(NamedTuple): - path: str - collection: str - noun: str - - -NETWORK = Kind("/v2/network-volumes", "networkVolumes", "network volume") -GLOBAL = Kind("/v2/global-volumes", "globalVolumes", "global volume") - - -def _kind(global_volume: bool) -> Kind: - return GLOBAL if global_volume else NETWORK - - -def _type_label(record: dict) -> Optional[str]: - return (record.get("type") or "").lower() or None +NETWORK_FIELDS = [ + ("id", lambda row: row["id"]), + ("name", lambda row: row["name"]), + ("size", lambda row: f"{row['size']} GB"), + ("data center", lambda row: row["dataCenter"]), + ("type", lambda row: row["type"].lower()), +] +GLOBAL_FIELDS = [ + ("id", lambda row: row["id"]), + ("name", lambda row: row["name"]), + ("created", lambda row: row["createdAt"]), + ("created by", lambda row: row["createdByUserId"]), +] +global_option = click.option( + "--global", + "kind", + flag_value="global", + default="network", + help="Manage global volumes, which mount in any data center.", +) -def _size_label(record: dict) -> Optional[str]: - return f"{record['size']} GB" if record.get("size") else None +def _label(record): + if "size" not in record: + return record["name"] + return ( + f"{record['name']} {record['size']} GB " + f"{record['dataCenter']} {record['type'].lower()}" + ) -def _columns(kind: Kind): - if kind is GLOBAL: - return [ - ("ID", lambda row: row["id"]), - ("NAME", lambda row: row.get("name")), - ("CREATED", lambda row: (row.get("createdAt") or "")[:10] or None), - ] +def _choices(kind): return [ - ("ID", lambda row: row["id"]), - ("NAME", lambda row: row.get("name")), - ("SIZE", _size_label), - ("DATA CENTER", lambda row: row.get("dataCenter")), - ("TYPE", _type_label), + (row["id"], _label(row)) + for row in list_all(f"/v2/{kind}-volumes", f"{kind}Volumes") ] -def _list(kind: Kind) -> list[dict]: - return compute.list_all(kind.path, kind.collection) - - -def _get(kind: Kind, volume_id: str) -> dict: - return run_rest_request("GET", f"{kind.path}/{volume_id}") or {} - - -def _label(kind: Kind, record: dict) -> str: - if kind is GLOBAL: - return record.get("name") or "" - details = [_size_label(record), record.get("dataCenter"), _type_label(record)] - return " ".join([record.get("name") or ""] + [d for d in details if d]) - - -def _choices(kind: Kind) -> list[tuple[str, str]]: - return [(record["id"], _label(kind, record)) for record in _list(kind)] - - def _complete_ids(ctx, param, incomplete): - kind = _kind(bool(ctx.params.get("global_volume"))) - return completer(lambda: _choices(kind))(ctx, param, incomplete) + return completer(lambda: _choices(ctx.params.get("kind", "network")))( + ctx, param, incomplete + ) -def _show(kind: Kind, record: dict, as_json: bool) -> None: +def _show(record, kind, as_json, action=None): if as_json: print_json(record) return - fields = [(header.lower(), fn) for header, fn in _columns(kind)] - if kind is GLOBAL: - fields[2:] = [ - ("created", lambda row: row.get("createdAt")), - ("created by", lambda row: row.get("createdByUserId")), - ] - print_fields(record, fields) - - -def _data_center( - regions: tuple[str, ...], data_center: Optional[str], volume_type: Optional[str] -) -> str: - """the single data center a network volume is created in. - - regions narrow to data centers offering the requested volume type - (or any type), then a person picks one; scripts get the candidates. - """ - if data_center: - return data_center + if action: + ui.success(f"{action} {kind} volume [white]{record['id']}[/white]") + print_fields(record, GLOBAL_FIELDS if kind == "global" else NETWORK_FIELDS) + + +def _data_center(regions, volume_type): datacenters = compute.list_datacenters() if regions: - allowed = set(compute.resolve_regions(regions, datacenters)) + allowed = compute.resolve_regions(regions, datacenters) datacenters = [dc for dc in datacenters if dc["id"] in allowed] candidates = [ dc for dc in datacenters - if (volume_type in (dc.get("networkVolumeTypes") or [])) - or (not volume_type and dc.get("networkVolumeTypes")) + if dc["networkVolumeTypes"] + and (not volume_type or volume_type in dc["networkVolumeTypes"]) ] scope = f"region {', '.join(regions)}" if regions else "the catalog" offered = ( @@ -125,51 +89,41 @@ def _data_center( raise compute.ResolutionError(f"no data center in {scope} offers {offered}") if len(candidates) == 1: return candidates[0]["id"] - ids = ", ".join(dc["id"] for dc in candidates) return pick( "data center", [ ( dc["id"], - f"{dc['id']} {(dc.get('region') or '').lower().replace('_', ' ')} " + f"{dc['id']} {dc['region'].lower().replace('_', ' ')} " + ", ".join(t.lower() for t in dc["networkVolumeTypes"]), ) for dc in candidates ], - f"several data centers in {scope} offer {offered} ({ids}). " - "pass one with --data-center", + f"several data centers in {scope} offer {offered} " + f"({', '.join(dc['id'] for dc in candidates)}). pass one with --data-center", ) -def _parse_growth(value: str) -> tuple[bool, int]: - """split an expand size into (relative, gigabytes): N or +N.""" - text = value.strip() +def _growth(volume_id, size, expand=False): + """resolve and validate a size against the current network volume.""" + text = str(size).strip() if not re.fullmatch(r"\+?\d+", text): raise click.BadParameter( - f"'{value}' is not a size. use N for a total or +N to add N GB", + f"'{size}' is not a size. use N for a total or +N to add N GB", param_hint="--size", ) - return text.startswith("+"), int(text.lstrip("+")) - - -def _check_growth(volume_id: str, current: int, size: int, strict: bool) -> None: - if size < current or (strict and size == current): + current = run_rest_request("GET", f"/v2/network-volumes/{volume_id}")["size"] + target = current + int(text) if text.startswith("+") else int(text) + if target < current or (expand and target == current): raise click.UsageError( f"network volumes can only grow: {volume_id} is {current} GB, " - f"requested {size} GB" + f"requested {target} GB" ) - if size > MAX_SIZE: + if target > 4096: raise click.UsageError( - f"network volumes hold at most {MAX_SIZE} GB, requested {size} GB" + f"network volumes hold at most 4096 GB, requested {target} GB" ) - - -global_option = click.option( - "--global", - "global_volume", - is_flag=True, - help="Manage global volumes, which mount in any data center.", -) + return current, target @click.group() @@ -180,16 +134,20 @@ def volume(): @volume.command(name="list") @global_option @json_option -def volume_list(global_volume, as_json): +def volume_list(kind, as_json): """List volumes with their size, data center, and type.""" - kind = _kind(global_volume) with api_errors(): - rows = _list(kind) + rows = list_all(f"/v2/{kind}-volumes", f"{kind}Volumes") if as_json: print_json(rows) return + fields = NETWORK_FIELDS + if kind == "global": + fields = GLOBAL_FIELDS[:2] + [("created", lambda row: row["createdAt"][:10])] print_table( - rows, _columns(kind), f"no {kind.noun}s. create one with rp volume create" + rows, + [(label.upper(), value) for label, value in fields], + f"no {kind} volumes. create one with rp volume create", ) @@ -197,23 +155,21 @@ def volume_list(global_volume, as_json): @click.argument("volume_id", required=False, shell_complete=_complete_ids) @global_option @json_option -def volume_get(volume_id, global_volume, as_json): +def volume_get(volume_id, kind, as_json): """Show one volume (pick one when VOLUME_ID is omitted).""" - kind = _kind(global_volume) with api_errors(): if volume_id is None: - volume_id = pick(kind.noun, _choices(kind), "pass VOLUME_ID") - record = _get(kind, volume_id) - _show(kind, record, as_json) + volume_id = pick(f"{kind} volume", _choices(kind), "pass VOLUME_ID") + record = run_rest_request("GET", f"/v2/{kind}-volumes/{volume_id}") + _show(record, kind, as_json) @volume.command(name="create") @click.option("--name", required=True, help="Human-readable name.") @click.option( "--size", - type=click.IntRange(MIN_SIZE, MAX_SIZE), - default=None, - help=f"Storage in GB ({MIN_SIZE}-{MAX_SIZE}). Required for network volumes.", + type=click.IntRange(10, 4096), + help="Storage in GB (10-4096). Required for network volumes.", ) @click.option( "--region", @@ -221,20 +177,17 @@ def volume_get(volume_id, global_volume, as_json): multiple=True, help="Region (eu, na), id prefix (us, eu-ro), or data center id. Repeatable.", ) -@click.option("--data-center", default=None, help="Exact data center id, like EU-RO-1.") +@click.option("--data-center", help="Exact data center id, like EU-RO-1.") @click.option( "--type", "volume_type", type=click.Choice(["standard", "high_performance"], case_sensitive=False), - default=None, help="Storage tier, immutable after creation. Defaults to the data center's " "primary tier.", ) @global_option @json_option -def volume_create( - name, size, regions, data_center, volume_type, global_volume, as_json -): # pylint: disable=too-many-arguments +def volume_create(name, size, regions, data_center, volume_type, kind, as_json): """Create a network volume in one data center, or a global volume. A region matching several data centers narrows to those offering @@ -245,82 +198,65 @@ def volume_create( rp volume create --name scratch --size 50 --data-center EU-RO-1 rp volume create --name shared --global """ - from runpod.rp_cli import console as ui - - kind = _kind(global_volume) - if global_volume: - given = [ - flag - for flag, value in ( - ("--size", size), - ("--region", regions), - ("--data-center", data_center), - ("--type", volume_type), - ) - if value - ] - if given: - raise click.UsageError(f"{', '.join(given)} do not apply to global volumes") - body = {"name": name} - else: - if size is None: - raise click.UsageError("pass --size in GB for a network volume") - if regions and data_center: - raise click.UsageError("use either --region or --data-center") - volume_type = volume_type.upper() if volume_type else None - with api_errors(): - body = { - "name": name, - "size": size, - "dataCenter": _data_center(regions, data_center, volume_type), - } - if volume_type: - body["type"] = volume_type - + body = {"name": name} with api_errors(): - record = run_rest_request("POST", kind.path, json=body) or {} - if as_json: - print_json(record) - return - ui.success(f"created {kind.noun} [white]{record.get('id')}[/white]") - _show(kind, record, as_json) + if kind == "global": + given = [ + flag + for flag, value in ( + ("--size", size), + ("--region", regions), + ("--data-center", data_center), + ("--type", volume_type), + ) + if value + ] + if given: + raise click.UsageError( + f"{', '.join(given)} do not apply to global volumes" + ) + else: + if size is None: + raise click.UsageError("pass --size in GB for a network volume") + if regions and data_center: + raise click.UsageError("use either --region or --data-center") + volume_type = volume_type.upper() if volume_type else None + body.update( + size=size, + dataCenter=data_center or _data_center(regions, volume_type), + ) + if volume_type: + body["type"] = volume_type + record = run_rest_request("POST", f"/v2/{kind}-volumes", json=body) + _show(record, kind, as_json, "created") @volume.command(name="update") @click.argument("volume_id", shell_complete=_complete_ids) -@click.option("--name", default=None, help="New name.") +@click.option("--name", help="New name.") @click.option( "--size", - type=click.IntRange(MIN_SIZE, MAX_SIZE), - default=None, + type=click.IntRange(10, 4096), help="New size in GB; network volumes can only grow.", ) @global_option @json_option -def volume_update(volume_id, name, size, global_volume, as_json): +def volume_update(volume_id, name, size, kind, as_json): """Rename and/or resize a volume.""" - from runpod.rp_cli import console as ui - - kind = _kind(global_volume) - if global_volume and size is not None: + if kind == "global" and size is not None: raise click.UsageError("global volumes have no size to change") - if name is None and size is None: - raise click.UsageError("pass --name and/or --size") body = { key: value for key, value in (("name", name), ("size", size)) if value is not None } + if not body: + raise click.UsageError("pass --name and/or --size") with api_errors(): if size is not None: - current = _get(kind, volume_id).get("size") or 0 - _check_growth(volume_id, current, size, strict=False) - record = run_rest_request("PATCH", f"{kind.path}/{volume_id}", json=body) or {} - if as_json: - print_json(record) - return - ui.success(f"updated {kind.noun} [white]{volume_id}[/white]") - _show(kind, record, as_json) + _growth(volume_id, size) + record = run_rest_request("PATCH", f"/v2/{kind}-volumes/{volume_id}", json=body) + _show(record, kind, as_json, "updated") @volume.command(name="expand") @@ -334,44 +270,32 @@ def volume_update(volume_id, name, size, global_volume, as_json): @json_option def volume_expand(volume_id, size, as_json): """Grow a network volume; storage can never shrink.""" - from runpod.rp_cli import console as ui - - relative, amount = _parse_growth(size) with api_errors(): - current = _get(NETWORK, volume_id).get("size") or 0 - target = current + amount if relative else amount - _check_growth(volume_id, current, target, strict=True) - record = ( - run_rest_request( - "PATCH", f"{NETWORK.path}/{volume_id}", json={"size": target} - ) - or {} + current, target = _growth(volume_id, size, expand=True) + record = run_rest_request( + "PATCH", f"/v2/network-volumes/{volume_id}", json={"size": target} ) if as_json: print_json(record) - return - ui.success( - f"expanded [white]{volume_id}[/white] from {current} GB to " - f"{record.get('size', target)} GB" - ) + else: + ui.success( + f"expanded [white]{volume_id}[/white] from {current} GB to {record['size']} GB" + ) @volume.command(name="delete") @click.argument("volume_id", shell_complete=_complete_ids) @click.option("--yes", "-y", is_flag=True, help="Skip the confirmation prompt.") @global_option -def volume_delete(volume_id, yes, global_volume): +def volume_delete(volume_id, yes, kind): """Delete a volume and everything stored on it.""" - from runpod.rp_cli import console as ui - - kind = _kind(global_volume) + path = f"/v2/{kind}-volumes/{volume_id}" with api_errors(): if not yes: - record = _get(kind, volume_id) + record = run_rest_request("GET", path) confirm( - f"delete {kind.noun} {volume_id} ({_label(kind, record)}) " - "and all its data?", + f"delete {kind} volume {volume_id} ({_label(record)}) and all its data?", yes, ) - run_rest_request("DELETE", f"{kind.path}/{volume_id}") - ui.success(f"deleted {kind.noun} [white]{volume_id}[/white]") + run_rest_request("DELETE", path) + ui.success(f"deleted {kind} volume [white]{volume_id}[/white]") diff --git a/tests/test_cli/test_volume_command.py b/tests/test_cli/test_volume_command.py index f66441a4f..5faa4a292 100644 --- a/tests/test_cli/test_volume_command.py +++ b/tests/test_cli/test_volume_command.py @@ -1,9 +1,10 @@ """tests for the rp volume command group.""" import json -from unittest.mock import patch +from unittest.mock import call, patch import pytest +from click.shell_completion import ShellComplete from click.testing import CliRunner from runpod import error @@ -17,10 +18,10 @@ "type": "STANDARD", } GLOBAL_VOLUME = { - "id": "gsb_abc", - "name": "shared", + "id": "cmvfsn771000007jp15tn8esm", + "name": "capy-test-global-renamed", "createdAt": "2026-09-24T00:00:00Z", - "createdByUserId": "user_123", + "createdByUserId": "user_example", } DATACENTERS = [ {"id": "EU-RO-1", "region": "EUROPE", "networkVolumeTypes": ["STANDARD"]}, @@ -35,19 +36,19 @@ ] -def invoke(*args): - return CliRunner().invoke(cli, ["volume", *args]) +def invoke(command, **kwargs): + return CliRunner().invoke(cli, f"volume {command}", **kwargs) @pytest.fixture def rest(): - with patch("runpod.rp_cli.volume.run_rest_request") as mock: + with patch("runpod.rp_cli.volume.run_rest_request", return_value=VOLUME) as mock: yield mock @pytest.fixture def listing(): - with patch("runpod.rp_cli.compute.list_all") as mock: + with patch("runpod.rp_cli.volume.list_all", return_value=[VOLUME]) as mock: yield mock @@ -59,406 +60,282 @@ def datacenters(): yield mock -def test_volume_is_mounted(): - assert "volume" in cli.commands - - -class TestList: - def test_network_table(self, listing): - listing.return_value = [VOLUME] - result = invoke("list") - assert result.exit_code == 0, result.output - listing.assert_called_once_with("/v2/network-volumes", "networkVolumes") - for text in ( - "ID", - "DATA CENTER", - "TYPE", - "vol1", - "models", - "100 GB", - "EU-RO-1", - "standard", - ): - assert text in result.output +@pytest.fixture(params=["network", "global"]) +def resource(request, rest): + kind = request.param + record = GLOBAL_VOLUME if kind == "global" else VOLUME + rest.return_value = record + return kind, record, "--global" if kind == "global" else "" + - def test_global_json(self, listing): - listing.return_value = [GLOBAL_VOLUME] - result = invoke("list", "--global", "--json") - assert result.exit_code == 0, result.output - listing.assert_called_once_with("/v2/global-volumes", "globalVolumes") - assert json.loads(result.output) == [GLOBAL_VOLUME] - - def test_global_table_keeps_long_ids_whole(self, listing): - listing.return_value = [ - dict( - GLOBAL_VOLUME, - id="cmvfsn771000007jp15tn8esm", - name="capy-test-global-renamed", - ) - ] - result = invoke("list", "--global") - assert "cmvfsn771000007jp15tn8esm" in result.output +def test_list_table_and_json(resource, listing): + kind, record, flag = resource + listing.return_value = [record] + result = invoke(f"list {flag}") + assert result.exit_code == 0, result.output + assert record["id"] in result.output + if kind == "global": assert "2026-09-24" in result.output - assert "user_123" not in result.output - - def test_empty(self, listing): - listing.return_value = [] - result = invoke("list") - assert result.exit_code == 0 - assert "no network volumes" in result.output - - def test_api_error(self, listing): - listing.side_effect = error.AuthenticationError("bad key") - result = invoke("list") - assert result.exit_code == 1 - assert "bad key" in result.output - - -class TestGet: - def test_fields(self, rest): - rest.return_value = VOLUME - result = invoke("get", "vol1") - assert result.exit_code == 0, result.output - rest.assert_called_once_with("GET", "/v2/network-volumes/vol1") - assert "data center" in result.output and "EU-RO-1" in result.output - assert "standard" in result.output - - def test_global_fields(self, rest): - rest.return_value = GLOBAL_VOLUME - result = invoke("get", "gsb_abc", "--global") - assert result.exit_code == 0, result.output - assert "2026-09-24T00:00:00Z" in result.output - assert "created by" in result.output and "user_123" in result.output - - def test_global_json(self, rest): - rest.return_value = GLOBAL_VOLUME - result = invoke("get", "gsb_abc", "--global", "--json") - rest.assert_called_once_with("GET", "/v2/global-volumes/gsb_abc") - assert json.loads(result.output) == GLOBAL_VOLUME - - def test_picker(self, rest, listing): - listing.return_value = [VOLUME] - rest.return_value = VOLUME - with patch("runpod.rp_cli.volume.pick", return_value="vol1") as pick: - result = invoke("get") - assert result.exit_code == 0, result.output - choices = pick.call_args.args[1] - assert choices == [("vol1", "models 100 GB EU-RO-1 standard")] - rest.assert_called_once_with("GET", "/v2/network-volumes/vol1") - - def test_missing_id_without_tty(self, rest, listing): - listing.return_value = [VOLUME] + assert record["createdByUserId"] not in result.output + else: + for text in ("ID", "DATA CENTER", "TYPE", "100 GB", "EU-RO-1", "standard"): + assert text in result.output + result = invoke(f"list {flag} --json") + assert json.loads(result.output) == [record] + listing.assert_called_with(f"/v2/{kind}-volumes", f"{kind}Volumes") + + +def test_list_empty(listing): + listing.return_value = [] + assert "no network volumes" in invoke("list").output + + +def test_get_fields_and_json(resource, rest): + kind, record, flag = resource + result = invoke(f"get {record['id']} {flag}") + assert result.exit_code == 0, result.output + for value in record.values(): + assert str(value).lower() in result.output.lower() + result = invoke(f"get {record['id']} {flag} --json") + assert json.loads(result.output) == record + rest.assert_called_with("GET", f"/v2/{kind}-volumes/{record['id']}") + + +def test_get_picker(rest, listing): + with patch("runpod.rp_cli.volume.pick", return_value="vol1") as pick: result = invoke("get") - assert result.exit_code == 2 - assert "pass VOLUME_ID" in result.output - rest.assert_not_called() - - def test_not_found(self, rest): - rest.side_effect = error.QueryError("network volume not found", status_code=404) - result = invoke("get", "nope") - assert result.exit_code == 1 - assert "not found" in result.output - - -class TestCreate: - def test_exact_data_center(self, rest, datacenters): - rest.return_value = VOLUME - result = invoke( - "create", - "--name", - "models", - "--size", - "100", - "--data-center", - "EU-RO-1", - "--type", - "standard", - ) - assert result.exit_code == 0, result.output - datacenters.assert_not_called() - rest.assert_called_once_with( - "POST", - "/v2/network-volumes", - json={ - "name": "models", - "size": 100, - "dataCenter": "EU-RO-1", - "type": "STANDARD", - }, - ) - assert "created network volume" in result.output and "vol1" in result.output - - def test_region_narrows_by_type(self, rest, datacenters): - rest.return_value = VOLUME - with patch("runpod.rp_cli.volume.pick", return_value="EU-CZ-1") as pick: - result = invoke( - "create", - "--name", - "m", - "--size", - "10", - "--region", - "eu", - "--type", - "high_performance", - "--json", - ) - assert result.exit_code == 0, result.output - assert [value for value, _ in pick.call_args.args[1]] == ["EU-FR-1", "EU-CZ-1"] - assert rest.call_args.kwargs["json"] == { - "name": "m", - "size": 10, - "dataCenter": "EU-CZ-1", - "type": "HIGH_PERFORMANCE", - } - assert json.loads(result.output) == VOLUME - - def test_region_with_one_candidate(self, rest, datacenters): - rest.return_value = VOLUME - result = invoke("create", "--name", "m", "--size", "10", "--region", "na") - assert result.exit_code == 0, result.output - assert rest.call_args.kwargs["json"] == { + assert result.exit_code == 0, result.output + assert pick.call_args.args[1] == [("vol1", "models 100 GB EU-RO-1 standard")] + rest.assert_called_once_with("GET", "/v2/network-volumes/vol1") + + +def test_missing_id_requires_terminal(rest, listing): + result = invoke("get") + assert result.exit_code == 2 + assert "pass VOLUME_ID" in result.output + rest.assert_not_called() + + +@pytest.mark.parametrize( + "command,exception", + [ + ("get vol1", error.QueryError("not found", status_code=404)), + ("get vol1", error.AuthenticationError("bad key")), + ( + "delete vol1 --global --yes", + error.QueryError("still mounted", status_code=409), + ), + ], +) +def test_api_errors_are_clean(rest, command, exception): + rest.side_effect = exception + result = invoke(command) + assert result.exit_code == 1 + assert str(exception) in result.output + + +def test_create_exact_data_center(rest, datacenters): + result = invoke( + "create --name models --size 100 --data-center EU-RO-1 --type standard" + ) + assert result.exit_code == 0, result.output + datacenters.assert_not_called() + rest.assert_called_once_with( + "POST", + "/v2/network-volumes", + json={ + "name": "models", + "size": 100, + "dataCenter": "EU-RO-1", + "type": "STANDARD", + }, + ) + assert "created network volume" in result.output + + +@pytest.mark.parametrize( + "options,candidates", + [ + ("--region eu --type high_performance", ["EU-FR-1", "EU-CZ-1"]), + ("--type standard", ["EU-RO-1", "EU-CZ-1", "US-TX-3"]), + ("--region eu --region na", ["EU-RO-1", "EU-FR-1", "EU-CZ-1", "US-TX-3"]), + ], +) +def test_create_filters_picker_candidates(rest, datacenters, options, candidates): + with patch("runpod.rp_cli.volume.pick", return_value=candidates[0]) as pick: + result = invoke(f"create --name m --size 10 {options} --json") + assert result.exit_code == 0, result.output + assert [value for value, _ in pick.call_args.args[1]] == candidates + assert rest.call_args.kwargs["json"]["dataCenter"] == candidates[0] + assert json.loads(result.output) == VOLUME + + +def test_create_single_region_match_uses_default_tier(rest, datacenters): + result = invoke("create --name m --size 10 --region na") + assert result.exit_code == 0, result.output + rest.assert_called_once_with( + "POST", + "/v2/network-volumes", + json={ "name": "m", "size": 10, "dataCenter": "US-TX-3", - } - - def test_ambiguous_region_without_tty_lists_candidates(self, rest, datacenters): - result = invoke("create", "--name", "m", "--size", "10", "--region", "eu") - assert result.exit_code == 2 - assert "EU-RO-1, EU-FR-1, EU-CZ-1" in result.output - assert "EU-SE-1" not in result.output - assert "--data-center" in result.output - rest.assert_not_called() - - def test_no_region_picks_from_every_data_center(self, rest, datacenters): - rest.return_value = VOLUME - with patch("runpod.rp_cli.volume.pick", return_value="US-TX-3") as pick: - result = invoke( - "create", "--name", "m", "--size", "10", "--type", "standard" - ) - assert result.exit_code == 0, result.output - labels = dict(pick.call_args.args[1]) - assert set(labels) == {"EU-RO-1", "EU-CZ-1", "US-TX-3"} - assert labels["EU-CZ-1"] == "EU-CZ-1 europe standard, high_performance" - - def test_region_without_type_support(self, rest, datacenters): - result = invoke( - "create", - "--name", - "m", - "--size", - "10", - "--region", - "na", - "--type", - "high_performance", - ) - assert result.exit_code == 1 - assert "no data center in region na offers high_performance" in result.output - rest.assert_not_called() - - def test_unknown_region(self, rest, datacenters): - result = invoke("create", "--name", "m", "--size", "10", "--region", "mars") - assert result.exit_code == 1 - assert "unknown region 'mars'" in result.output - - def test_requires_size(self, rest): - result = invoke("create", "--name", "m", "--data-center", "EU-RO-1") - assert result.exit_code == 2 - assert "--size" in result.output - rest.assert_not_called() - - def test_size_bounds(self, rest): - result = invoke( - "create", "--name", "m", "--size", "5000", "--data-center", "EU-RO-1" - ) - assert result.exit_code == 2 - rest.assert_not_called() - - def test_region_and_data_center_conflict(self, rest): - result = invoke( - "create", - "--name", - "m", - "--size", - "10", - "--region", - "eu", - "--data-center", - "EU-RO-1", - ) - assert result.exit_code == 2 - assert "either --region or --data-center" in result.output - - def test_global(self, rest): - rest.return_value = GLOBAL_VOLUME - result = invoke("create", "--name", "shared", "--global") - assert result.exit_code == 0, result.output - rest.assert_called_once_with( - "POST", "/v2/global-volumes", json={"name": "shared"} - ) - assert "created global volume" in result.output and "gsb_abc" in result.output - - def test_global_rejects_network_flags(self, rest): - result = invoke( - "create", "--name", "s", "--global", "--size", "10", "--region", "eu" - ) - assert result.exit_code == 2 - assert "--size, --region do not apply to global volumes" in result.output - rest.assert_not_called() - - -class TestUpdate: - def test_rename(self, rest): - rest.return_value = dict(VOLUME, name="renamed") - result = invoke("update", "vol1", "--name", "renamed") - assert result.exit_code == 0, result.output - rest.assert_called_once_with( - "PATCH", "/v2/network-volumes/vol1", json={"name": "renamed"} - ) - assert "renamed" in result.output - - def test_rename_and_resize(self, rest): - rest.side_effect = [VOLUME, dict(VOLUME, name="r", size=200)] - result = invoke("update", "vol1", "--name", "r", "--size", "200", "--json") - assert result.exit_code == 0, result.output - assert rest.call_args.args == ("PATCH", "/v2/network-volumes/vol1") - assert rest.call_args.kwargs["json"] == {"name": "r", "size": 200} - assert json.loads(result.output)["size"] == 200 - - def test_same_size_is_allowed(self, rest): - rest.side_effect = [VOLUME, VOLUME] - result = invoke("update", "vol1", "--size", "100") - assert result.exit_code == 0, result.output - - def test_rejects_shrinking(self, rest): - rest.return_value = VOLUME - result = invoke("update", "vol1", "--size", "50") - assert result.exit_code == 2 - assert "can only grow: vol1 is 100 GB, requested 50 GB" in result.output - rest.assert_called_once_with("GET", "/v2/network-volumes/vol1") - - def test_requires_a_change(self, rest): - result = invoke("update", "vol1") - assert result.exit_code == 2 - assert "--name and/or --size" in result.output - - def test_global_rename(self, rest): - rest.return_value = dict(GLOBAL_VOLUME, name="renamed") - result = invoke("update", "gsb_abc", "--global", "--name", "renamed") - assert result.exit_code == 0, result.output - rest.assert_called_once_with( - "PATCH", "/v2/global-volumes/gsb_abc", json={"name": "renamed"} - ) - - def test_global_rejects_size(self, rest): - result = invoke("update", "gsb_abc", "--global", "--size", "20") - assert result.exit_code == 2 - assert "global volumes have no size" in result.output - rest.assert_not_called() - - -class TestExpand: - def test_absolute(self, rest): - rest.side_effect = [VOLUME, dict(VOLUME, size=250)] - result = invoke("expand", "vol1", "--size", "250") - assert result.exit_code == 0, result.output - rest.assert_called_with("PATCH", "/v2/network-volumes/vol1", json={"size": 250}) - assert "from 100 GB to 250 GB" in result.output - - def test_relative(self, rest): - rest.side_effect = [VOLUME, dict(VOLUME, size=150)] - result = invoke("expand", "vol1", "--size", "+50", "--json") - assert result.exit_code == 0, result.output - rest.assert_called_with("PATCH", "/v2/network-volumes/vol1", json={"size": 150}) - assert json.loads(result.output)["size"] == 150 - - @pytest.mark.parametrize("size", ["50", "100", "+0"]) - def test_rejects_shrinking_or_no_growth(self, rest, size): - rest.return_value = VOLUME - result = invoke("expand", "vol1", "--size", size) - assert result.exit_code == 2 - assert "network volumes can only grow: vol1 is 100 GB" in result.output - assert rest.call_count == 1 - - def test_rejects_past_maximum(self, rest): - rest.return_value = VOLUME - result = invoke("expand", "vol1", "--size", "+4000") - assert result.exit_code == 2 - assert "at most 4096 GB, requested 4100 GB" in result.output - assert rest.call_count == 1 - - @pytest.mark.parametrize("size", ["abc", "-5", "1.5"]) - def test_rejects_malformed_size(self, rest, size): - result = invoke("expand", "vol1", "--size", size) - assert result.exit_code == 2 - assert "is not a size" in result.output - rest.assert_not_called() - - -class TestDelete: - def test_yes_skips_confirmation(self, rest): - result = invoke("delete", "vol1", "--yes") - assert result.exit_code == 0, result.output - rest.assert_called_once_with("DELETE", "/v2/network-volumes/vol1") - assert "deleted network volume" in result.output - - def test_confirms_interactively(self, rest): - rest.return_value = VOLUME - with patch("runpod.rp_cli.output.interactive", return_value=True): - result = CliRunner().invoke(cli, ["volume", "delete", "vol1"], input="y\n") - assert result.exit_code == 0, result.output - assert "models 100 GB EU-RO-1 standard" in result.output - rest.assert_called_with("DELETE", "/v2/network-volumes/vol1") - - def test_declined(self, rest): - rest.return_value = VOLUME - with patch("runpod.rp_cli.output.interactive", return_value=True): - result = CliRunner().invoke(cli, ["volume", "delete", "vol1"], input="n\n") - assert result.exit_code == 1 - assert all(call.args[0] == "GET" for call in rest.call_args_list) - - def test_requires_yes_without_tty(self, rest): - rest.return_value = VOLUME - result = invoke("delete", "vol1") - assert result.exit_code == 2 - assert "--yes" in result.output - assert all(call.args[0] == "GET" for call in rest.call_args_list) - - def test_global(self, rest): - result = invoke("delete", "gsb_abc", "--global", "--yes") - assert result.exit_code == 0, result.output - rest.assert_called_once_with("DELETE", "/v2/global-volumes/gsb_abc") - - def test_mounted_global_volume(self, rest): - rest.side_effect = error.QueryError("volume is still mounted", status_code=409) - result = invoke("delete", "gsb_abc", "--global", "--yes") - assert result.exit_code == 1 - assert "still mounted" in result.output - - -class TestCompletion: - def _complete(self, args, incomplete): - from click.shell_completion import ShellComplete - - return [ - item.value - for item in ShellComplete(cli, {}, "rp", "_RP_COMPLETE").get_completions( - args, incomplete - ) - ] - - def test_network_ids(self, listing): - listing.return_value = [VOLUME, dict(VOLUME, id="other")] - assert self._complete(["volume", "get"], "v") == ["vol1"] - listing.assert_called_with("/v2/network-volumes", "networkVolumes") - - def test_global_ids(self, listing): - listing.return_value = [GLOBAL_VOLUME] - assert self._complete(["volume", "delete", "--global"], "") == ["gsb_abc"] - listing.assert_called_with("/v2/global-volumes", "globalVolumes") - - def test_errors_complete_nothing(self, listing): - listing.side_effect = error.AuthenticationError("no key") - assert self._complete(["volume", "update"], "") == [] + }, + ) + + +@pytest.mark.parametrize( + "options,code,message", + [ + ("--region eu", 2, "EU-RO-1, EU-FR-1, EU-CZ-1"), + ("--region na --type high_performance", 1, "no data center"), + ("--region mars", 1, "unknown region"), + ("--region eu --data-center EU-RO-1", 2, "either --region or --data-center"), + ("--size 5000", 2, "4096"), + ], +) +def test_create_rejects_invalid_placement_and_size( + rest, datacenters, options, code, message +): + result = invoke(f"create --name m --size 10 {options}") + assert result.exit_code == code + assert message in result.output + rest.assert_not_called() + + +def test_create_requires_network_size(rest): + result = invoke("create --name m --data-center EU-RO-1") + assert result.exit_code == 2 + assert "pass --size" in result.output + rest.assert_not_called() + + +def test_create_global(rest): + rest.return_value = GLOBAL_VOLUME + result = invoke("create --name shared --global") + assert result.exit_code == 0, result.output + rest.assert_called_once_with("POST", "/v2/global-volumes", json={"name": "shared"}) + + +@pytest.mark.parametrize( + "command,message", + [ + ("create --name s --global --size 10 --region eu", "do not apply"), + ("update vol1 --global --size 20", "global volumes have no size"), + ("update vol1", "pass --name and/or --size"), + ], +) +def test_invalid_updates_and_global_options(rest, command, message): + result = invoke(command) + assert result.exit_code == 2 + assert message in result.output + rest.assert_not_called() + + +def test_rename_without_fetching(resource, rest): + kind, record, flag = resource + result = invoke(f"update {record['id']} {flag} --name renamed") + assert result.exit_code == 0, result.output + rest.assert_called_once_with( + "PATCH", f"/v2/{kind}-volumes/{record['id']}", json={"name": "renamed"} + ) + + +@pytest.mark.parametrize( + "command,target", + [ + ("update vol1 --size 200 --name renamed", 200), + ("update vol1 --size 100", 100), + ("expand vol1 --size 250", 250), + ("expand vol1 --size +50", 150), + ], +) +def test_resize_requests(rest, command, target): + updated = dict(VOLUME, size=target) + rest.side_effect = [VOLUME, updated] + result = invoke(f"{command} --json") + assert result.exit_code == 0, result.output + assert json.loads(result.output) == updated + body = {"size": target} + if "--name" in command: + body["name"] = "renamed" + assert rest.call_args_list == [ + call("GET", "/v2/network-volumes/vol1"), + call("PATCH", "/v2/network-volumes/vol1", json=body), + ] + + +def test_expand_human_output(rest): + rest.side_effect = [VOLUME, dict(VOLUME, size=101)] + result = invoke("expand vol1 --size +1") + assert result.exit_code == 0, result.output + assert "from 100 GB to 101 GB" in result.output + + +@pytest.mark.parametrize( + "command,message", + [ + ("update vol1 --size 50", "can only grow"), + ("expand vol1 --size 50", "can only grow"), + ("expand vol1 --size +0", "can only grow"), + ("expand vol1 --size +4000", "at most 4096 GB"), + ], +) +def test_resize_rejects_shrink_no_growth_and_overflow(rest, command, message): + result = invoke(command) + assert result.exit_code == 2 + assert message in result.output + rest.assert_called_once_with("GET", "/v2/network-volumes/vol1") + + +@pytest.mark.parametrize("size", ["abc", "-5", "1.5"]) +def test_expand_rejects_malformed_size_without_fetching(rest, size): + result = invoke(f"expand vol1 --size {size}") + assert result.exit_code == 2 + assert "is not a size" in result.output + rest.assert_not_called() + + +def test_delete_yes(resource, rest): + kind, record, flag = resource + result = invoke(f"delete {record['id']} {flag} --yes") + assert result.exit_code == 0, result.output + rest.assert_called_once_with("DELETE", f"/v2/{kind}-volumes/{record['id']}") + + +@pytest.mark.parametrize("answer,code", [("y", 0), ("n", 1)]) +def test_delete_confirmation(rest, answer, code): + with patch("runpod.rp_cli.output.interactive", return_value=True): + result = invoke("delete vol1", input=f"{answer}\n") + assert result.exit_code == code + assert "models 100 GB EU-RO-1 standard" in result.output + expected = [call("GET", "/v2/network-volumes/vol1")] + if answer == "y": + expected.append(call("DELETE", "/v2/network-volumes/vol1")) + assert rest.call_args_list == expected + + +def test_delete_requires_yes_without_terminal(rest): + result = invoke("delete vol1") + assert result.exit_code == 2 + assert "pass --yes" in result.output + rest.assert_called_once_with("GET", "/v2/network-volumes/vol1") + + +def test_completion_by_kind(resource, listing): + kind, record, flag = resource + listing.return_value = [record, dict(record, id="other")] + args = ["volume", "get"] + ([flag] if flag else []) + items = ShellComplete(cli, {}, "rp", "_RP_COMPLETE").get_completions( + args, record["id"][:1] + ) + assert [item.value for item in items] == [record["id"]] + listing.assert_called_once_with(f"/v2/{kind}-volumes", f"{kind}Volumes") + + +def test_completion_suppresses_errors(listing): + listing.side_effect = error.AuthenticationError("no key") + items = ShellComplete(cli, {}, "rp", "_RP_COMPLETE").get_completions( + ["volume", "expand"], "" + ) + assert items == [] From bc5e6595ba42179e91ef20843f46e3d851746f2d Mon Sep 17 00:00:00 2001 From: zeke <40004347+KAJdev@users.noreply.github.com> Date: Mon, 19 Oct 2026 16:07:32 -0700 Subject: [PATCH 4/4] style: remove redundant volume docstrings --- runpod/rp_cli/volume.py | 3 --- tests/test_cli/test_volume_command.py | 2 -- 2 files changed, 5 deletions(-) diff --git a/runpod/rp_cli/volume.py b/runpod/rp_cli/volume.py index 61898d79c..95d2e5509 100644 --- a/runpod/rp_cli/volume.py +++ b/runpod/rp_cli/volume.py @@ -1,5 +1,3 @@ -"""`rp volume`: network and global volume management.""" - import re import click @@ -105,7 +103,6 @@ def _data_center(regions, volume_type): def _growth(volume_id, size, expand=False): - """resolve and validate a size against the current network volume.""" text = str(size).strip() if not re.fullmatch(r"\+?\d+", text): raise click.BadParameter( diff --git a/tests/test_cli/test_volume_command.py b/tests/test_cli/test_volume_command.py index 5faa4a292..40a4d1dcf 100644 --- a/tests/test_cli/test_volume_command.py +++ b/tests/test_cli/test_volume_command.py @@ -1,5 +1,3 @@ -"""tests for the rp volume command group.""" - import json from unittest.mock import call, patch