76 lines
2.5 KiB
Python
76 lines
2.5 KiB
Python
"""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
|
|
assert call_kwargs["user"] == "napalm@pam!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}
|