120 lines
2.7 KiB
Python
120 lines
2.7 KiB
Python
from __future__ import annotations
|
|
|
|
from sshctl.config.resolver import ResolvedHost
|
|
from sshctl.operations.checker import HostCheck, check_host
|
|
|
|
|
|
def test_host_check_dataclass() -> None:
|
|
h = HostCheck(
|
|
alias="test",
|
|
hostname="10.0.0.1",
|
|
user="admin",
|
|
port=22,
|
|
proxy_jump="",
|
|
)
|
|
assert h.alias == "test"
|
|
assert h.hostname == "10.0.0.1"
|
|
assert h.user == "admin"
|
|
assert h.port == 22
|
|
|
|
|
|
def test_host_check_passed_both_ok() -> None:
|
|
h = HostCheck(
|
|
alias="test",
|
|
hostname="10.0.0.1",
|
|
user="admin",
|
|
port=22,
|
|
proxy_jump="",
|
|
)
|
|
assert not h.passed
|
|
h.dns_ok = True
|
|
assert not h.passed
|
|
h.tcp_ok = True
|
|
assert h.passed
|
|
|
|
|
|
def test_host_check_summary_all_ok() -> None:
|
|
h = HostCheck(
|
|
alias="test",
|
|
hostname="10.0.0.1",
|
|
user="admin",
|
|
port=22,
|
|
proxy_jump="",
|
|
)
|
|
h.dns_ok = True
|
|
h.tcp_ok = True
|
|
h.ssh_ok = True
|
|
assert h.summary == "DNS/TCP/SSH"
|
|
|
|
|
|
def test_host_check_summary_dns_fail() -> None:
|
|
h = HostCheck(
|
|
alias="test",
|
|
hostname="10.0.0.1",
|
|
user="admin",
|
|
port=22,
|
|
proxy_jump="",
|
|
)
|
|
h.dns_error = "timeout"
|
|
assert h.summary == "DNS!/TCP!/---"
|
|
|
|
|
|
def test_host_check_summary_ssh_untested() -> None:
|
|
h = HostCheck(
|
|
alias="test",
|
|
hostname="10.0.0.1",
|
|
user="admin",
|
|
port=22,
|
|
proxy_jump="",
|
|
)
|
|
h.dns_ok = True
|
|
h.tcp_ok = True
|
|
assert h.summary == "DNS/TCP/---"
|
|
|
|
|
|
def test_host_check_summary_ssh_error() -> None:
|
|
h = HostCheck(
|
|
alias="test",
|
|
hostname="10.0.0.1",
|
|
user="admin",
|
|
port=22,
|
|
proxy_jump="",
|
|
)
|
|
h.dns_ok = True
|
|
h.tcp_ok = True
|
|
h.ssh_error = "refused"
|
|
assert h.summary == "DNS/TCP/SSH!"
|
|
|
|
|
|
async def test_check_host_dns_failure() -> None:
|
|
host = ResolvedHost(
|
|
alias="nonexistent",
|
|
hostname="192.0.2.999",
|
|
user="root",
|
|
port=22,
|
|
proxy_jump="",
|
|
identity_file="",
|
|
options={},
|
|
)
|
|
result = await check_host(host, timeout=1)
|
|
assert isinstance(result.rtt_ms, float)
|
|
assert isinstance(result.dns_ok, bool)
|
|
assert isinstance(result.tcp_ok, bool)
|
|
|
|
|
|
async def test_check_host_constructs_correctly() -> None:
|
|
host = ResolvedHost(
|
|
alias="test-box",
|
|
hostname="192.0.2.1",
|
|
user="deploy",
|
|
port=2222,
|
|
proxy_jump="gw",
|
|
identity_file="",
|
|
options={},
|
|
)
|
|
result = await check_host(host, timeout=2)
|
|
assert result.alias == "test-box"
|
|
assert result.hostname == "192.0.2.1"
|
|
assert result.port == 2222
|
|
assert result.rtt_ms >= 0
|