272 lines
8.6 KiB
Python
272 lines
8.6 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
from sshctl.operations.jump import (
|
|
add_network,
|
|
current_ssid,
|
|
is_internal_by_ssid,
|
|
is_internal_by_subnet,
|
|
known_networks,
|
|
list_hosts,
|
|
load_state,
|
|
off,
|
|
on,
|
|
save_state,
|
|
status,
|
|
)
|
|
|
|
|
|
def _mk_conf(ssh_dir: Path, name: str, text: str) -> Path:
|
|
conf_d = ssh_dir / "conf.d"
|
|
conf_d.mkdir(parents=True, exist_ok=True)
|
|
p = conf_d / name
|
|
p.write_text(text)
|
|
return p
|
|
|
|
|
|
def test_known_networks_defaults(ssh_dir: Path) -> None:
|
|
assert known_networks(ssh_dir) == ["MobinNet*", "Mobin*"]
|
|
|
|
|
|
def test_known_networks_custom(ssh_dir: Path) -> None:
|
|
(ssh_dir / ".internal_ssids").write_text("Home*\n # comment\nOffice\n")
|
|
assert known_networks(ssh_dir) == ["Home*", "Office"]
|
|
|
|
|
|
def test_add_network_creates_file(ssh_dir: Path) -> None:
|
|
add_network(ssh_dir, "MyWiFi")
|
|
lines = (ssh_dir / ".internal_ssids").read_text().splitlines()
|
|
assert lines[-1] == "MyWiFi"
|
|
|
|
|
|
def test_add_network_does_not_duplicate(ssh_dir: Path) -> None:
|
|
add_network(ssh_dir, "TestNet")
|
|
add_network(ssh_dir, "TestNet")
|
|
assert (ssh_dir / ".internal_ssids").read_text().strip().count("TestNet") == 1
|
|
|
|
|
|
def test_current_ssid_detected() -> None:
|
|
with patch("sshctl.operations.jump.subprocess.run") as m:
|
|
m.return_value.stdout = "yes:MyWiFi\nno:OldNet\n"
|
|
assert current_ssid() == "MyWiFi"
|
|
|
|
|
|
def test_current_ssid_none() -> None:
|
|
with patch("sshctl.operations.jump.subprocess.run") as m:
|
|
m.return_value.stdout = "no:OldNet\n"
|
|
assert current_ssid() is None
|
|
|
|
|
|
def test_current_ssid_subprocess_error() -> None:
|
|
with patch("sshctl.operations.jump.subprocess.run", side_effect=FileNotFoundError):
|
|
assert current_ssid() is None
|
|
|
|
|
|
def test_is_internal_by_ssid_no_wifi(tmp_path: Path) -> None:
|
|
with patch("sshctl.operations.jump.current_ssid", return_value=None):
|
|
assert is_internal_by_ssid(tmp_path) is None
|
|
|
|
|
|
def test_is_internal_by_ssid_match(tmp_path: Path) -> None:
|
|
(tmp_path / ".internal_ssids").write_text("MobinNet*\n")
|
|
with patch("sshctl.operations.jump.current_ssid", return_value="MobinNet-5G"):
|
|
assert is_internal_by_ssid(tmp_path) is True
|
|
|
|
|
|
def test_is_internal_by_ssid_no_match(tmp_path: Path) -> None:
|
|
(tmp_path / ".internal_ssids").write_text("MobinNet*\n")
|
|
with patch("sshctl.operations.jump.current_ssid", return_value="CoffeeShop"):
|
|
assert is_internal_by_ssid(tmp_path) is False
|
|
|
|
|
|
def test_is_internal_by_ssid_default_patterns(tmp_path: Path) -> None:
|
|
with patch("sshctl.operations.jump.current_ssid", return_value="Mobin-Office"):
|
|
assert is_internal_by_ssid(tmp_path) is True
|
|
|
|
|
|
def test_list_hosts(ssh_dir: Path) -> None:
|
|
_mk_conf(
|
|
ssh_dir,
|
|
"10-servers.conf",
|
|
"Host gw\n HostName 10.0.0.254\n\nHost web01\n HostName 10.0.0.1\n ProxyJump gw\n",
|
|
)
|
|
hosts = list_hosts(ssh_dir)
|
|
assert len(hosts) == 1
|
|
assert hosts[0]["alias"] == "web01"
|
|
assert hosts[0]["jump"] == "gw"
|
|
assert hosts[0]["file"] == "conf.d/10-servers.conf"
|
|
|
|
|
|
def test_list_hosts_empty(ssh_dir: Path) -> None:
|
|
assert list_hosts(ssh_dir) == []
|
|
|
|
|
|
def test_list_hosts_trailing_host_with_proxy(ssh_dir: Path) -> None:
|
|
_mk_conf(
|
|
ssh_dir,
|
|
"10-servers.conf",
|
|
"Host web01\n HostName 10.0.0.1\n ProxyJump gw\n",
|
|
)
|
|
hosts = list_hosts(ssh_dir)
|
|
assert len(hosts) == 1
|
|
assert hosts[0]["alias"] == "web01"
|
|
|
|
|
|
def test_save_state(ssh_dir: Path) -> None:
|
|
_mk_conf(
|
|
ssh_dir,
|
|
"10-servers.conf",
|
|
"Host web01\n HostName 10.0.0.1\n ProxyJump gw\n",
|
|
)
|
|
records = save_state(ssh_dir)
|
|
assert len(records) == 1
|
|
assert records[0]["file"] == "conf.d/10-servers.conf"
|
|
assert records[0]["text"] == " ProxyJump gw"
|
|
assert (ssh_dir / ".proxyjump_save").exists()
|
|
|
|
|
|
def test_save_state_no_proxy(ssh_dir: Path) -> None:
|
|
_mk_conf(ssh_dir, "10-servers.conf", "Host web01\n HostName 10.0.0.1\n")
|
|
assert save_state(ssh_dir) == []
|
|
|
|
|
|
def test_load_state_missing(ssh_dir: Path) -> None:
|
|
assert load_state(ssh_dir) is None
|
|
|
|
|
|
def test_load_state_invalid_json(ssh_dir: Path) -> None:
|
|
(ssh_dir / ".proxyjump_save").write_text("{not json")
|
|
assert load_state(ssh_dir) is None
|
|
|
|
|
|
def test_load_state_valid(ssh_dir: Path) -> None:
|
|
(ssh_dir / ".proxyjump_save").write_text(
|
|
json.dumps([{"file": "conf.d/x.conf", "line": 3, "text": " ProxyJump gw"}])
|
|
)
|
|
records = load_state(ssh_dir)
|
|
assert records is not None
|
|
assert records[0]["file"] == "conf.d/x.conf"
|
|
|
|
|
|
def test_off_noop(ssh_dir: Path) -> None:
|
|
_mk_conf(ssh_dir, "10-servers.conf", "Host web01\n HostName 10.0.0.1\n")
|
|
result = off(ssh_dir)
|
|
assert result["status"] == "noop"
|
|
|
|
|
|
def test_off_removes_proxy_jump(ssh_dir: Path) -> None:
|
|
f = _mk_conf(
|
|
ssh_dir,
|
|
"10-servers.conf",
|
|
"Host web01\n HostName 10.0.0.1\n ProxyJump gw\n",
|
|
)
|
|
result = off(ssh_dir)
|
|
assert result["status"] == "ok"
|
|
assert result["count"] == 1
|
|
assert "ProxyJump" not in f.read_text()
|
|
assert (ssh_dir / ".proxyjump_save").exists()
|
|
|
|
|
|
def test_off_multiple_files(ssh_dir: Path) -> None:
|
|
_mk_conf(ssh_dir, "10-servers.conf", "Host web01\n ProxyJump gw\n")
|
|
_mk_conf(ssh_dir, "20-servers.conf", "Host web02\n ProxyJump gw\n")
|
|
result = off(ssh_dir)
|
|
assert result["status"] == "ok"
|
|
assert result["count"] == 2
|
|
assert len(result["files"]) == 2
|
|
|
|
|
|
def test_on_noop(ssh_dir: Path) -> None:
|
|
result = on(ssh_dir)
|
|
assert result["status"] == "noop"
|
|
|
|
|
|
def test_on_restores(ssh_dir: Path) -> None:
|
|
_mk_conf(
|
|
ssh_dir,
|
|
"10-servers.conf",
|
|
"Host web01\n HostName 10.0.0.1\n ProxyJump gw\n",
|
|
)
|
|
off(ssh_dir)
|
|
f = ssh_dir / "conf.d" / "10-servers.conf"
|
|
|
|
result = on(ssh_dir)
|
|
assert result["status"] == "ok"
|
|
assert "ProxyJump gw" in f.read_text()
|
|
assert not (ssh_dir / ".proxyjump_save").exists()
|
|
|
|
|
|
def test_on_restores_at_original_line(ssh_dir: Path) -> None:
|
|
conf = "Host web01\n HostName 10.0.0.1\n ProxyJump gw\n\nHost web02\n HostName 10.0.0.2\n"
|
|
_mk_conf(ssh_dir, "10-servers.conf", conf)
|
|
off(ssh_dir)
|
|
on(ssh_dir)
|
|
text = (ssh_dir / "conf.d" / "10-servers.conf").read_text()
|
|
assert text.index("ProxyJump") < text.index("web02")
|
|
|
|
|
|
def test_on_skips_file_with_existing_proxy(ssh_dir: Path) -> None:
|
|
_mk_conf(ssh_dir, "10-servers.conf", "Host web01\n ProxyJump gw\n")
|
|
off(ssh_dir)
|
|
(ssh_dir / "conf.d" / "10-servers.conf").write_text("Host web01\n ProxyJump newgw\n")
|
|
result = on(ssh_dir)
|
|
assert result["files"] == []
|
|
assert "newgw" in (ssh_dir / "conf.d" / "10-servers.conf").read_text()
|
|
|
|
|
|
def test_on_missing_file_skipped(ssh_dir: Path) -> None:
|
|
_mk_conf(ssh_dir, "10-servers.conf", "Host web01\n ProxyJump gw\n")
|
|
off(ssh_dir)
|
|
(ssh_dir / "conf.d" / "10-servers.conf").unlink()
|
|
result = on(ssh_dir)
|
|
assert result["status"] == "ok"
|
|
|
|
|
|
def test_on_line_appended_when_out_of_range(ssh_dir: Path) -> None:
|
|
_mk_conf(ssh_dir, "10-servers.conf", "Host web01\n ProxyJump gw\n")
|
|
off(ssh_dir)
|
|
(ssh_dir / "conf.d" / "10-servers.conf").write_text("Host web01\n")
|
|
result = on(ssh_dir)
|
|
assert result["status"] == "ok"
|
|
assert "ProxyJump gw" in (ssh_dir / "conf.d" / "10-servers.conf").read_text()
|
|
|
|
|
|
def test_status(ssh_dir: Path) -> None:
|
|
_mk_conf(ssh_dir, "10-servers.conf", "Host web01\n ProxyJump gw\n")
|
|
info = status(ssh_dir)
|
|
assert info["count"] == 1
|
|
assert info["saved"] is False
|
|
|
|
|
|
def test_status_saved(ssh_dir: Path) -> None:
|
|
_mk_conf(ssh_dir, "10-servers.conf", "Host web01\n ProxyJump gw\n")
|
|
off(ssh_dir)
|
|
info = status(ssh_dir)
|
|
assert info["saved"] is True
|
|
|
|
|
|
def test_is_internal_by_subnet_no_gateways(ssh_dir: Path) -> None:
|
|
reachable, probes = is_internal_by_subnet(ssh_dir)
|
|
assert reachable is False
|
|
assert probes == []
|
|
|
|
|
|
def test_is_internal_by_subnet_reachable(ssh_dir: Path) -> None:
|
|
_mk_conf(ssh_dir, "10-servers.conf", "Host web01\n ProxyJump gw\n")
|
|
probes = [{"gateway": "gw", "ip": "10.0.0.254", "port": 22, "reachable": True, "error": ""}]
|
|
with patch("sshctl.operations.jump._subnet_reachable", return_value=probes):
|
|
reachable, result = is_internal_by_subnet(ssh_dir)
|
|
assert reachable is True
|
|
assert result == probes
|
|
|
|
|
|
def test_jump_gateways_collects_unique(ssh_dir: Path) -> None:
|
|
from sshctl.operations.jump import _jump_gateways
|
|
|
|
_mk_conf(ssh_dir, "10-servers.conf", "Host a\n ProxyJump gw\n")
|
|
_mk_conf(ssh_dir, "20-servers.conf", "Host b\n ProxyJump gw:2222\n ProxyJump other\n")
|
|
assert _jump_gateways(ssh_dir) == {"gw", "other"}
|