From daaffd8291b44796f3af743716a90fcf51794783 Mon Sep 17 00:00:00 2001 From: tech Date: Thu, 21 May 2026 12:34:09 +0800 Subject: [PATCH] feat: add CLI config management commands (init, add, remove, list, show, config) Co-Authored-By: Claude Sonnet 4.6 --- src/autossh_mgr/cli.py | 161 ++++++++++++++++++++++++++++++++++ tests/unit/test_cli_config.py | 128 +++++++++++++++++++++++++++ 2 files changed, 289 insertions(+) create mode 100644 src/autossh_mgr/cli.py create mode 100644 tests/unit/test_cli_config.py diff --git a/src/autossh_mgr/cli.py b/src/autossh_mgr/cli.py new file mode 100644 index 0000000..b2e58fa --- /dev/null +++ b/src/autossh_mgr/cli.py @@ -0,0 +1,161 @@ +import click +from autossh_mgr.config import ( + TunnelConfig, get_config_dir, ensure_dirs, + load_tunnels, save_tunnels, get_tunnel, +) +from autossh_mgr.process import get_status +from autossh_mgr.display import ( + print_tunnel_table, print_tunnel_detail, print_status_table, +) + + +_FIELD_TYPES: dict[str, type] = { + "host": str, + "port": int, + "user": str, + "identity_file": str, + "local_host": str, + "local_port": int, + "remote_host": str, + "remote_port": int, + "ssh_options": str, +} + + +@click.group() +@click.pass_context +def cli(ctx: click.Context) -> None: + ctx.ensure_object(dict) + config_dir = get_config_dir() + ensure_dirs(config_dir) + ctx.obj["config_dir"] = config_dir + + +@cli.command() +@click.pass_context +def init(ctx: click.Context) -> None: + """Initialize config directory.""" + config_dir = ctx.obj["config_dir"] + ensure_dirs(config_dir) + click.echo(f"Initialized {config_dir}") + + +@cli.command() +@click.argument("name") +@click.option("--host", default=None) +@click.option("--port", type=int, default=None) +@click.option("--user", default=None) +@click.option("--identity-file", default=None) +@click.option("--local-host", default=None) +@click.option("--local-port", type=int, default=None) +@click.option("--remote-host", default=None) +@click.option("--remote-port", type=int, default=None) +@click.option("--ssh-options", default=None) +@click.pass_context +def add( + ctx, name, host, port, user, identity_file, + local_host, local_port, remote_host, remote_port, ssh_options, +): + """Add a new tunnel configuration.""" + config_dir = ctx.obj["config_dir"] + tunnels = load_tunnels(config_dir) + if any(t.name == name for t in tunnels): + raise click.ClickException(f"Tunnel '{name}' already exists") + + if host is None: + host = click.prompt("Public server host") + if user is None: + user = click.prompt("SSH user") + if local_port is None: + local_port = click.prompt("Local port", type=int) + if remote_port is None: + remote_port = click.prompt("Remote port", type=int) + if port is None: + port = 22 + if identity_file is None: + identity_file = "~/.ssh/id_ed25519" + if local_host is None: + local_host = "127.0.0.1" + if remote_host is None: + remote_host = "0.0.0.0" + if ssh_options is None: + ssh_options = "" + + tunnel = TunnelConfig( + name=name, host=host, port=port, user=user, + identity_file=identity_file, local_host=local_host, + local_port=local_port, remote_host=remote_host, + remote_port=remote_port, ssh_options=ssh_options, + ) + tunnels.append(tunnel) + save_tunnels(config_dir, tunnels) + click.echo(f"Added tunnel '{name}'") + + +@cli.command() +@click.argument("name") +@click.pass_context +def remove(ctx, name): + """Remove a tunnel configuration.""" + config_dir = ctx.obj["config_dir"] + get_tunnel(config_dir, name) # raises if not found + status = get_status(config_dir, name) + if status.state == "running": + raise click.ClickException(f"Stop '{name}' before removing it") + if not click.confirm(f"Remove tunnel '{name}'?"): + return + tunnels = [t for t in load_tunnels(config_dir) if t.name != name] + save_tunnels(config_dir, tunnels) + click.echo(f"Removed tunnel '{name}'") + + +@cli.command(name="list") +@click.pass_context +def list_cmd(ctx): + """List all tunnel configurations.""" + config_dir = ctx.obj["config_dir"] + tunnels = load_tunnels(config_dir) + statuses = {t.name: get_status(config_dir, t.name) for t in tunnels} + print_tunnel_table(tunnels, statuses) + + +@cli.command() +@click.argument("name") +@click.pass_context +def show(ctx, name): + """Show full details for a tunnel.""" + config_dir = ctx.obj["config_dir"] + tunnel = get_tunnel(config_dir, name) + status = get_status(config_dir, name) + print_tunnel_detail(tunnel, status) + + +@cli.command(name="config") +@click.argument("name") +@click.argument("key") +@click.argument("value") +@click.pass_context +def config_cmd(ctx, name, key, value): + """Update a single tunnel configuration field.""" + config_dir = ctx.obj["config_dir"] + if key == "name": + raise click.ClickException("Cannot rename a tunnel via config. Remove and re-add.") + if key not in _FIELD_TYPES: + raise click.ClickException(f"Unknown field '{key}'") + try: + typed_value = _FIELD_TYPES[key](value) + except ValueError: + raise click.ClickException( + f"Invalid value for '{key}': expected {_FIELD_TYPES[key].__name__}" + ) + tunnels = load_tunnels(config_dir) + found = False + for t in tunnels: + if t.name == name: + setattr(t, key, typed_value) + found = True + break + if not found: + raise click.ClickException(f"No tunnel named '{name}'") + save_tunnels(config_dir, tunnels) + click.echo(f"Updated {name}.{key} = {typed_value}") diff --git a/tests/unit/test_cli_config.py b/tests/unit/test_cli_config.py new file mode 100644 index 0000000..3316164 --- /dev/null +++ b/tests/unit/test_cli_config.py @@ -0,0 +1,128 @@ +import pytest +from click.testing import CliRunner +from autossh_mgr.cli import cli +from autossh_mgr.config import load_tunnels, save_tunnels, TunnelConfig, ensure_dirs + + +@pytest.fixture(autouse=True) +def set_config_dir(config_dir, monkeypatch): + monkeypatch.setenv("AUTOSSH_MGR_CONFIG_DIR", str(config_dir)) + + +@pytest.fixture +def runner(): + return CliRunner() + + +def test_init_creates_structure(runner, tmp_path, monkeypatch): + new_dir = tmp_path / "new-config" + monkeypatch.setenv("AUTOSSH_MGR_CONFIG_DIR", str(new_dir)) + result = runner.invoke(cli, ["init"]) + assert result.exit_code == 0 + assert (new_dir / "tunnels.yaml").exists() + assert (new_dir / "pids").is_dir() + assert (new_dir / "logs").is_dir() + + +def test_init_idempotent(runner): + result = runner.invoke(cli, ["init"]) + assert result.exit_code == 0 + result = runner.invoke(cli, ["init"]) + assert result.exit_code == 0 + + +def test_add_non_interactive(runner, config_dir): + result = runner.invoke(cli, [ + "add", "web-service", + "--host", "relay.example.com", + "--user", "deploy", + "--local-port", "8080", + "--remote-port", "18080", + ]) + assert result.exit_code == 0, result.output + tunnels = load_tunnels(config_dir) + assert len(tunnels) == 1 + assert tunnels[0].name == "web-service" + assert tunnels[0].host == "relay.example.com" + + +def test_add_duplicate_name_fails(runner, config_dir, sample_tunnel): + save_tunnels(config_dir, [sample_tunnel]) + result = runner.invoke(cli, [ + "add", "web-service", + "--host", "other.com", "--user", "u", + "--local-port", "9090", "--remote-port", "19090", + ]) + assert result.exit_code != 0 + assert "already exists" in result.output + + +def test_remove_non_running(runner, config_dir, sample_tunnel): + save_tunnels(config_dir, [sample_tunnel]) + result = runner.invoke(cli, ["remove", "web-service"], input="y\n") + assert result.exit_code == 0 + assert load_tunnels(config_dir) == [] + + +def test_remove_unknown_name(runner): + result = runner.invoke(cli, ["remove", "missing"], input="y\n") + assert result.exit_code != 0 + assert "No tunnel named" in result.output + + +def test_remove_aborted(runner, config_dir, sample_tunnel): + save_tunnels(config_dir, [sample_tunnel]) + result = runner.invoke(cli, ["remove", "web-service"], input="n\n") + assert result.exit_code == 0 + assert len(load_tunnels(config_dir)) == 1 + + +def test_list_empty(runner): + result = runner.invoke(cli, ["list"]) + assert result.exit_code == 0 + assert "No tunnels" in result.output + + +def test_list_shows_tunnels(runner, config_dir, sample_tunnel): + save_tunnels(config_dir, [sample_tunnel]) + result = runner.invoke(cli, ["list"]) + assert result.exit_code == 0 + assert "web-service" in result.output + assert "relay.example.com" in result.output + + +def test_show_tunnel(runner, config_dir, sample_tunnel): + save_tunnels(config_dir, [sample_tunnel]) + result = runner.invoke(cli, ["show", "web-service"]) + assert result.exit_code == 0 + assert "relay.example.com" in result.output + assert "8080" in result.output + assert "18080" in result.output + + +def test_config_update_field(runner, config_dir, sample_tunnel): + save_tunnels(config_dir, [sample_tunnel]) + result = runner.invoke(cli, ["config", "web-service", "remote_port", "19000"]) + assert result.exit_code == 0 + t = load_tunnels(config_dir)[0] + assert t.remote_port == 19000 + + +def test_config_unknown_key(runner, config_dir, sample_tunnel): + save_tunnels(config_dir, [sample_tunnel]) + result = runner.invoke(cli, ["config", "web-service", "bad_key", "val"]) + assert result.exit_code != 0 + assert "Unknown field" in result.output + + +def test_config_rename_disallowed(runner, config_dir, sample_tunnel): + save_tunnels(config_dir, [sample_tunnel]) + result = runner.invoke(cli, ["config", "web-service", "name", "new-name"]) + assert result.exit_code != 0 + assert "Cannot rename" in result.output + + +def test_config_invalid_type(runner, config_dir, sample_tunnel): + save_tunnels(config_dir, [sample_tunnel]) + result = runner.invoke(cli, ["config", "web-service", "port", "notanumber"]) + assert result.exit_code != 0