from __future__ import annotations from pathlib import Path import pytest from sshctl.config.resolver import ( ResolvedHost, _parse_ssh_g_output, resolve_all_hosts, resolve_host, ) from sshctl.core.exceptions import ResolverError def test_resolved_host_namedtuple() -> None: host = ResolvedHost( alias="server1", hostname="10.0.0.1", user="admin", port=22, proxy_jump="gw", identity_file="~/.ssh/id_rsa", options={"hostname": "10.0.0.1", "user": "admin"}, ) assert host.alias == "server1" assert host.hostname == "10.0.0.1" assert host.port == 22 assert host.proxy_jump == "gw" def test_parse_ssh_g_output_basic() -> None: text = "hostname 10.0.0.1\nuser admin\nport 2222\n" opts = _parse_ssh_g_output(text) assert opts["hostname"] == "10.0.0.1" assert opts["user"] == "admin" assert opts["port"] == "2222" def test_parse_ssh_g_output_empty() -> None: assert _parse_ssh_g_output("") == {} def test_parse_ssh_g_output_blank_lines() -> None: text = "\n\nhostname test\n\n" opts = _parse_ssh_g_output(text) assert opts == {"hostname": "test"} def test_parse_ssh_g_output_first_wins() -> None: text = "hostname first\nhostname second\n" opts = _parse_ssh_g_output(text) assert opts["hostname"] == "first" def test_parse_ssh_g_output_quoted_values() -> None: text = 'identityfile "~/.ssh/key"\n' opts = _parse_ssh_g_output(text) assert opts["identityfile"] == "~/.ssh/key" def test_parse_ssh_g_output_single_word_line() -> None: text = "invalid\nhostname test\n" opts = _parse_ssh_g_output(text) assert opts == {"hostname": "test"} def test_resolve_host_ssh_not_found() -> None: with pytest.raises(ResolverError, match="Failed to resolve"): resolve_host("test", "/nonexistent/config", ssh_bin="ssh-nonexistent") def test_resolve_all_hosts_empty_config(tmp_path: Path) -> None: config_path = tmp_path / "config" config_path.write_text("Host *\n User root\n") results = resolve_all_hosts(config_path, ssh_bin="ssh-nonexistent") assert results == [] def test_resolve_host_uses_correct_cmd() -> None: import subprocess def _fake_run(*_args: object, **_kwargs: object) -> object: class FakeResult: returncode = 0 stdout = "hostname 10.0.0.1\nuser testuser\nport 2222\nproxyjump jumpbox\n" stderr = "" return FakeResult() original_run = subprocess.run subprocess.run = _fake_run # type: ignore[assignment] try: host = resolve_host("myserver", "/tmp/fake_config") assert host.alias == "myserver" assert host.hostname == "10.0.0.1" assert host.user == "testuser" assert host.port == 2222 assert host.proxy_jump == "jumpbox" finally: subprocess.run = original_run