From 05d9c2201956c131d388b5d459595bdbb1dab186 Mon Sep 17 00:00:00 2001 From: Bleblas2 Date: Wed, 12 Aug 2026 23:45:30 +0200 Subject: [PATCH] Add SSH jump host functionality and related configuration options - Introduced `ignored_device_types` and `ignored_name_patterns` to filter devices based on type and name patterns. - Added support for configuring a jump host in `config.toml` and `README.md`. - Implemented logic to persistently enable/disable jump host for devices. - Updated CLI and TUI to handle jump host settings and display status. - Added tests for new jump host features and filtering functionality. --- README.md | 23 +++++++++- config.toml | 8 ++++ src/netbox_ssh/cli.py | 11 ++++- src/netbox_ssh/config.py | 25 +++++++++++ src/netbox_ssh/editor.py | 10 +++++ src/netbox_ssh/jump_state.py | 40 +++++++++++++++++ src/netbox_ssh/manual.py | 11 ++++- src/netbox_ssh/model.py | 10 ++++- src/netbox_ssh/service.py | 41 ++++++++++++++++- src/netbox_ssh/terminal.py | 20 +++++++-- src/netbox_ssh/tui.py | 87 +++++++++++++++++++++++++++++++++--- tests/test_cli.py | 27 ++++++++++- tests/test_config.py | 17 +++++++ tests/test_jump_state.py | 22 +++++++++ tests/test_terminal.py | 14 ++++++ tests/test_tui.py | 37 ++++++++++++++- 16 files changed, 385 insertions(+), 18 deletions(-) create mode 100644 src/netbox_ssh/jump_state.py create mode 100644 tests/test_jump_state.py diff --git a/README.md b/README.md index 7eff199..5d6d363 100644 --- a/README.md +++ b/README.md @@ -247,6 +247,8 @@ CA. [sync] device_statuses = ["active"] ignored_manufacturers = ["Example Manufacturer"] +ignored_device_types = ["MX*"] +ignored_name_patterns = ["*CORE", "TEST-*"] device_roles = [ "Access Switch", "Core Router", @@ -260,9 +262,26 @@ device_roles = [ case-insensitively. An empty list includes every role. - `ignored_manufacturers` accepts manufacturer names, slugs, or display values, compared case-insensitively. An empty list excludes nothing. -- Status filters are sent to the NetBox API. Role and manufacturer filters are +- `ignored_device_types` contains case-insensitive glob patterns matched against + a device type's model, slug, and display value. +- `ignored_name_patterns` contains case-insensitive glob patterns matched against + device names. `*` matches any text and `?` matches one character. +- Status filters are sent to the NetBox API. Other inventory filters are applied before the cache is written. +### SSH jump host + +```toml +[ssh] +jump_host = "jump-host" +``` + +The value may be a hostname, IP address, `user@host`, or an alias defined in +`~/.ssh/config`. Highlight a device and press `J` to persistently enable or +disable the jump host for it. Marked devices display `J` and are opened with +OpenSSH ProxyJump (`ssh -J jump-host target`). SSH keys remain managed by the +local OpenSSH client. + ## First Run 1. Set `url` and `api_token` in the private user `config.toml`. @@ -289,6 +308,7 @@ required API requests and filters complete successfully. | `Enter` | Open a location or start SSH for a device | | `Ctrl+T` / `Space` | Select or unselect a device for a multi-session launch | | `Ctrl+U` | Clear all selected devices | +| `J` | Enable or disable the configured jump host for a device | | `Esc` | Close search or return to the previous level | | `/` | Search all cached devices by name or primary IP | | `S` | Sync from NetBox | @@ -422,6 +442,7 @@ The implementation is split by responsibility under `src/netbox_ssh`: - `model.py` builds and prunes the location tree. - `cache.py` validates and atomically writes cache version 2. - `manual.py` validates, stores, and merges persistent manual devices. +- `jump_state.py` stores persistent per-device jump-host choices. - `tui.py` implements navigation, search, background sync, and SSH handoff. - `terminal.py` contains the optional multi-tab iTerm2 integration. diff --git a/config.toml b/config.toml index b7cc93a..188abce 100644 --- a/config.toml +++ b/config.toml @@ -14,8 +14,16 @@ device_statuses = ["active"] # Manufacturer names or slugs to exclude from synchronization. ignored_manufacturers = ["Example Manufacturer"] +# Case-insensitive glob patterns (* and ? are supported). +ignored_device_types = [] +ignored_name_patterns = [] + device_roles = [ "Access Switch", "Core Router", "Edge Router", ] + +[ssh] +# Hostname, IP, user@host, or an alias from ~/.ssh/config. +jump_host = "" diff --git a/src/netbox_ssh/cli.py b/src/netbox_ssh/cli.py index 8b76eaa..557752d 100644 --- a/src/netbox_ssh/cli.py +++ b/src/netbox_ssh/cli.py @@ -6,6 +6,7 @@ from . import __version__ from .cache import load_cache from .config import Config +from .jump_state import load_jump_devices from .manual import load_manual_devices from .service import filter_device_roles from .tui import NetBoxSSHApp @@ -35,10 +36,16 @@ def main(argv: list[str] | None = None) -> int: return 1 try: manual_devices = load_manual_devices(config.manual_path) + jump_devices = load_jump_devices( + config.jump_state_path + or config.manual_path.with_name("jump-host-devices.json") + ) except ValueError as error: - print(f"Manual inventory error: {error}", file=sys.stderr) + print(f"Local data error: {error}", file=sys.stderr) return 1 - NetBoxSSHApp(config, load_cache(config.cache_path), manual_devices).run() + NetBoxSSHApp( + config, load_cache(config.cache_path), manual_devices, jump_devices + ).run() return 0 diff --git a/src/netbox_ssh/config.py b/src/netbox_ssh/config.py index 4a92ae4..624c079 100644 --- a/src/netbox_ssh/config.py +++ b/src/netbox_ssh/config.py @@ -21,6 +21,10 @@ class Config: device_roles: tuple[str, ...] device_statuses: tuple[str, ...] ignored_manufacturers: tuple[str, ...] + ignored_device_types: tuple[str, ...] = () + ignored_name_patterns: tuple[str, ...] = () + jump_host: str | None = None + jump_state_path: Path | None = None @classmethod def from_env(cls) -> "Config": @@ -47,6 +51,8 @@ def from_env(cls) -> "Config": file_config = _read_config(config_path) netbox = file_config.get("netbox", {}) sync = file_config.get("sync", {}) + ssh = file_config.get("ssh", {}) + jump_host = _clean_ssh_value(ssh.get("jump_host"), "ssh.jump_host") return cls( # Zmienne powłoki celowo nadpisują ustawienia zapisane w TOML. netbox_url=_clean_url(os.environ.get("NETBOX_URL") or netbox.get("url")), @@ -58,6 +64,7 @@ def from_env(cls) -> "Config": ), cache_path=cache_home / "devices.json", manual_path=data_home / "manual.json", + jump_state_path=data_home / "jump-host-devices.json", config_path=config_path, device_roles=tuple(str(role) for role in sync.get("device_roles", [])), device_statuses=tuple( @@ -67,6 +74,13 @@ def from_env(cls) -> "Config": str(manufacturer) for manufacturer in sync.get("ignored_manufacturers", []) ), + ignored_device_types=tuple( + str(value) for value in sync.get("ignored_device_types", []) + ), + ignored_name_patterns=tuple( + str(value) for value in sync.get("ignored_name_patterns", []) + ), + jump_host=jump_host, ) def validate_sync(self) -> None: @@ -94,6 +108,17 @@ def _as_bool(value: str) -> bool: return value.lower() not in {"0", "false", "no", "off"} +def _clean_ssh_value(value: Any, setting: str) -> str | None: + if value is None or not str(value).strip(): + return None + result = str(value).strip() + if result.startswith("-") or any(character.isspace() for character in result): + raise ValueError( + f"{setting} must be a hostname, IP, or SSH alias without whitespace" + ) + return result + + def _read_config(path: Path) -> dict[str, Any]: """Czyta opcjonalny TOML; brak pliku jest prawidłową konfiguracją domyślną.""" try: diff --git a/src/netbox_ssh/editor.py b/src/netbox_ssh/editor.py index 9b52bd5..b51702e 100644 --- a/src/netbox_ssh/editor.py +++ b/src/netbox_ssh/editor.py @@ -21,9 +21,19 @@ # Empty [] does not exclude any manufacturer. # Example: ignored_manufacturers = ["Cisco", "Juniper", "Arista"] ignored_manufacturers = [] +# Glob patterns matched case-insensitively against model, slug, and display. +# Example: ignored_device_types = ["MX*", "ISR4451"] +ignored_device_types = [] +# Glob patterns matched case-insensitively against device names. +# Example: ignored_name_patterns = ["*CORE", "TEST-*"] +ignored_name_patterns = [] # Empty [] imports devices with every role. # Example: device_roles = ["Router", "Core Switch", "Distribution Switch"] device_roles = [] + +[ssh] +# Hostname, IP, user@host, or an alias from ~/.ssh/config. +jump_host = "" """ diff --git a/src/netbox_ssh/jump_state.py b/src/netbox_ssh/jump_state.py new file mode 100644 index 0000000..b9a42a0 --- /dev/null +++ b/src/netbox_ssh/jump_state.py @@ -0,0 +1,40 @@ +from __future__ import annotations + +import json +import os +import tempfile +from pathlib import Path + + +def load_jump_devices(path: Path) -> set[str]: + """Loads stable device identifiers that should use the configured jump host.""" + try: + data = json.loads(path.read_text(encoding="utf-8")) + if data.get("version") != 1 or not isinstance(data.get("devices"), list): + raise ValueError("Unsupported jump-host state format; expected version 1") + if not all(isinstance(value, str) and value for value in data["devices"]): + raise ValueError("Jump-host device identifiers must be non-empty strings") + return set(data["devices"]) + except FileNotFoundError: + return set() + except (OSError, json.JSONDecodeError, AttributeError, TypeError) as error: + raise ValueError(f"Cannot read {path}: {error}") from error + + +def save_jump_devices(path: Path, identifiers: set[str]) -> None: + """Atomically persists jump-host choices as private user data.""" + payload = {"version": 1, "devices": sorted(identifiers)} + path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) + os.chmod(path.parent, 0o700) + fd, temporary_name = tempfile.mkstemp( + prefix="jump-host-", suffix=".json", dir=path.parent + ) + try: + with os.fdopen(fd, "w", encoding="utf-8") as handle: + json.dump(payload, handle, ensure_ascii=False, indent=2) + handle.write("\n") + os.replace(temporary_name, path) + os.chmod(path, 0o600) + finally: + if os.path.exists(temporary_name): + os.unlink(temporary_name) diff --git a/src/netbox_ssh/manual.py b/src/netbox_ssh/manual.py index 114c82e..5bb6e5f 100644 --- a/src/netbox_ssh/manual.py +++ b/src/netbox_ssh/manual.py @@ -89,7 +89,16 @@ def merge_manual_devices(regions: list[Node], devices: list[ManualDevice]) -> li nodes = current.children assert current is not None current.devices.append( - Device(manual.name, manual.role, manual.target, source="manual") + Device( + manual.name, + manual.role, + manual.target, + source="manual", + identifier=( + "manual:" + + "/".join((*manual.location_path, manual.name)).casefold() + ), + ) ) _sort_tree(merged) return merged diff --git a/src/netbox_ssh/model.py b/src/netbox_ssh/model.py index bc57011..2c3f45b 100644 --- a/src/netbox_ssh/model.py +++ b/src/netbox_ssh/model.py @@ -12,6 +12,8 @@ class Device: role: str primary_ip: str | None = None source: str = "netbox" + identifier: str | None = None + use_jump_host: bool = False @property def ssh_target(self) -> str: @@ -24,6 +26,7 @@ def to_dict(self) -> dict[str, Any]: "role": self.role, "primary_ip": self.primary_ip, "source": self.source, + "identifier": self.identifier, } @classmethod @@ -33,6 +36,7 @@ def from_dict(cls, data: dict[str, Any]) -> "Device": data["role"], data.get("primary_ip"), data.get("source", "netbox"), + data.get("identifier"), ) @@ -109,7 +113,11 @@ def build_tree( ip = raw.get("primary_ip4") or raw.get("primary_ip6") if isinstance(ip, dict): ip = ip.get("address") or ip.get("display") - target_node.devices.append(Device(str(device_name), str(role_name), ip)) + device_id = raw.get("id") + identifier = f"netbox:{device_id}" if device_id is not None else None + target_node.devices.append( + Device(str(device_name), str(role_name), ip, identifier=identifier) + ) _prune_and_sort(roots) return roots diff --git a/src/netbox_ssh/service.py b/src/netbox_ssh/service.py index d07b1e2..bb76f0d 100644 --- a/src/netbox_ssh/service.py +++ b/src/netbox_ssh/service.py @@ -1,4 +1,4 @@ -from __future__ import annotations +from fnmatch import fnmatchcase import httpx @@ -17,6 +17,8 @@ def synchronize(config: Config) -> tuple[Cache, int]: regions, sites, devices = client.fetch_inventory(config.device_statuses) devices = filter_device_roles(devices, config.device_roles) devices = filter_ignored_manufacturers(devices, config.ignored_manufacturers) + devices = filter_ignored_device_types(devices, config.ignored_device_types) + devices = filter_ignored_name_patterns(devices, config.ignored_name_patterns) region_tree = build_tree(regions, sites, devices) return save_cache(config.cache_path, region_tree), len(devices) @@ -57,6 +59,43 @@ def filter_ignored_manufacturers( return result +def _matches_any(value: object, patterns: tuple[str, ...]) -> bool: + text = str(value or "").casefold() + return bool(text) and any( + fnmatchcase(text, pattern.casefold()) for pattern in patterns + ) + + +def filter_ignored_device_types( + devices: list[dict], ignored_patterns: tuple[str, ...] +) -> list[dict]: + """Removes devices whose model, slug, or display matches a glob pattern.""" + if not ignored_patterns: + return devices + result = [] + for device in devices: + device_type = device.get("device_type") or {} + values = (device_type.get(field) for field in ("model", "slug", "display")) + if not any(_matches_any(value, ignored_patterns) for value in values): + result.append(device) + return result + + +def filter_ignored_name_patterns( + devices: list[dict], ignored_patterns: tuple[str, ...] +) -> list[dict]: + """Removes devices whose name (or display fallback) matches a glob pattern.""" + if not ignored_patterns: + return devices + return [ + device + for device in devices + if not _matches_any( + device.get("name") or device.get("display"), ignored_patterns + ) + ] + + def describe_sync_error(error: Exception) -> str: """Zamienia techniczne wyjątki HTTP na komunikaty zrozumiałe w TUI.""" if isinstance(error, ValueError): diff --git a/src/netbox_ssh/terminal.py b/src/netbox_ssh/terminal.py index 12a7869..5320cb0 100644 --- a/src/netbox_ssh/terminal.py +++ b/src/netbox_ssh/terminal.py @@ -32,7 +32,19 @@ def is_iterm2() -> bool: return platform.system() == "Darwin" and os.environ.get("TERM_PROGRAM") == "iTerm.app" -def run_system_ssh(devices: Sequence[Device]) -> list[tuple[Device, int]]: +def ssh_arguments(device: Device, jump_host: str | None = None) -> list[str]: + arguments = ["ssh"] + if device.use_jump_host: + if not jump_host: + raise ValueError("No SSH jump host is configured.") + arguments.extend(["-J", jump_host]) + arguments.append(device.ssh_target) + return arguments + + +def run_system_ssh( + devices: Sequence[Device], jump_host: str | None = None +) -> list[tuple[Device, int]]: """Uruchamia systemowy OpenSSH, przenośnie także na Linuxie i WSL.""" environment = os.environ.copy() environment.pop("NETBOX_API_TOKEN", None) @@ -40,13 +52,13 @@ def run_system_ssh(devices: Sequence[Device]) -> list[tuple[Device, int]]: results = [] for device in devices: result = subprocess.run( - ["ssh", device.ssh_target], check=False, env=environment + ssh_arguments(device, jump_host), check=False, env=environment ) results.append((device, result.returncode)) return results -def open_iterm_tabs(devices: Sequence[Device]) -> None: +def open_iterm_tabs(devices: Sequence[Device], jump_host: str | None = None) -> None: """Otwiera osobną kartę iTerm2 dla każdego urządzenia. Polecenie SSH jest cytowane jako pojedynczy argument powłoki, a sam @@ -58,7 +70,7 @@ def open_iterm_tabs(devices: Sequence[Device]) -> None: if not is_iterm2(): raise RuntimeError("Opening multiple sessions requires iTerm2 on macOS.") - commands = [f"ssh {shlex.quote(device.ssh_target)}" for device in devices] + commands = [shlex.join(ssh_arguments(device, jump_host)) for device in devices] environment = os.environ.copy() environment.pop("NETBOX_API_TOKEN", None) environment.pop("NETBOX_URL", None) diff --git a/src/netbox_ssh/tui.py b/src/netbox_ssh/tui.py index 734585e..a8fa24d 100644 --- a/src/netbox_ssh/tui.py +++ b/src/netbox_ssh/tui.py @@ -17,6 +17,7 @@ from .cache import Cache from .config import Config from .editor import editor_command, ensure_config_file, ensure_manual_file +from .jump_state import save_jump_devices from .manual import ( ManualDevice, load_manual_devices, @@ -154,6 +155,7 @@ class NetBoxSSHApp(App[None]): Binding("ctrl+t", "toggle_selection", "Select device"), Binding("space", "toggle_selection", "Select device", show=False), Binding("ctrl+u", "clear_selection", "Clear selection"), + Binding("j", "toggle_jump_host", "Toggle jump host"), ] def __init__( @@ -161,12 +163,15 @@ def __init__( config: Config, cache: Cache | None, manual_devices: list[ManualDevice] | None = None, + jump_devices: set[str] | None = None, ) -> None: super().__init__() self.config = config self.cache = cache self.manual_devices = list(manual_devices or []) + self.jump_devices = set(jump_devices or set()) self.regions = self._merged_regions() + self._apply_jump_state() self.views: list[View] = [View("Countries", path=())] self.visible_entries: list[Entry] = [] self.item_entries: dict[int, Entry] = {} @@ -288,7 +293,12 @@ async def _render_entries(self) -> None: icon = "✓" else: icon = "◇" if entry.kind == "device" and entry.value.source == "manual" else icons.get(entry.kind, " ") - line.append(f"{icon} {entry.label}") + jump_marker = ( + "J" + if entry.kind == "device" and entry.value.use_jump_host + else " " + ) + line.append(f"{icon} {jump_marker} {entry.label}") if entry.detail: line.append(f" {entry.detail}", style="dim") item = ListItem( @@ -380,6 +390,43 @@ async def action_clear_selection(self) -> None: list_view.index = current_index self._set_status("Device selection cleared.") + async def action_toggle_jump_host(self) -> None: + """Trwale przełącza ProxyJump dla wskazanego urządzenia.""" + list_view = self.query_one(ListView) + if list_view.index is None or list_view.index >= len(self.visible_entries): + return + entry = self.visible_entries[list_view.index] + if entry.kind != "device": + self._set_status("Jump host can only be set for a device.", "error") + return + if not self.config.jump_host: + self._set_status("Configure ssh.jump_host before marking devices.", "error") + return + device = entry.value + if not device.identifier: + self._set_status("This device has no stable identifier.", "error") + return + was_enabled = device.identifier in self.jump_devices + if was_enabled: + self.jump_devices.remove(device.identifier) + else: + self.jump_devices.add(device.identifier) + try: + save_jump_devices(self._jump_state_path(), self.jump_devices) + except OSError as error: + if was_enabled: + self.jump_devices.add(device.identifier) + else: + self.jump_devices.discard(device.identifier) + self._set_status(f"Could not save jump-host setting: {error}", "error") + return + self._apply_jump_state() + current_index = list_view.index + await self._render_entries() + list_view.index = current_index + state = "disabled" if was_enabled else "enabled" + self._set_status(f"Jump host {state} for {device.name}.", "success") + def _connect_selected(self) -> None: """Otwiera wiele sesji tylko w kartach iTerm2.""" devices = list(self.selected_devices.values()) @@ -391,8 +438,8 @@ def _connect_selected(self) -> None: ) return try: - open_iterm_tabs(devices) - except (OSError, RuntimeError) as error: + open_iterm_tabs(devices, self.config.jump_host) + except (OSError, RuntimeError, ValueError) as error: self._set_status(f"Could not start SSH: {error}", "error") return self.selected_devices.clear() @@ -494,6 +541,7 @@ def action_edit_manual(self) -> None: return self.manual_devices = manual_devices self.regions = self._merged_regions() + self._apply_jump_state() self.views = [View("Countries", path=())] self.selected_devices.clear() self.run_worker(self._reset_and_render(), exclusive=True) @@ -545,7 +593,16 @@ def _manual_device_added(self, manual: ManualDevice | None) -> None: current_node = self.views[-1].node assert current_node is not None current_node.devices.append( - Device(manual.name, manual.role, manual.target, source="manual") + Device( + manual.name, + manual.role, + manual.target, + source="manual", + identifier=( + "manual:" + + "/".join((*manual.location_path, manual.name)).casefold() + ), + ) ) current_node.devices.sort( key=lambda device: (device.role.casefold(), device.name.casefold()) @@ -571,6 +628,7 @@ def _sync_finished(self, cache: Cache, device_count: int) -> None: self.syncing = False self.cache = cache self.regions = self._merged_regions() + self._apply_jump_state() self.views = [View("Countries", path=())] self.run_worker(self._reset_and_render(), exclusive=True) country_count = sum(len(region.children) for region in cache.regions) @@ -585,6 +643,23 @@ def _merged_regions(self) -> list[Node]: self.cache.regions if self.cache else [], self.manual_devices ) + def _jump_state_path(self): + return self.config.jump_state_path or self.config.manual_path.with_name( + "jump-host-devices.json" + ) + + def _apply_jump_state(self) -> None: + def visit(node: Node) -> None: + for device in node.devices: + device.use_jump_host = bool( + device.identifier and device.identifier in self.jump_devices + ) + for child in node.children: + visit(child) + + for region in self.regions: + visit(region) + def _connect_ssh(self, device: Device) -> None: environment = os.environ.copy() # Proces SSH nie potrzebuje sekretów NetBoxa i nie powinien ich dziedziczyć. @@ -595,8 +670,8 @@ def _connect_ssh(self, device: Device) -> None: # Na czas SSH oddajemy terminal klientowi systemowemu, a po jego # zakończeniu Textual odtwarza poprzedni ekran. with self.suspend(): - results = run_system_ssh([device]) - except OSError as error: + results = run_system_ssh([device], self.config.jump_host) + except (OSError, ValueError) as error: self._set_status(f"Could not start ssh: {error}", "error") else: return_code = results[0][1] diff --git a/tests/test_cli.py b/tests/test_cli.py index d725d47..5df0e86 100644 --- a/tests/test_cli.py +++ b/tests/test_cli.py @@ -1,7 +1,11 @@ import unittest from netbox_ssh.cli import filter_device_roles -from netbox_ssh.service import filter_ignored_manufacturers +from netbox_ssh.service import ( + filter_ignored_device_types, + filter_ignored_manufacturers, + filter_ignored_name_patterns, +) class RoleFilterTests(unittest.TestCase): @@ -60,5 +64,26 @@ def test_missing_device_type_is_kept_when_filtering_manufacturers(self) -> None: ) +class GlobFilterTests(unittest.TestCase): + def test_ignores_device_type_fields_case_insensitively(self) -> None: + devices = [ + {"name": "one", "device_type": {"model": "MX67", "slug": "mx67"}}, + {"name": "two", "device_type": {"display": "ISR4451-X"}}, + {"name": "three", "device_type": {"model": "C9300"}}, + {"name": "unknown", "device_type": None}, + ] + result = filter_ignored_device_types(devices, ("mx*", "ISR????-X")) + self.assertEqual([item["name"] for item in result], ["three", "unknown"]) + + def test_ignores_device_names_by_glob(self) -> None: + devices = [ + {"name": "WAW-CORE"}, + {"name": "test-router"}, + {"name": "access-01"}, + ] + result = filter_ignored_name_patterns(devices, ("*core", "TEST-*")) + self.assertEqual([item["name"] for item in result], ["access-01"]) + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_config.py b/tests/test_config.py index 794f8b9..7627409 100644 --- a/tests/test_config.py +++ b/tests/test_config.py @@ -28,6 +28,7 @@ def test_loads_roles_from_user_config(self) -> None: self.assertEqual(config.device_statuses, ()) self.assertEqual(config.ignored_manufacturers, ()) self.assertEqual(config.manual_path.name, "manual.json") + self.assertEqual(config.jump_state_path.name, "jump-host-devices.json") def test_default_reads_all_device_roles(self) -> None: with tempfile.TemporaryDirectory() as directory: @@ -43,6 +44,22 @@ def test_default_reads_all_device_roles(self) -> None: self.assertEqual(config.device_statuses, ()) self.assertEqual(config.ignored_manufacturers, ()) + def test_loads_jump_host_and_glob_filters(self) -> None: + with tempfile.TemporaryDirectory() as directory: + root = Path(directory) + (root / "config.toml").write_text( + '[sync]\nignored_device_types = ["MX*"]\n' + 'ignored_name_patterns = ["*CORE"]\n[ssh]\njump_host = "jump-alias"\n', + encoding="utf-8", + ) + with patch.dict(os.environ, {"XDG_CONFIG_HOME": str(root / "missing")}, clear=True), patch( + "pathlib.Path.cwd", return_value=root + ): + config = Config.from_env() + self.assertEqual(config.ignored_device_types, ("MX*",)) + self.assertEqual(config.ignored_name_patterns, ("*CORE",)) + self.assertEqual(config.jump_host, "jump-alias") + def test_uses_local_config_when_user_config_is_missing(self) -> None: with tempfile.TemporaryDirectory() as directory: root = Path(directory) diff --git a/tests/test_jump_state.py b/tests/test_jump_state.py new file mode 100644 index 0000000..d11c762 --- /dev/null +++ b/tests/test_jump_state.py @@ -0,0 +1,22 @@ +import stat +import tempfile +import unittest +from pathlib import Path + +from netbox_ssh.jump_state import load_jump_devices, save_jump_devices + + +class JumpStateTests(unittest.TestCase): + def test_round_trip_and_permissions(self) -> None: + with tempfile.TemporaryDirectory() as directory: + path = Path(directory) / "data" / "jump-host-devices.json" + save_jump_devices(path, {"netbox:2", "netbox:1"}) + self.assertEqual(load_jump_devices(path), {"netbox:1", "netbox:2"}) + self.assertEqual(stat.S_IMODE(path.stat().st_mode), 0o600) + + def test_missing_file_is_empty(self) -> None: + self.assertEqual(load_jump_devices(Path("/definitely/missing/state.json")), set()) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_terminal.py b/tests/test_terminal.py index c8dc4eb..63c53cf 100644 --- a/tests/test_terminal.py +++ b/tests/test_terminal.py @@ -54,6 +54,20 @@ def test_runs_single_device_without_secrets(self, run) -> None: self.assertNotIn("NETBOX_API_TOKEN", run.call_args.kwargs["env"]) self.assertNotIn("NETBOX_URL", run.call_args.kwargs["env"]) + @patch("netbox_ssh.terminal.subprocess.run") + def test_runs_marked_device_through_jump_host(self, run) -> None: + run.return_value.returncode = 0 + device = Device("switch-one", "Core", "192.0.2.1", use_jump_host=True) + run_system_ssh([device], "jump-alias") + self.assertEqual( + run.call_args.args[0], ["ssh", "-J", "jump-alias", "192.0.2.1"] + ) + + def test_rejects_marked_device_without_configured_jump_host(self) -> None: + device = Device("switch-one", "Core", use_jump_host=True) + with self.assertRaisesRegex(ValueError, "No SSH jump host"): + run_system_ssh([device]) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_tui.py b/tests/test_tui.py index 2799647..526d8de 100644 --- a/tests/test_tui.py +++ b/tests/test_tui.py @@ -135,9 +135,44 @@ async def test_selects_devices_and_opens_them_as_batch(self) -> None: self.assertEqual(list(app.selected_devices.values()), [first, second]) await pilot.press("enter") await pilot.pause() - open_tabs.assert_called_once_with([first, second]) + open_tabs.assert_called_once_with([first, second], None) self.assertEqual(app.selected_devices, {}) + async def test_toggles_and_persists_jump_host_for_device(self) -> None: + device = Device( + "switch-one", "Access Switch", "192.0.2.1/24", identifier="netbox:42" + ) + branch = Node("branch-a-01", devices=[device]) + country = Node("Country A", children=[Node("City A", children=[branch])]) + with tempfile.TemporaryDirectory() as directory: + app = self.make_app( + Cache( + "2026-08-02T00:00:00+02:00", + [Node("Region Group A", children=[country])], + ), + Path(directory), + ) + app.config = Config( + **{ + **app.config.__dict__, + "jump_host": "jump-alias", + "jump_state_path": Path(directory) / "jump.json", + } + ) + async with app.run_test() as pilot: + await pilot.press("enter", "enter", "j") + await pilot.pause() + visible_device = next( + entry.value for entry in app.visible_entries if entry.kind == "device" + ) + self.assertTrue(visible_device.use_jump_host) + self.assertEqual(app.jump_devices, {"netbox:42"}) + self.assertTrue(app.config.jump_state_path.is_file()) + await pilot.press("j") + await pilot.pause() + self.assertFalse(visible_device.use_jump_host) + self.assertEqual(app.jump_devices, set()) + async def test_clears_selected_devices(self) -> None: device = Device("switch-one", "Access Switch", "192.0.2.1/24") branch = Node("branch-a-01", devices=[device])