diff --git a/README.md b/README.md index fa01984..8f01a91 100644 --- a/README.md +++ b/README.md @@ -107,6 +107,36 @@ Commands: Use `dropkit --help` for detailed help on any command. +### Machine-readable output + +Read commands render formatted tables by default. For scripting and agents, +pass `--json` to emit untruncated structured data instead. Supported on +`list`, `info`, `list-ssh-keys`, and `version`: + +```bash +dropkit list --json | jq -r '.droplets[].name' +dropkit info my-droplet --json | jq -r '.ip' +dropkit list-ssh-keys --json | jq -r '.ssh_keys[].fingerprint' +``` + +- **`list --json`** — each droplet's `id`, `name`, `status`, `ip`, + `tailscale_ip`, `region`, `size`, `cost_monthly`, `in_ssh_config`, + `ssh_hostname`, and `tags`, plus a `hibernated` list and `total_monthly_cost`. +- **`info --json`** — the `list` fields plus `created_at`, `vcpus`, + `memory_mb`, `disk_gb`, `transfer_tb`, `image`, `features`, and full + `networks` (including IPv6). +- **`list-ssh-keys --json`** — `username` and an `ssh_keys` list of + `name`/`id`/`fingerprint`. +- **`version --json`** — the dropkit `version` string. + +For command execution errors, `--json` keeps `stdout` a clean JSON stream: the +error is written to `stderr` as `{"error": "..."}` and the command exits non-zero. +Argument parsing errors still use Typer's normal usage output. Numeric fields +are `null` when the value is unknown (e.g. `cost_monthly` is `null` for a size +with no price, distinct from `0` for a genuinely free resource). The list's +`total_monthly_cost` is `null` if any resource cost is unknown. Update notices +are suppressed in JSON mode. + ## Listing and filtering options At any region, size, image, or project prompt, `?` lists the available choices. diff --git a/dropkit/main.py b/dropkit/main.py index ad5561d..0186c85 100644 --- a/dropkit/main.py +++ b/dropkit/main.py @@ -46,11 +46,33 @@ # https://docs.digitalocean.com/products/snapshots/details/pricing/ SNAPSHOT_COST_PER_GB_MONTHLY = 0.06 +JSON_HELP = "Emit machine-readable JSON instead of formatted output (for scripting and agents)" + + +def emit_json(payload: Any) -> None: + """Print a payload as formatted JSON on stdout for scripting and agents.""" + sys.stdout.write(json.dumps(payload, indent=2) + "\n") + + +def emit_error(message: str, *, json_output: bool) -> None: + """Report an error, honoring JSON mode. + + In JSON mode, emit a ``{"error": ...}`` object so stdout stays a clean JSON + stream for consumers piping to ``jq``. Otherwise preserve the usual console output. + """ + if json_output: + sys.stderr.write(json.dumps({"error": message}, indent=2) + "\n") + else: + console.print(f"[red]Error: {message}[/red]") + @app.callback() -def main_callback(): +def main_callback(ctx: typer.Context): """Run before any command - checks for updates once per day.""" - check_for_updates() + # Commands with --json handle the update notice after parsing their own + # options, so it cannot contaminate a JSON response or error stream. + if ctx.invoked_subcommand not in {"list", "ls", "info", "list-ssh-keys", "version"}: + check_for_updates() # Helper functions @@ -205,7 +227,7 @@ def complete_droplet_or_snapshot_name(incomplete: str) -> list[str]: return list(dict.fromkeys(droplet_names + snapshot_names)) -def load_config_and_api() -> tuple[Config, DigitalOceanAPI]: +def load_config_and_api(*, json_output: bool = False) -> tuple[Config, DigitalOceanAPI]: """ Load configuration and create API client. @@ -216,17 +238,24 @@ def load_config_and_api() -> tuple[Config, DigitalOceanAPI]: typer.Exit: If config doesn't exist or fails to load """ if not Config.exists(): - console.print("[red]Error: Config not found. Run 'dropkit init' first.[/red]") + if json_output: + emit_error("Config not found. Run 'dropkit init' first.", json_output=True) + else: + console.print("[red]Error: Config not found. Run 'dropkit init' first.[/red]") raise typer.Exit(1) config_manager = Config() try: config_manager.load() except Exception as e: - console.print(f"[red]Error loading config: {e}[/red]") - console.print( - "[yellow]Config file may be invalid. Try running[/yellow] [cyan]dropkit init --force[/cyan]" - ) + if json_output: + emit_error(f"loading config: {e}. Try running dropkit init --force", json_output=True) + else: + console.print(f"[red]Error loading config: {e}[/red]") + console.print( + "[yellow]Config file may be invalid. Try running[/yellow] " + "[cyan]dropkit init --force[/cyan]" + ) raise typer.Exit(1) config = config_manager.config @@ -1543,7 +1572,9 @@ def setup_tailscale( return tailscale_ip -def find_user_droplet(api: DigitalOceanAPI, droplet_name: str) -> tuple[dict | None, str]: +def find_user_droplet( + api: DigitalOceanAPI, droplet_name: str, *, json_output: bool = False +) -> tuple[dict | None, str]: """ Find a droplet by name, filtered by current user's tag. @@ -1560,7 +1591,10 @@ def find_user_droplet(api: DigitalOceanAPI, droplet_name: str) -> tuple[dict | N try: username = api.get_username() except DigitalOceanAPIError as e: - console.print(f"[red]Error fetching username from DigitalOcean: {e}[/red]") + if json_output: + emit_error(f"fetching username from DigitalOcean: {e}", json_output=True) + else: + console.print(f"[red]Error fetching username from DigitalOcean: {e}[/red]") raise typer.Exit(1) # Get droplets tagged with this user @@ -1568,7 +1602,10 @@ def find_user_droplet(api: DigitalOceanAPI, droplet_name: str) -> tuple[dict | N tag_name = get_user_tag(username) droplets = api.list_droplets(tag_name=tag_name) except DigitalOceanAPIError as e: - console.print(f"[red]Error listing droplets: {e}[/red]") + if json_output: + emit_error(f"listing droplets: {e}", json_output=True) + else: + console.print(f"[red]Error listing droplets: {e}[/red]") raise typer.Exit(1) # Find droplet by name @@ -2250,22 +2287,45 @@ def create( @app.command(name="ls", hidden=True) def list_droplets( cost: bool = typer.Option(True, "--cost/--no-cost", help="Show monthly cost column"), + json_output: bool = typer.Option(False, "--json", help=JSON_HELP), ): """List droplets and hibernated snapshots tagged with owner:.""" + if not json_output: + check_for_updates() try: # Load config and API - config_manager, api = load_config_and_api() + config_manager, api = load_config_and_api(json_output=json_output) config = config_manager.config # Get username from DigitalOcean for tag filtering try: username = api.get_username() except DigitalOceanAPIError as e: - console.print(f"[red]Error fetching username from DigitalOcean: {e}[/red]") + emit_error(f"fetching username from DigitalOcean: {e}", json_output=json_output) raise typer.Exit(1) tag_name = get_user_tag(username) + if json_output: + droplets = api.list_droplets(tag_name=tag_name) + hibernated = get_user_hibernated_snapshots(api, tag_name) + droplet_records = [build_droplet_record(d, config.ssh.config_path) for d in droplets] + hibernated_records = [build_hibernated_record(s) for s in hibernated] + costs = [r["cost_monthly"] for r in (*droplet_records, *hibernated_records)] + known_costs = [value for value in costs if value is not None] + total_monthly_cost = ( + round(sum(known_costs), 2) if len(known_costs) == len(costs) else None + ) + emit_json( + { + "tag": tag_name, + "droplets": droplet_records, + "hibernated": hibernated_records, + "total_monthly_cost": total_monthly_cost, + } + ) + return + console.print(f"[dim]Fetching resources with tag: [cyan]{tag_name}[/cyan][/dim]\n") # Track total monthly cost across all resources @@ -2390,10 +2450,110 @@ def list_droplets( ) except DigitalOceanAPIError as e: - console.print(f"[red]Error: {e}[/red]") + emit_error(str(e), json_output=json_output) raise typer.Exit(1) +def build_droplet_record(droplet: dict[str, Any], ssh_config_path: str) -> dict[str, Any]: + """Extract the agent-relevant fields from a raw droplet API object. + + Missing values are represented as ``None`` rather than placeholder strings + so JSON consumers can branch on them cleanly. The one exception is ``name``, + which falls back to an empty string because it seeds the SSH hostname. + ``cost_monthly`` stays ``None`` when the size carries no price, so a genuine + ``$0`` resource is distinguishable from one with an unknown price. + """ + name = droplet.get("name", "") + + ip_address = None + for network in (droplet.get("networks") or {}).get("v4", []): + if network.get("type") == "public": + ip_address = network.get("ip_address") + break + + ssh_hostname = get_ssh_hostname(name) + in_ssh_config = host_exists(ssh_config_path, ssh_hostname) + ssh_ip = get_ssh_host_ip(ssh_config_path, ssh_hostname) + tailscale_ip = ssh_ip if ssh_ip and is_tailscale_ip(ssh_ip) else None + + price_monthly = (droplet.get("size") or {}).get("price_monthly") + + return { + "id": droplet.get("id"), + "name": name, + "status": droplet.get("status"), + "ip": ip_address, + "tailscale_ip": tailscale_ip, + "region": (droplet.get("region") or {}).get("slug"), + "size": droplet.get("size_slug"), + "cost_monthly": float(price_monthly) if price_monthly is not None else None, + "in_ssh_config": in_ssh_config, + "ssh_hostname": ssh_hostname, + "tags": droplet.get("tags", []), + } + + +def build_hibernated_record(snapshot: dict[str, Any]) -> dict[str, Any]: + """Extract the agent-relevant fields from a hibernated-snapshot API object.""" + snapshot_name = snapshot.get("name", "") + droplet_name = get_droplet_name_from_snapshot(snapshot_name) or snapshot_name + + droplet_size = None + for tag in snapshot.get("tags", []): + if tag.startswith("size:"): + droplet_size = tag.removeprefix("size:") + + raw_size_gb = snapshot.get("size_gigabytes") + size_gb = float(raw_size_gb) if raw_size_gb is not None else None + regions = snapshot.get("regions", []) + + return { + "name": droplet_name, + "snapshot_name": snapshot_name, + "droplet_size": droplet_size, + "image_size_gb": size_gb, + "region": regions[0] if regions else None, + "cost_monthly": size_gb * SNAPSHOT_COST_PER_GB_MONTHLY if size_gb is not None else None, + } + + +def build_droplet_detail(droplet: dict[str, Any], ssh_config_path: str) -> dict[str, Any]: + """Build the detailed droplet record emitted by ``info --json``. + + Extends :func:`build_droplet_record` with hardware specs, image details, + timestamps, and full network data (including IPv6). + """ + record = build_droplet_record(droplet, ssh_config_path) + size = droplet.get("size") or {} + image = droplet.get("image") or {} + record.update( + { + "created_at": droplet.get("created_at"), + "vcpus": size.get("vcpus"), + "memory_mb": size.get("memory"), + "disk_gb": size.get("disk"), + "transfer_tb": size.get("transfer"), + "image": { + "distribution": image.get("distribution"), + "name": image.get("name"), + "slug": image.get("slug"), + }, + "features": droplet.get("features", []), + "networks": droplet.get("networks", {}), + } + ) + return record + + +def build_ssh_key_record(key: dict[str, Any]) -> dict[str, Any]: + """Extract the agent-relevant fields from an SSH key API object.""" + return { + "name": key.get("name"), + "id": key.get("id"), + "fingerprint": key.get("fingerprint"), + } + + @app.command() @requires_lock("config-ssh") def config_ssh( @@ -2485,20 +2645,28 @@ def config_ssh( @app.command() -def info(droplet_name: str = typer.Argument(..., autocompletion=complete_droplet_name)): +def info( + droplet_name: str = typer.Argument(..., autocompletion=complete_droplet_name), + json_output: bool = typer.Option(False, "--json", help=JSON_HELP), +): """Show detailed information about a droplet.""" + if not json_output: + check_for_updates() try: # Load config and API - config_manager, api = load_config_and_api() + config_manager, api = load_config_and_api(json_output=json_output) config = config_manager.config # Find the droplet - console.print(f"[dim]Looking for droplet: [cyan]{droplet_name}[/cyan][/dim]\n") - droplet, username = find_user_droplet(api, droplet_name) + if not json_output: + console.print(f"[dim]Looking for droplet: [cyan]{droplet_name}[/cyan][/dim]\n") + droplet, username = find_user_droplet(api, droplet_name, json_output=json_output) if not droplet: tag = get_user_tag(username) - console.print(f"[red]Error: Droplet '{droplet_name}' not found with tag {tag}[/red]") + emit_error( + f"Droplet '{droplet_name}' not found with tag {tag}", json_output=json_output + ) raise typer.Exit(1) # Get detailed droplet info @@ -2506,6 +2674,10 @@ def info(droplet_name: str = typer.Argument(..., autocompletion=complete_droplet if droplet_id: droplet = api.get_droplet(droplet_id) + if json_output: + emit_json(build_droplet_detail(droplet, config.ssh.config_path)) + return + # Display information in a nice format console.print( Panel.fit( @@ -2626,7 +2798,7 @@ def info(droplet_name: str = typer.Argument(..., autocompletion=complete_droplet console.print(f" [dim]Manual SSH: ssh root@{public_ip}[/dim]") except DigitalOceanAPIError as e: - console.print(f"[red]Error: {e}[/red]") + emit_error(str(e), json_output=json_output) raise typer.Exit(1) @@ -4406,30 +4578,44 @@ def enable_tailscale( @app.command(name="list-ssh-keys") -def list_ssh_keys_cmd(): +def list_ssh_keys_cmd( + json_output: bool = typer.Option(False, "--json", help=JSON_HELP), +): """List SSH keys registered via dropkit. Use 'dropkit add-ssh-key' to add or import additional SSH keys. """ + if not json_output: + check_for_updates() try: # Load config and API - _, api = load_config_and_api() + _, api = load_config_and_api(json_output=json_output) # Get username from DigitalOcean for filtering try: username = api.get_username() except DigitalOceanAPIError as e: - console.print(f"[red]Error fetching username from DigitalOcean: {e}[/red]") + emit_error(f"fetching username from DigitalOcean: {e}", json_output=json_output) raise typer.Exit(1) # Fetch all SSH keys - console.print("[dim]Fetching SSH keys from DigitalOcean...[/dim]\n") + if not json_output: + console.print("[dim]Fetching SSH keys from DigitalOcean...[/dim]\n") all_keys = api.list_ssh_keys() # Filter keys registered via dropkit (prefixed with dropkit-{username}-) prefix = f"dropkit-{username}-" dropkit_keys = [key for key in all_keys if key.get("name", "").startswith(prefix)] + if json_output: + emit_json( + { + "username": username, + "ssh_keys": [build_ssh_key_record(key) for key in dropkit_keys], + } + ) + return + if not dropkit_keys: console.print( f"[yellow]No SSH keys found registered via dropkit for user: {username}[/yellow]" @@ -4457,7 +4643,7 @@ def list_ssh_keys_cmd(): console.print(f"\n[dim]Total: {len(dropkit_keys)} key(s)[/dim]") except DigitalOceanAPIError as e: - console.print(f"[red]Error: {e}[/red]") + emit_error(str(e), json_output=json_output) raise typer.Exit(1) @@ -4657,10 +4843,17 @@ def delete_ssh_key_cmd( @app.command() -def version(): +def version( + json_output: bool = typer.Option(False, "--json", help=JSON_HELP), +): """Show the version of dropkit.""" from dropkit import __version__ + if json_output: + emit_json({"version": __version__}) + return + + check_for_updates() console.print(f"dropkit version [cyan]{__version__}[/cyan]") diff --git a/tests/test_json_output.py b/tests/test_json_output.py new file mode 100644 index 0000000..9f854b9 --- /dev/null +++ b/tests/test_json_output.py @@ -0,0 +1,257 @@ +"""Tests for machine-readable --json output across read commands.""" + +import json +from unittest.mock import MagicMock, patch + +import pytest +from rich.console import Console +from typer.testing import CliRunner + +from dropkit.api import DigitalOceanAPIError +from dropkit.main import app, build_droplet_detail, build_droplet_record, build_ssh_key_record + +runner = CliRunner() + + +def make_detail_droplet() -> dict: + """Build a raw droplet API object with detail fields for info tests.""" + return { + "id": 1, + "name": "srv", + "status": "active", + "created_at": "2026-01-01T00:00:00Z", + "networks": { + "v4": [{"type": "public", "ip_address": "1.2.3.4"}], + "v6": [{"type": "public", "ip_address": "2001:db8::1"}], + }, + "region": {"slug": "nyc3"}, + "size_slug": "s-1vcpu-1gb", + "size": {"price_monthly": 6, "vcpus": 1, "memory": 1024, "disk": 25, "transfer": 1}, + "image": {"distribution": "Ubuntu", "name": "24.04", "slug": "ubuntu-24-04-x64"}, + "tags": ["owner:me"], + "features": ["monitoring"], + } + + +class TestBuildDropletDetail: + """Tests for build_droplet_detail field extraction.""" + + @patch("dropkit.main.is_tailscale_ip", return_value=False) + @patch("dropkit.main.get_ssh_host_ip", return_value=None) + @patch("dropkit.main.host_exists", return_value=True) + def test_includes_base_and_detail_fields(self, mock_exists, mock_ip, mock_ts): + detail = build_droplet_detail(make_detail_droplet(), "~/.ssh/config") + # Base fields from build_droplet_record + assert detail["ip"] == "1.2.3.4" + assert detail["in_ssh_config"] is True + # Detail-only fields + assert detail["vcpus"] == 1 + assert detail["memory_mb"] == 1024 + assert detail["disk_gb"] == 25 + assert detail["image"]["slug"] == "ubuntu-24-04-x64" + assert detail["features"] == ["monitoring"] + # Full networks include IPv6 + assert detail["networks"]["v6"][0]["ip_address"] == "2001:db8::1" + + +class TestBuildDropletRecord: + """Tests for build_droplet_record contract guarantees.""" + + @pytest.mark.parametrize( + ("size", "expected_cost"), + [({}, None), ({"price_monthly": 0}, 0.0)], + ids=["unknown-price", "free"], + ) + @patch("dropkit.main.is_tailscale_ip", return_value=False) + @patch("dropkit.main.get_ssh_host_ip", return_value=None) + @patch("dropkit.main.host_exists", return_value=False) + def test_cost_monthly_distinguishes_unknown_from_free( + self, mock_exists, mock_ip, mock_ts, size, expected_cost + ): + record = build_droplet_record({"name": "srv", "size": size}, "~/.ssh/config") + assert record["cost_monthly"] == expected_cost + + @patch("dropkit.main.is_tailscale_ip", return_value=False) + @patch("dropkit.main.get_ssh_host_ip", return_value=None) + @patch("dropkit.main.host_exists", return_value=False) + def test_explicit_null_nested_objects_do_not_raise(self, mock_exists, mock_ip, mock_ts): + # DO can serialize absent relations as null rather than omitting the key. + droplet = {"name": "srv", "size": None, "region": None, "networks": None} + record = build_droplet_record(droplet, "~/.ssh/config") + assert record["cost_monthly"] is None + assert record["region"] is None + assert record["ip"] is None + + +class TestBuildDropletDetailNullSafety: + """Tests that build_droplet_detail tolerates explicit-null nested objects.""" + + @patch("dropkit.main.is_tailscale_ip", return_value=False) + @patch("dropkit.main.get_ssh_host_ip", return_value=None) + @patch("dropkit.main.host_exists", return_value=False) + def test_null_size_and_image(self, mock_exists, mock_ip, mock_ts): + detail = build_droplet_detail({"name": "srv", "size": None, "image": None}, "~/.ssh/config") + assert detail["vcpus"] is None + assert detail["image"]["slug"] is None + + +class TestInfoJsonErrors: + """Tests that error paths keep stdout a clean JSON stream in --json mode.""" + + @patch("dropkit.main.find_user_droplet") + @patch("dropkit.main.load_config_and_api") + def test_not_found_emits_json_error_on_stderr(self, mock_load, mock_find): + mock_find.return_value = (None, "me") + mock_load.return_value = (MagicMock(), MagicMock()) + + result = runner.invoke(app, ["info", "missing", "--json"]) + + assert result.exit_code == 1 + # stdout stays clean so `... --json | jq .` never sees the error text. + assert result.stdout == "" + error = json.loads(result.stderr) + assert "missing" in error["error"] + + @pytest.mark.parametrize("failure", ["username", "listing"]) + @patch("dropkit.main.load_config_and_api") + def test_lookup_api_failure_emits_json_error(self, mock_load, failure): + api = MagicMock() + if failure == "username": + api.get_username.side_effect = DigitalOceanAPIError("account unavailable") + else: + api.get_username.return_value = "me" + api.list_droplets.side_effect = DigitalOceanAPIError("droplets unavailable") + mock_load.return_value = (MagicMock(), api) + + result = runner.invoke(app, ["info", "srv", "--json"]) + + assert result.exit_code == 1 + assert result.stdout == "" + assert "unavailable" in json.loads(result.stderr)["error"] + + +@pytest.mark.parametrize("command", [["list"], ["info", "srv"], ["list-ssh-keys"]]) +class TestJsonConfigErrors: + """The config loader must honor JSON mode before any command can emit output.""" + + @pytest.mark.parametrize( + ("config_exists", "expected_error"), + [(False, "Config not found"), (True, "invalid token")], + ids=["missing-config", "invalid-config"], + ) + def test_error_is_json_without_terminal_styling(self, command, config_exists, expected_error): + terminal_console = Console(force_terminal=True, color_system="standard") + with ( + patch("dropkit.main.console", terminal_console), + patch("dropkit.main.Config.exists", return_value=config_exists), + patch("dropkit.main.Config.load", side_effect=ValueError("invalid token")), + ): + result = runner.invoke(app, [*command, "--json"]) + + assert result.exit_code == 1 + assert result.stdout == "" + assert "\x1b" not in result.stderr + assert expected_error in json.loads(result.stderr)["error"] + + +class TestBuildSshKeyRecord: + """Tests for build_ssh_key_record field extraction.""" + + def test_extracts_name_id_fingerprint(self): + record = build_ssh_key_record({"name": "k", "id": 9, "fingerprint": "aa:bb"}) + assert record == {"name": "k", "id": 9, "fingerprint": "aa:bb"} + + def test_missing_fields_are_none(self): + assert build_ssh_key_record({}) == {"name": None, "id": None, "fingerprint": None} + + +class TestInfoJson: + """Tests for `dropkit info --json`.""" + + @patch("dropkit.main.is_tailscale_ip", return_value=False) + @patch("dropkit.main.get_ssh_host_ip", return_value=None) + @patch("dropkit.main.host_exists", return_value=True) + @patch("dropkit.main.find_user_droplet") + @patch("dropkit.main.load_config_and_api") + def test_info_json_parseable(self, mock_load, mock_find, mock_exists, mock_ip, mock_ts): + droplet = make_detail_droplet() + mock_find.return_value = (droplet, "me") + api = MagicMock() + api.get_droplet.return_value = droplet + cm = MagicMock() + cm.config.ssh.config_path = "~/.ssh/config" + mock_load.return_value = (cm, api) + + result = runner.invoke(app, ["info", "srv", "--json"]) + + assert result.exit_code == 0 + payload = json.loads(result.output) + assert payload["name"] == "srv" + assert payload["vcpus"] == 1 + + +class TestListSshKeysJson: + """Tests for `dropkit list-ssh-keys --json`.""" + + @patch("dropkit.main.load_config_and_api") + def test_only_dropkit_keys_returned(self, mock_load): + api = MagicMock() + api.get_username.return_value = "me" + api.list_ssh_keys.return_value = [ + {"name": "dropkit-me-laptop", "id": 9, "fingerprint": "aa:bb"}, + {"name": "someone-elses-key", "id": 2, "fingerprint": "cc:dd"}, + ] + mock_load.return_value = (MagicMock(), api) + + result = runner.invoke(app, ["list-ssh-keys", "--json"]) + + assert result.exit_code == 0 + payload = json.loads(result.output) + assert payload["username"] == "me" + assert [k["name"] for k in payload["ssh_keys"]] == ["dropkit-me-laptop"] + + @patch("dropkit.main.load_config_and_api") + def test_empty_keys(self, mock_load): + api = MagicMock() + api.get_username.return_value = "me" + api.list_ssh_keys.return_value = [] + mock_load.return_value = (MagicMock(), api) + + result = runner.invoke(app, ["list-ssh-keys", "--json"]) + + assert result.exit_code == 0 + assert json.loads(result.output)["ssh_keys"] == [] + + +class TestVersionJson: + """Tests for `dropkit version --json`.""" + + def test_version_json(self): + result = runner.invoke(app, ["version", "--json"]) + assert result.exit_code == 0 + assert "version" in json.loads(result.output) + + def test_json_has_no_terminal_styling(self): + terminal_console = Console(force_terminal=True, color_system="standard") + with patch("dropkit.main.console", terminal_console): + result = runner.invoke(app, ["version", "--json"]) + + assert result.exit_code == 0 + assert "\x1b" not in result.stdout + assert "version" in json.loads(result.stdout) + + @patch("dropkit.main.check_for_updates") + def test_update_notice_does_not_pollute_json(self, mock_check): + result = runner.invoke(app, ["version", "--json"]) + + assert result.exit_code == 0 + assert json.loads(result.stdout)["version"] + assert result.stderr == "" + mock_check.assert_not_called() + + @patch("dropkit.main.check_for_updates") + def test_table_mode_still_checks_for_updates(self, mock_check): + result = runner.invoke(app, ["version"]) + + assert result.exit_code == 0 + mock_check.assert_called_once_with() diff --git a/tests/test_list.py b/tests/test_list.py new file mode 100644 index 0000000..26e7358 --- /dev/null +++ b/tests/test_list.py @@ -0,0 +1,199 @@ +"""Tests for the list command, including agent-friendly JSON output.""" + +import json +from unittest.mock import MagicMock, patch + +import pytest +from typer.testing import CliRunner + +from dropkit.main import app, build_droplet_record, build_hibernated_record + +runner = CliRunner() + + +def make_droplet(name: str = "web-prod-uksouth-01") -> dict: + """Build a raw droplet API object for tests.""" + return { + "id": 123, + "name": name, + "status": "active", + "networks": { + "v4": [ + {"type": "private", "ip_address": "10.0.0.1"}, + {"type": "public", "ip_address": "164.92.1.5"}, + ] + }, + "region": {"slug": "nyc3"}, + "size_slug": "s-1vcpu-1gb", + "size": {"price_monthly": 6}, + "tags": ["owner:me", "firewall"], + } + + +class TestBuildDropletRecord: + """Tests for build_droplet_record field extraction.""" + + @patch("dropkit.main.is_tailscale_ip", return_value=True) + @patch("dropkit.main.get_ssh_host_ip", return_value="100.1.2.3") + @patch("dropkit.main.host_exists", return_value=True) + def test_extracts_public_ip_and_ssh_state(self, mock_exists, mock_ip, mock_ts): + record = build_droplet_record(make_droplet(), "~/.ssh/config") + assert record["ip"] == "164.92.1.5" + assert record["tailscale_ip"] == "100.1.2.3" + assert record["in_ssh_config"] is True + assert record["ssh_hostname"] == "dropkit.web-prod-uksouth-01" + assert record["cost_monthly"] == 6.0 + + @patch("dropkit.main.is_tailscale_ip", return_value=False) + @patch("dropkit.main.get_ssh_host_ip", return_value=None) + @patch("dropkit.main.host_exists", return_value=False) + def test_missing_values_are_none(self, mock_exists, mock_ip, mock_ts): + bare = {"id": 1, "name": "x", "status": "new", "networks": {}} + record = build_droplet_record(bare, "~/.ssh/config") + assert record["ip"] is None + assert record["tailscale_ip"] is None + assert record["region"] is None + assert record["size"] is None + assert record["in_ssh_config"] is False + + +class TestBuildHibernatedRecord: + """Tests for build_hibernated_record field extraction.""" + + def test_extracts_name_size_and_cost(self): + snapshot = { + "name": "dropkit-web-prod", + "tags": ["size:s-2vcpu-2gb"], + "size_gigabytes": 25, + "regions": ["nyc3"], + } + record = build_hibernated_record(snapshot) + assert record["name"] == "web-prod" + assert record["droplet_size"] == "s-2vcpu-2gb" + assert record["image_size_gb"] == 25.0 + assert record["region"] == "nyc3" + assert record["cost_monthly"] > 0 + + @pytest.mark.parametrize( + ("size_gb", "expected_cost"), + [(None, None), (0, 0.0)], + ids=["unknown-size", "zero-size"], + ) + def test_cost_distinguishes_unknown_from_zero_size(self, size_gb, expected_cost): + snapshot = {"name": "dropkit-web-prod", "size_gigabytes": size_gb} + record = build_hibernated_record(snapshot) + assert record["image_size_gb"] == size_gb + assert record["cost_monthly"] == expected_cost + + +class TestListJsonOutput: + """Tests for `dropkit list --json`.""" + + @patch("dropkit.main.is_tailscale_ip", return_value=True) + @patch("dropkit.main.get_ssh_host_ip", return_value="100.1.2.3") + @patch("dropkit.main.host_exists", return_value=True) + @patch("dropkit.main.get_user_hibernated_snapshots", return_value=[]) + @patch("dropkit.main.load_config_and_api") + def test_json_is_parseable_and_untruncated( + self, mock_load, mock_snaps, mock_exists, mock_ip, mock_ts + ): + mock_api = MagicMock() + mock_api.get_username.return_value = "me" + mock_api.list_droplets.return_value = [make_droplet("a-very-long-droplet-name-uksouth-01")] + mock_config_manager = MagicMock() + mock_config_manager.config.ssh.config_path = "~/.ssh/config" + mock_load.return_value = (mock_config_manager, mock_api) + + result = runner.invoke(app, ["list", "--json"]) + + assert result.exit_code == 0 + payload = json.loads(result.output) + assert payload["tag"] == "owner:me" + assert payload["droplets"][0]["name"] == "a-very-long-droplet-name-uksouth-01" + assert payload["droplets"][0]["id"] == 123 + assert payload["total_monthly_cost"] == 6.0 + assert payload["hibernated"] == [] + + @patch("dropkit.main.get_user_hibernated_snapshots", return_value=[]) + @patch("dropkit.main.load_config_and_api") + def test_json_empty(self, mock_load, mock_snaps): + mock_api = MagicMock() + mock_api.get_username.return_value = "me" + mock_api.list_droplets.return_value = [] + mock_config_manager = MagicMock() + mock_config_manager.config.ssh.config_path = "~/.ssh/config" + mock_load.return_value = (mock_config_manager, mock_api) + + result = runner.invoke(app, ["list", "--json"]) + + assert result.exit_code == 0 + payload = json.loads(result.output) + assert payload["droplets"] == [] + assert payload["total_monthly_cost"] == 0 + + @pytest.mark.parametrize( + ("droplet_price", "snapshot_size", "expected_total"), + [(None, 25, None), (6, None, None), (6, 25, 7.5)], + ids=["unknown-droplet-price", "unknown-snapshot-size", "known-costs"], + ) + @patch("dropkit.main.get_ssh_host_ip", return_value=None) + @patch("dropkit.main.host_exists", return_value=False) + @patch("dropkit.main.get_user_hibernated_snapshots") + @patch("dropkit.main.load_config_and_api") + def test_total_requires_known_costs( + self, + mock_load, + mock_snaps, + mock_exists, + mock_ip, + droplet_price, + snapshot_size, + expected_total, + ): + droplet = make_droplet() + droplet["size"] = {"price_monthly": droplet_price} + api = MagicMock() + api.get_username.return_value = "me" + api.list_droplets.return_value = [droplet] + mock_snaps.return_value = [ + {"name": "dropkit-web-prod", "size_gigabytes": snapshot_size, "regions": ["nyc3"]} + ] + cm = MagicMock() + cm.config.ssh.config_path = "~/.ssh/config" + mock_load.return_value = (cm, api) + + result = runner.invoke(app, ["list", "--json"]) + + assert result.exit_code == 0 + payload = json.loads(result.stdout) + assert payload["total_monthly_cost"] == expected_total + assert payload["droplets"][0]["cost_monthly"] == droplet_price + assert payload["hibernated"][0]["image_size_gb"] == snapshot_size + + +class TestListTableOutput: + """Adding JSON must not change the existing human-readable list.""" + + @pytest.mark.parametrize("cost", [True, False], ids=["with-cost", "without-cost"]) + @patch("dropkit.main.get_ssh_host_ip", return_value=None) + @patch("dropkit.main.host_exists", return_value=False) + @patch("dropkit.main.get_user_hibernated_snapshots", return_value=[]) + @patch("dropkit.main.load_config_and_api") + @patch("dropkit.main.check_for_updates") + def test_default_list_keeps_table_and_cost_option( + self, mock_updates, mock_load, mock_snaps, mock_exists, mock_ip, cost + ): + api = MagicMock() + api.get_username.return_value = "me" + api.list_droplets.return_value = [make_droplet("web")] + cm = MagicMock() + cm.config.ssh.config_path = "~/.ssh/config" + mock_load.return_value = (cm, api) + + result = runner.invoke(app, ["list", *([] if cost else ["--no-cost"])]) + + assert result.exit_code == 0 + assert "Droplets:" in result.stdout + assert "web" in result.stdout + assert ("$6.00/mo" in result.stdout) is cost + mock_updates.assert_called_once_with()