Christian Manivong
890e84ca6b
CI / test (3.10) (push) Successful in 33s
CI / test (3.11) (push) Successful in 32s
CI / test (3.12) (push) Successful in 35s
CI / test (3.10) (pull_request) Successful in 32s
CI / test (3.11) (pull_request) Successful in 30s
CI / test (3.12) (pull_request) Successful in 35s
fix_snmp overwrote snmpd.conf with an application's values (community "public", "Managed by netOrk", netork@localhost) through a fixed /tmp/netork_snmpd.conf, and opened ufw for a /24 guessed from ss; install_lldpd was an action of its own. Neither read an exit status. The driver implements napalm-device-types 4.13's templates instead: - configure_snmp_agent(settings): net-snmp installed, snmpd.conf with one rocommunity line per allowed source (mktemp, install as root), ufw opened for those sources only (pve-firewall stays the operator's), snmpd enabled and restarted; - get_management_peer() from SSH_CONNECTION; - install_package(name), as the contract says (raises when apt fails), so ensure_package comes from the base package. Every step reads its exit status. BREAKING CHANGE: run_device_action is removed. Requires napalm-device-types 4.13. Part of #24.
119 lines
4.2 KiB
Python
119 lines
4.2 KiB
Python
"""The fixes as napalm-device-types templates, with the caller's values (netOrk #1105).
|
|
|
|
``fix_snmp`` wrote an application's snmpd.conf (community "public", "Managed by
|
|
netOrk") and opened ufw for the caller's /24 guessed from ``ss``;
|
|
``install_lldpd`` was an action of its own. Now ``configure_snmp_agent(settings)``
|
|
takes every value from the caller, and ``ensure_package`` installs through the
|
|
contract's ``install_package``. Every step reads its exit status.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
|
|
import pytest
|
|
|
|
from napalm_proxmox import ProxmoxDriver
|
|
|
|
SETTINGS = {"community": "c0mm", "allowed_sources": ["10.7.224.20"], "location": "rack 1"}
|
|
|
|
|
|
class _Node:
|
|
"""The node's shell: answers by command fragment, with an exit status."""
|
|
|
|
def __init__(self, **status):
|
|
self.status = status # fragment -> exit status
|
|
self.ran: list[str] = []
|
|
|
|
def __call__(self, command: str) -> str:
|
|
self.ran.append(command)
|
|
if "SSH_CONNECTION" in command:
|
|
return "10.7.224.20 51234 10.0.0.9 22\n"
|
|
if command.startswith("dpkg-query"):
|
|
return "curl\tinstalled\n"
|
|
rc = next((code for frag, code in self.status.items() if frag in command), 0)
|
|
out = "ok" if rc == 0 else "E: failed"
|
|
return f"{out}\n__PVE_RC={rc}" if "__PVE_RC" in command else out
|
|
|
|
|
|
@pytest.fixture
|
|
def node():
|
|
return _Node()
|
|
|
|
|
|
@pytest.fixture
|
|
def driver(node):
|
|
d = ProxmoxDriver("pve", "root@pam", "pw")
|
|
d._exec_ssh_command = node
|
|
return d
|
|
|
|
|
|
class TestSnmpAgent:
|
|
def test_install_write_open_start_with_the_callers_values(self, driver, node):
|
|
result = driver.configure_snmp_agent(SETTINGS)
|
|
|
|
assert result["success"] is True
|
|
assert [s["step"] for s in result["steps"]] == ["install", "configure", "firewall", "start"]
|
|
assert any("apt-get install -y snmpd snmp" in c for c in node.ran)
|
|
write = next(c for c in node.ran if "/etc/snmp/snmpd.conf" in c)
|
|
assert "install -m 644" in write
|
|
b64 = re.search(r"echo ([A-Za-z0-9+/=]+) \| base64 -d", write).group(1)
|
|
import base64
|
|
|
|
conf = base64.b64decode(b64).decode()
|
|
assert "rocommunity c0mm 10.7.224.20/32" in conf
|
|
assert "sysLocation rack 1" in conf
|
|
assert "public" not in conf and "netork" not in conf.lower()
|
|
assert any("systemctl restart snmpd" in c for c in node.ran)
|
|
|
|
def test_no_temp_file_with_an_application_name(self, driver, node):
|
|
driver.configure_snmp_agent(SETTINGS)
|
|
|
|
assert not any("netork" in c for c in node.ran)
|
|
|
|
def test_ufw_opens_only_for_the_allowed_sources(self, driver, node):
|
|
node.status["command -v ufw"] = 0
|
|
|
|
driver.configure_snmp_agent(SETTINGS)
|
|
|
|
assert any("ufw allow from 10.7.224.20/32 to any port 161 proto udp" in c for c in node.ran)
|
|
assert not any(".0/24" in c for c in node.ran)
|
|
|
|
def test_a_failed_install_stops_the_fix(self, driver, node):
|
|
node.status["apt-get install -y snmpd"] = 100
|
|
|
|
result = driver.configure_snmp_agent(SETTINGS)
|
|
|
|
assert result["success"] is False
|
|
assert not any("/etc/snmp/snmpd.conf" in c for c in node.ran)
|
|
|
|
def test_the_management_peer(self, driver):
|
|
assert driver.get_management_peer() == "10.7.224.20"
|
|
|
|
|
|
class TestPackages:
|
|
def test_ensure_package_installs_what_is_missing(self, driver, node):
|
|
result = driver.ensure_package("lldpd")
|
|
|
|
assert result["success"] is True
|
|
assert any("apt-get install -y lldpd" in c for c in node.ran)
|
|
|
|
def test_ensure_package_leaves_alone_what_is_there(self, driver, node):
|
|
assert driver.ensure_package("curl")["success"] is True
|
|
assert not any("apt-get install" in c for c in node.ran)
|
|
|
|
def test_install_package_raises_when_apt_fails(self, driver, node):
|
|
node.status["apt-get install -y lldpd"] = 100
|
|
|
|
with pytest.raises(RuntimeError, match="lldpd"):
|
|
driver.install_package("lldpd")
|
|
|
|
def test_a_name_that_is_no_package_name_is_refused(self, driver):
|
|
with pytest.raises(ValueError):
|
|
driver.install_package("lldpd; reboot")
|
|
|
|
|
|
def test_the_old_actions_are_gone():
|
|
for name in ("run_device_action", "_action_fix_snmp", "_action_install_lldpd"):
|
|
assert not hasattr(ProxmoxDriver, name), name
|