"""The node's SSH session verifies its host key, and reports the one it sees (NetOrk/netork#1167).

Both ways onto the node's shell — the best-effort ``_exec_ssh_command`` and the
provisioning steps' ``_run_node_command`` — share one session, opened on port 22.
"""

from __future__ import annotations

from unittest.mock import MagicMock

import paramiko
import pytest
from napalm_device_types.ssh_host_keys import (
    KNOWN_KEYS_ARG,
    OBSERVER_ARG,
    SshHostKey,
    SshHostKeyMismatch,
)

from napalm_proxmox.driver import ProxmoxDriver

SERVER_KEY = paramiko.ECDSAKey.generate()
OTHER_KEY = paramiko.ECDSAKey.generate()


class _Stream:
    def __init__(self, data: bytes) -> None:
        self._data = data
        self.channel = MagicMock()
        self.channel.recv_exit_status.return_value = 0

    def read(self) -> bytes:
        return self._data


class _Client:
    """paramiko.SSHClient as far as the driver uses it; refuses like paramiko would."""

    def __init__(self) -> None:
        self.lines: list[str] = []
        self.policy = None
        self.connected_with: dict | None = None
        self.commands: list[str] = []

    def set_missing_host_key_policy(self, policy) -> None:
        self.policy = policy

    def get_host_keys(self):
        client = self

        class _Keys:
            def load(self, path: str) -> None:
                with open(path) as f:
                    client.lines += f.read().splitlines()

        return _Keys()

    def connect(self, hostname, **kwargs) -> None:
        if self.lines and SshHostKey.of(SERVER_KEY).key not in self.lines[0]:
            raise paramiko.BadHostKeyException(hostname, SERVER_KEY, OTHER_KEY)
        if not self.lines:
            self.policy.missing_host_key(self, hostname, SERVER_KEY)
        self.connected_with = {"hostname": hostname, **kwargs}

    def get_transport(self):
        transport = MagicMock()
        transport.get_remote_server_key.return_value = SERVER_KEY
        return transport

    def exec_command(self, command, timeout=None):
        self.commands.append(command)
        return None, _Stream(b"ok"), _Stream(b"")

    def close(self) -> None:
        pass


@pytest.fixture
def client(monkeypatch) -> _Client:
    fake = _Client()
    monkeypatch.setattr(paramiko, "SSHClient", lambda: fake)
    return fake


def _driver(seen: list, known: dict | None = None) -> ProxmoxDriver:
    args: dict = {"port": 8006, OBSERVER_ARG: seen.append}
    if known is not None:
        args[KNOWN_KEYS_ARG] = known
    driver = ProxmoxDriver("10.0.0.9", "root", "s3cret", optional_args=args)
    driver._node_api = MagicMock(side_effect=RuntimeError("no exec endpoint"))  # type: ignore[method-assign]
    return driver


def test_first_use_runs_the_command_and_reports_the_key(client):
    seen: list = []
    assert _driver(seen)._run_node_command("true", timeout=5) == "ok"
    assert client.connected_with["port"] == 22
    assert seen[0]["accepted"] is True
    assert (seen[0]["host"], seen[0]["port"]) == ("10.0.0.9", 22)


def test_the_key_for_the_api_port_is_not_the_ssh_one(client):
    seen: list = []
    other = SshHostKey.of(OTHER_KEY).as_dict()
    _driver(seen, {"8006": other})._run_node_command("true", timeout=5)
    assert seen[0]["expected"] is None


def test_a_changed_key_stops_a_provisioning_step(client):
    seen: list = []
    driver = _driver(seen, {"22": SshHostKey.of(OTHER_KEY).as_dict()})
    with pytest.raises(SshHostKeyMismatch):
        driver._run_node_command("qm importdisk …", timeout=5)
    assert client.commands == []
    assert client.connected_with is None
    assert seen[0]["accepted"] is False


def test_a_changed_key_leaves_the_best_effort_read_empty(client):
    seen: list = []
    driver = _driver(seen, {"22": SshHostKey.of(OTHER_KEY).as_dict()})
    assert driver._exec_ssh_command("cat /etc/os-release") == ""
    assert client.commands == []
    assert seen[0]["accepted"] is False


def test_the_known_key_reuses_one_session(client):
    seen: list = []
    driver = _driver(seen, {"22": SshHostKey.of(SERVER_KEY).as_dict()})
    driver._exec_ssh_command("uptime")
    driver._run_node_command("true", timeout=5)
    assert client.commands == ["uptime", "true"]
    assert len(seen) == 1 and seen[0]["accepted"] is True