feat: add CLI config management commands (init, add, remove, list, show, config)
Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
161
src/autossh_mgr/cli.py
Normal file
161
src/autossh_mgr/cli.py
Normal file
@@ -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}")
|
||||
Reference in New Issue
Block a user