feat: add TunnelConfig dataclass and YAML I/O

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-05-21 11:31:19 +08:00
parent 113cc994b1
commit 53133ccb67
3 changed files with 151 additions and 0 deletions

71
src/autossh_mgr/config.py Normal file
View File

@@ -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}'")

19
tests/conftest.py Normal file
View File

@@ -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,
)

61
tests/unit/test_config.py Normal file
View File

@@ -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