102 lines
2.8 KiB
Python
102 lines
2.8 KiB
Python
|
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
|