diff --git a/README.md b/README.md index 68b7c68..e769f21 100644 --- a/README.md +++ b/README.md @@ -107,6 +107,33 @@ 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. + +On failure, `--json` keeps `stdout` a clean JSON stream: the error is written to +`stderr` as `{"error": "..."}` and the command exits non-zero. 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). + ## Configuration Configuration files are stored in `~/.config/dropkit/`: diff --git a/dropkit/main.py b/dropkit/main.py index 7b31fa1..4d25f8f 100644 --- a/dropkit/main.py +++ b/dropkit/main.py @@ -41,11 +41,31 @@ help="Manage DigitalOcean droplets for ToB engineers", ) console = Console() +err_console = Console(stderr=True) # DigitalOcean snapshot storage rate: $0.06/GB/month # 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.""" + console.print_json(data=payload) + + +def emit_error(message: str, *, json_output: bool) -> None: + """Report an error on stderr, honoring JSON mode. + + In JSON mode, emit a ``{"error": ...}`` object so stdout stays a clean JSON + stream for consumers piping to ``jq``. Otherwise print a red message. + """ + if json_output: + err_console.print_json(data={"error": message}) + else: + err_console.print(f"[red]Error: {message}[/red]") + @app.callback() def main_callback(): @@ -2250,6 +2270,7 @@ 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:.""" try: @@ -2261,137 +2282,250 @@ def list_droplets( 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) - console.print(f"[dim]Fetching resources with tag: [cyan]{tag_name}[/cyan][/dim]\n") - - # Track total monthly cost across all resources - total_monthly_cost = 0.0 + if not json_output: + console.print(f"[dim]Fetching resources with tag: [cyan]{tag_name}[/cyan][/dim]\n") - # List droplets droplets = api.list_droplets(tag_name=tag_name) + hibernated = get_user_hibernated_snapshots(api, tag_name) - if droplets: - # Create droplets table - console.print("[bold]Droplets:[/bold]") - table = Table(show_header=True, header_style="bold cyan") - table.add_column("Name", style="white", no_wrap=True) - table.add_column("Status", style="white", no_wrap=True) - table.add_column("IP Address", style="cyan", no_wrap=True) - table.add_column("Tailscale IP", style="magenta", no_wrap=True) - table.add_column("Region", style="white", no_wrap=True) - table.add_column("Size", style="white", no_wrap=True) - if cost: - table.add_column("Cost", style="green", no_wrap=True, justify="right") - table.add_column("SSH", style="white", no_wrap=True) - - # Add rows - for droplet in droplets: - name = droplet.get("name", "N/A") - status = droplet.get("status", "N/A") - - # Get public IP - ip_address = "N/A" - v4_networks = droplet.get("networks", {}).get("v4", []) - for network in v4_networks: - if network.get("type") == "public": - ip_address = network.get("ip_address", "N/A") - break - - region = droplet.get("region", {}).get("slug", "N/A") - size = droplet.get("size_slug", "N/A") - - # Get monthly price from the droplet's size object - price_monthly = droplet.get("size", {}).get("price_monthly", 0) - total_monthly_cost += float(price_monthly) - - # Check if in SSH config and get Tailscale IP - ssh_hostname = get_ssh_hostname(name) - in_ssh_config = "✓" if host_exists(config.ssh.config_path, ssh_hostname) else "✗" - ssh_ip = get_ssh_host_ip(config.ssh.config_path, ssh_hostname) - tailscale_ip = ssh_ip if ssh_ip and is_tailscale_ip(ssh_ip) else "—" - - # Color status - if status == "active": - status_colored = f"[green]{status}[/green]" - elif status == "new": - status_colored = f"[yellow]{status}[/yellow]" - else: - status_colored = f"[red]{status}[/red]" + droplet_records = [build_droplet_record(d, config.ssh.config_path) for d in droplets] + hibernated_records = [build_hibernated_record(s) for s in hibernated] + total_monthly_cost = sum(r["cost_monthly"] or 0 for r in droplet_records) + sum( + r["cost_monthly"] or 0 for r in hibernated_records + ) - row = [name, status_colored, ip_address, tailscale_ip, region, size] - if cost: - row.append(f"${float(price_monthly):.2f}/mo") - row.append(in_ssh_config) - table.add_row(*row) + if json_output: + emit_json( + { + "tag": tag_name, + "droplets": droplet_records, + "hibernated": hibernated_records, + "total_monthly_cost": round(total_monthly_cost, 2), + } + ) + return - console.print(table) + render_droplet_list(droplet_records, hibernated_records, total_monthly_cost, cost, tag_name) - # List hibernated snapshots - hibernated = get_user_hibernated_snapshots(api, tag_name) + except DigitalOceanAPIError as e: + emit_error(str(e), json_output=json_output) + raise typer.Exit(1) - if hibernated: - if droplets: - console.print() # Spacing between tables - console.print("[bold]Hibernated:[/bold]") - snap_table = Table(show_header=True, header_style="bold cyan") - snap_table.add_column("Name", style="white", no_wrap=True) - snap_table.add_column("Droplet Size", style="white", no_wrap=True) - snap_table.add_column("Image Size", style="white", no_wrap=True) - snap_table.add_column("Region", style="white", no_wrap=True) - if cost: - snap_table.add_column("Cost", style="green", no_wrap=True, justify="right") - - for snapshot in hibernated: - snapshot_name = snapshot.get("name", "") - # Extract droplet name from snapshot name (remove "dropkit-" prefix) - droplet_name = get_droplet_name_from_snapshot(snapshot_name) or snapshot_name - - # Extract droplet size slug from size: tag - droplet_size = "N/A" - for tag in snapshot.get("tags", []): - if tag.startswith("size:"): - droplet_size = tag.removeprefix("size:") - size_gb = snapshot.get("size_gigabytes", 0) - snapshot_cost = float(size_gb) * SNAPSHOT_COST_PER_GB_MONTHLY - total_monthly_cost += snapshot_cost - regions = snapshot.get("regions", []) - region = regions[0] if regions else "N/A" +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. - row = [droplet_name, droplet_size, f"{size_gb} GB", region] - if cost: - row.append(f"${snapshot_cost:.2f}/mo") - snap_table.add_row(*row) + 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", "") - console.print(snap_table) - console.print() - console.print("[dim]Wake with: dropkit wake [/dim]") + ip_address = None + for network in (droplet.get("networks") or {}).get("v4", []): + if network.get("type") == "public": + ip_address = network.get("ip_address") + break - # Summary - if droplets or hibernated: - console.print() - parts = [] - if droplets: - parts.append(f"{len(droplets)} droplet(s)") - if hibernated: - parts.append(f"{len(hibernated)} hibernated") - summary = f"[dim]Total: {', '.join(parts)}" - if cost: - summary += f" — [green]${total_monthly_cost:.2f}/mo[/green]" - summary += "[/dim]" - console.print(summary) - else: - console.print( - f"[yellow]No droplets or hibernated snapshots found with tag: {tag_name}[/yellow]" - ) + 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:") - except DigitalOceanAPIError as e: - console.print(f"[red]Error: {e}[/red]") - raise typer.Exit(1) + size_gb = float(snapshot.get("size_gigabytes", 0)) + 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, + } + + +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"), + } + + +def _format_cost(cost_monthly: float | None) -> str: + """Format a monthly cost, showing ``N/A`` when the price is unknown.""" + return f"${cost_monthly:.2f}/mo" if cost_monthly is not None else "N/A" + + +def _render_active_droplets(droplets: list[dict[str, Any]], cost: bool) -> None: + """Render the active-droplet table.""" + status_colors = {"active": "green", "new": "yellow"} + + console.print("[bold]Droplets:[/bold]") + table = Table(show_header=True, header_style="bold cyan") + table.add_column("Name", style="white", no_wrap=True) + table.add_column("Status", style="white", no_wrap=True) + table.add_column("IP Address", style="cyan", no_wrap=True) + table.add_column("Tailscale IP", style="magenta", no_wrap=True) + table.add_column("Region", style="white", no_wrap=True) + table.add_column("Size", style="white", no_wrap=True) + if cost: + table.add_column("Cost", style="green", no_wrap=True, justify="right") + table.add_column("SSH", style="white", no_wrap=True) + + for d in droplets: + status = d["status"] or "N/A" + color = status_colors.get(status, "red") + row = [ + d["name"], + f"[{color}]{status}[/{color}]", + d["ip"] or "N/A", + d["tailscale_ip"] or "—", + d["region"] or "N/A", + d["size"] or "N/A", + ] + if cost: + row.append(_format_cost(d["cost_monthly"])) + row.append("✓" if d["in_ssh_config"] else "✗") + table.add_row(*row) + + console.print(table) + + +def _render_hibernated(hibernated: list[dict[str, Any]], cost: bool) -> None: + """Render the hibernated-snapshot table.""" + console.print("[bold]Hibernated:[/bold]") + snap_table = Table(show_header=True, header_style="bold cyan") + snap_table.add_column("Name", style="white", no_wrap=True) + snap_table.add_column("Droplet Size", style="white", no_wrap=True) + snap_table.add_column("Image Size", style="white", no_wrap=True) + snap_table.add_column("Region", style="white", no_wrap=True) + if cost: + snap_table.add_column("Cost", style="green", no_wrap=True, justify="right") + + for s in hibernated: + row = [ + s["name"], + s["droplet_size"] or "N/A", + f"{s['image_size_gb']:g} GB", + s["region"] or "N/A", + ] + if cost: + row.append(_format_cost(s["cost_monthly"])) + snap_table.add_row(*row) + + console.print(snap_table) + console.print() + console.print("[dim]Wake with: dropkit wake [/dim]") + + +def _render_list_summary( + droplets: list[dict[str, Any]], + hibernated: list[dict[str, Any]], + total_monthly_cost: float, + cost: bool, +) -> None: + """Render the trailing summary line for the droplet list.""" + parts = [] + if droplets: + parts.append(f"{len(droplets)} droplet(s)") + if hibernated: + parts.append(f"{len(hibernated)} hibernated") + summary = f"[dim]Total: {', '.join(parts)}" + if cost: + summary += f" — [green]${total_monthly_cost:.2f}/mo[/green]" + summary += "[/dim]" + console.print(summary) + + +def render_droplet_list( + droplets: list[dict[str, Any]], + hibernated: list[dict[str, Any]], + total_monthly_cost: float, + cost: bool, + tag_name: str, +) -> None: + """Render droplet and hibernated-snapshot records as Rich tables.""" + if not droplets and not hibernated: + console.print( + f"[yellow]No droplets or hibernated snapshots found with tag: {tag_name}[/yellow]" + ) + return + + if droplets: + _render_active_droplets(droplets, cost) + + if hibernated: + if droplets: + console.print() # Spacing between tables + _render_hibernated(hibernated, cost) + + console.print() + _render_list_summary(droplets, hibernated, total_monthly_cost, cost) @app.command() @@ -2485,7 +2619,10 @@ 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.""" try: # Load config and API @@ -2493,12 +2630,15 @@ def info(droplet_name: str = typer.Argument(..., autocompletion=complete_droplet config = config_manager.config # Find the droplet - console.print(f"[dim]Looking for droplet: [cyan]{droplet_name}[/cyan][/dim]\n") + 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) 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 +2646,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 +2770,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) @@ -4350,7 +4494,9 @@ 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. @@ -4363,17 +4509,27 @@ def list_ssh_keys_cmd(): 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]" @@ -4401,7 +4557,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) @@ -4601,10 +4757,16 @@ 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 + 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..6f9282a --- /dev/null +++ b/tests/test_json_output.py @@ -0,0 +1,191 @@ +"""Tests for machine-readable --json output across read commands.""" + +import json +from unittest.mock import MagicMock, patch + +from typer.testing import CliRunner + +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.""" + + @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_none_when_price_absent(self, mock_exists, mock_ip, mock_ts): + record = build_droplet_record({"name": "srv", "size": {}}, "~/.ssh/config") + assert record["cost_monthly"] is None + + @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_zero_is_preserved(self, mock_exists, mock_ip, mock_ts): + # A genuine $0 resource must stay 0.0, distinct from an unknown price. + record = build_droplet_record( + {"name": "srv", "size": {"price_monthly": 0}}, "~/.ssh/config" + ) + assert record["cost_monthly"] == 0.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_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"] + + +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) diff --git a/tests/test_list.py b/tests/test_list.py new file mode 100644 index 0000000..6cc3572 --- /dev/null +++ b/tests/test_list.py @@ -0,0 +1,120 @@ +"""Tests for the list command, including agent-friendly JSON output.""" + +import json +from unittest.mock import MagicMock, patch + +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 + + +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