napalm-proxmox/tests/test_connection.py

"""Tests for ProxmoxDriver connection management."""

from __future__ import annotations

import pytest
from unittest.mock import MagicMock, patch

from napalm.base.exceptions import ConnectionException
from napalm_proxmox.driver import ProxmoxDriver
from tests.conftest import _build_mock_api, NODES_LIST


class TestOpen:
    def test_open_password_auth(self):
        drv = ProxmoxDriver("pve1", "root", "secret")
        mock_api = _build_mock_api()
        with patch("napalm_proxmox.driver.ProxmoxAPI", return_value=mock_api) as mock_cls:
            drv.open()
        call_kwargs = mock_cls.call_args.kwargs
        assert call_kwargs["user"] == "root@pam"
        assert call_kwargs["password"] == "secret"

    def test_open_token_auth(self):
        drv = ProxmoxDriver(
            "pve1",
            "root",
            "",
            optional_args={
                "token_name": "napalm@pam!mytoken",
                "token_value": "super-secret",
            },
        )
        mock_api = _build_mock_api()
        with patch("napalm_proxmox.driver.ProxmoxAPI", return_value=mock_api) as mock_cls:
            drv.open()
        call_kwargs = mock_cls.call_args.kwargs
        # proxmoxer wants the two halves separately, not the combined
        # "<user>!<tokenid>" string that Proxmox's UI displays.
        assert call_kwargs["user"] == "napalm@pam"
        assert call_kwargs["token_name"] == "mytoken"
        assert call_kwargs["token_value"] == "super-secret"

    def test_open_connection_error(self):
        drv = ProxmoxDriver("badhost", "root", "bad")
        with patch(
            "napalm_proxmox.driver.ProxmoxAPI",
            side_effect=Exception("Connection refused"),
        ):
            with pytest.raises(ConnectionException):
                drv.open()

    def test_node_resolution_by_hostname(self, driver):
        assert driver._node_name == "pve1"

    def test_node_override_via_optional_args(self):
        drv = ProxmoxDriver(
            "pve1.example.com",
            "root",
            "secret",
            optional_args={"node": "custom-node"},
        )
        mock_api = _build_mock_api()
        with patch("napalm_proxmox.driver.ProxmoxAPI", return_value=mock_api):
            drv.open()
        assert drv._node_name == "custom-node"

    def test_close(self, driver):
        driver.close()
        assert driver._api is None


class TestIsAlive:
    def test_alive_when_api_responds(self, driver):
        assert driver.is_alive() == {"is_alive": True}

    def test_not_alive_when_api_fails(self, driver):
        driver._api.version.get.side_effect = Exception("timeout")
        assert driver.is_alive() == {"is_alive": False}


# A four-node cluster as GET /cluster/status reports it. The node the API
# session landed on carries local=1 — here the third one, so taking the first
# entry of /nodes (the old behaviour) picks the wrong host.
CLUSTER_NODES = [
    {"type": "node", "name": "pve-01", "ip": "172.22.8.101", "local": 0},
    {"type": "node", "name": "pve-02", "ip": "172.22.8.102", "local": 0},
    {"type": "node", "name": "pve-dual", "ip": "172.22.8.120", "local": 1},
    {"type": "node", "name": "pve-garden", "ip": "172.22.8.5", "local": 0},
]
CLUSTER_STATUS = [{"type": "cluster", "name": "home", "nodes": 4}, *CLUSTER_NODES]
CLUSTER_NODES_LIST = [{"node": n["name"], "status": "online"} for n in CLUSTER_NODES]


def _open_with(hostname, **api_kwargs):
    drv = ProxmoxDriver(hostname, "root", "secret")
    mock_api = _build_mock_api(**api_kwargs)
    with patch("napalm_proxmox.driver.ProxmoxAPI", return_value=mock_api):
        drv.open()
    return drv


class TestClusterNodeResolution:
    def test_local_node_wins_over_first_listed(self):
        drv = _open_with(
            "172.22.8.120", nodes=CLUSTER_NODES_LIST, cluster_status=CLUSTER_STATUS
        )
        assert drv._node_name == "pve-dual"

    def test_matches_by_ip_without_local_flag(self):
        status = [{**n, "local": 0} for n in CLUSTER_STATUS if n["type"] == "node"]
        drv = _open_with("172.22.8.102", nodes=CLUSTER_NODES_LIST, cluster_status=status)
        assert drv._node_name == "pve-02"

    def test_matches_by_name_without_local_flag(self):
        status = [{**n, "local": 0} for n in CLUSTER_STATUS if n["type"] == "node"]
        drv = _open_with("pve-garden", nodes=CLUSTER_NODES_LIST, cluster_status=status)
        assert drv._node_name == "pve-garden"

    def test_matches_short_name_of_fqdn(self):
        status = [{**n, "local": 0} for n in CLUSTER_STATUS if n["type"] == "node"]
        drv = _open_with(
            "pve-01.mgmt.example.com", nodes=CLUSTER_NODES_LIST, cluster_status=status
        )
        assert drv._node_name == "pve-01"

    def test_single_node_without_cluster_status(self):
        drv = _open_with("10.0.0.9", nodes=[{"node": "solo", "status": "online"}])
        assert drv._node_name == "solo"

    def test_ambiguous_cluster_refuses_to_guess(self):
        status = [{**n, "local": 0} for n in CLUSTER_STATUS if n["type"] == "node"]
        with pytest.raises(ConnectionException, match="node"):
            _open_with("10.9.9.9", nodes=CLUSTER_NODES_LIST, cluster_status=status)

    def test_api_error_fails_open_instead_of_guessing(self):
        # A TLS failure used to be swallowed here: the IP became the node
        # name, open() succeeded and every later call failed quietly.
        drv = ProxmoxDriver("172.22.8.120", "root", "secret")
        mock_api = _build_mock_api()
        mock_api.cluster.status.get.side_effect = Exception("CERTIFICATE_VERIFY_FAILED")
        mock_api.nodes.get.side_effect = Exception("CERTIFICATE_VERIFY_FAILED")
        with patch("napalm_proxmox.driver.ProxmoxAPI", return_value=mock_api):
            with pytest.raises(ConnectionException, match="CERTIFICATE_VERIFY_FAILED"):
                drv.open()

    def test_explicit_node_skips_lookup(self):
        drv = ProxmoxDriver("172.22.8.120", "root", "secret", optional_args={"node": "pve-dual"})
        mock_api = _build_mock_api()
        with patch("napalm_proxmox.driver.ProxmoxAPI", return_value=mock_api):
            drv.open()
        assert drv._node_name == "pve-dual"
        mock_api.cluster.status.get.assert_not_called()
        mock_api.nodes.get.assert_not_called()