sshctl/tests/test_jump.py

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"}