diff --git a/src/autossh_mgr/config.py b/src/autossh_mgr/config.py new file mode 100644 index 0000000..a9ace4c --- /dev/null +++ b/src/autossh_mgr/config.py @@ -0,0 +1,71 @@ +import os +from dataclasses import dataclass, asdict, field +from pathlib import Path +from typing import Optional + +import click +import yaml + + +def get_config_dir() -> Path: + env = os.environ.get("AUTOSSH_MGR_CONFIG_DIR") + if env: + return Path(env) + return Path.home() / ".config" / "autossh-mgr" + + +@dataclass +class TunnelConfig: + name: str + host: str + user: str + local_port: int + remote_port: int + port: int = 22 + identity_file: str = "~/.ssh/id_ed25519" + local_host: str = "127.0.0.1" + remote_host: str = "0.0.0.0" + ssh_options: str = "" + + +_DEFAULTS = TunnelConfig( + name="", host="", user="", local_port=0, remote_port=0 +) + + +def ensure_dirs(config_dir: Path) -> None: + config_dir.mkdir(parents=True, exist_ok=True) + (config_dir / "pids").mkdir(exist_ok=True) + (config_dir / "logs").mkdir(exist_ok=True) + tunnels_file = config_dir / "tunnels.yaml" + if not tunnels_file.exists(): + tunnels_file.write_text("tunnels: []\n") + + +def load_tunnels(config_dir: Path) -> list[TunnelConfig]: + tunnels_file = config_dir / "tunnels.yaml" + if not tunnels_file.exists(): + return [] + data = yaml.safe_load(tunnels_file.read_text()) or {} + return [TunnelConfig(**entry) for entry in data.get("tunnels", [])] + + +def save_tunnels(config_dir: Path, tunnels: list[TunnelConfig]) -> None: + rows = [] + for t in tunnels: + d = asdict(t) + # Omit optional fields that are at their default values + for key in ("port", "identity_file", "local_host", "remote_host", "ssh_options"): + if d[key] == getattr(_DEFAULTS, key): + del d[key] + rows.append(d) + tmp = config_dir / "tunnels.yaml.tmp" + tmp.write_text(yaml.dump({"tunnels": rows}, default_flow_style=False)) + tmp.replace(config_dir / "tunnels.yaml") + + +def get_tunnel(config_dir: Path, name: str) -> TunnelConfig: + for t in load_tunnels(config_dir): + if t.name == name: + return t + raise click.ClickException(f"No tunnel named '{name}'") diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..7bfe76a --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,19 @@ +import pytest +from autossh_mgr.config import TunnelConfig, ensure_dirs + + +@pytest.fixture +def config_dir(tmp_path): + ensure_dirs(tmp_path) + return tmp_path + + +@pytest.fixture +def sample_tunnel(): + return TunnelConfig( + name="web-service", + host="relay.example.com", + user="deploy", + local_port=8080, + remote_port=18080, + ) diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py new file mode 100644 index 0000000..80d9596 --- /dev/null +++ b/tests/unit/test_config.py @@ -0,0 +1,61 @@ +import pytest +import click +from autossh_mgr.config import ( + TunnelConfig, ensure_dirs, load_tunnels, save_tunnels, get_tunnel +) + + +def test_ensure_dirs_creates_structure(tmp_path): + ensure_dirs(tmp_path) + assert (tmp_path / "tunnels.yaml").exists() + assert (tmp_path / "pids").is_dir() + assert (tmp_path / "logs").is_dir() + + +def test_ensure_dirs_is_idempotent(tmp_path): + ensure_dirs(tmp_path) + ensure_dirs(tmp_path) # must not raise + + +def test_load_tunnels_empty(config_dir): + assert load_tunnels(config_dir) == [] + + +def test_save_and_load_roundtrip(config_dir, sample_tunnel): + save_tunnels(config_dir, [sample_tunnel]) + loaded = load_tunnels(config_dir) + assert len(loaded) == 1 + t = loaded[0] + assert t.name == "web-service" + assert t.host == "relay.example.com" + assert t.user == "deploy" + assert t.local_port == 8080 + assert t.remote_port == 18080 + assert t.port == 22 + assert t.identity_file == "~/.ssh/id_ed25519" + assert t.local_host == "127.0.0.1" + assert t.remote_host == "0.0.0.0" + assert t.ssh_options == "" + + +def test_save_atomic_write(config_dir, sample_tunnel): + save_tunnels(config_dir, [sample_tunnel]) + # No .tmp file left behind + assert not (config_dir / "tunnels.yaml.tmp").exists() + + +def test_get_tunnel_found(config_dir, sample_tunnel): + save_tunnels(config_dir, [sample_tunnel]) + t = get_tunnel(config_dir, "web-service") + assert t.name == "web-service" + + +def test_get_tunnel_not_found(config_dir): + with pytest.raises(click.ClickException, match="No tunnel named 'missing'"): + get_tunnel(config_dir, "missing") + + +def test_get_config_dir_env_override(monkeypatch, tmp_path): + monkeypatch.setenv("AUTOSSH_MGR_CONFIG_DIR", str(tmp_path)) + from autossh_mgr.config import get_config_dir + assert get_config_dir() == tmp_path