diff --git a/changelog/69793.fixed.md b/changelog/69793.fixed.md new file mode 100644 index 000000000000..8e00dd50bfd5 --- /dev/null +++ b/changelog/69793.fixed.md @@ -0,0 +1,9 @@ +Fixed NTP, SNMP and RPM-probe configuration on NAPALM (proxy) minions. +``ntp.set_peers`` / ``set_servers`` / ``delete_peers`` / ``delete_servers``, +``snmp.update_config`` / ``remove_config`` and ``probes.set_probes`` / +``delete_probes`` / ``schedule_probes`` no longer fail with ``Local file source +set_ntp_peers does not exist``. Like ``users.set_users`` (see #62170), these +functions passed bare template names to ``net.load_template``, which stopped +resolving when native NAPALM template support was removed in the Sodium release. +They now resolve the NAPALM-shipped per-driver template to an absolute path and +render it through the Salt pipeline. diff --git a/salt/modules/napalm_ntp.py b/salt/modules/napalm_ntp.py index d737c149e23f..1f865f62033d 100644 --- a/salt/modules/napalm_ntp.py +++ b/salt/modules/napalm_ntp.py @@ -230,8 +230,11 @@ def set_peers(*peers, **options): commit = options.pop("commit", True) # pylint: disable=undefined-variable + resolved = salt.utils.napalm.template_path(napalm_device, "set_ntp_peers") + if resolved is None: + return salt.utils.napalm.template_not_available("set_ntp_peers", napalm_device) return __salt__["net.load_template"]( - "set_ntp_peers", + resolved, peers=peers, test=test, commit=commit, @@ -268,8 +271,13 @@ def set_servers(*servers, **options): commit = options.pop("commit", True) # pylint: disable=undefined-variable + resolved = salt.utils.napalm.template_path(napalm_device, "set_ntp_servers") + if resolved is None: + return salt.utils.napalm.template_not_available( + "set_ntp_servers", napalm_device + ) return __salt__["net.load_template"]( - "set_ntp_servers", + resolved, servers=servers, test=test, commit=commit, @@ -307,8 +315,13 @@ def delete_peers(*peers, **options): commit = options.pop("commit", True) # pylint: disable=undefined-variable + resolved = salt.utils.napalm.template_path(napalm_device, "delete_ntp_peers") + if resolved is None: + return salt.utils.napalm.template_not_available( + "delete_ntp_peers", napalm_device + ) return __salt__["net.load_template"]( - "delete_ntp_peers", + resolved, peers=peers, test=test, commit=commit, @@ -347,8 +360,13 @@ def delete_servers(*servers, **options): commit = options.pop("commit", True) # pylint: disable=undefined-variable + resolved = salt.utils.napalm.template_path(napalm_device, "delete_ntp_servers") + if resolved is None: + return salt.utils.napalm.template_not_available( + "delete_ntp_servers", napalm_device + ) return __salt__["net.load_template"]( - "delete_ntp_servers", + resolved, servers=servers, test=test, commit=commit, diff --git a/salt/modules/napalm_probes.py b/salt/modules/napalm_probes.py index 132b984d6b19..3bf37858cedc 100644 --- a/salt/modules/napalm_probes.py +++ b/salt/modules/napalm_probes.py @@ -254,8 +254,11 @@ def set_probes( """ # pylint: disable=undefined-variable + resolved = salt.utils.napalm.template_path(napalm_device, "set_probes") + if resolved is None: + return salt.utils.napalm.template_not_available("set_probes", napalm_device) return __salt__["net.load_template"]( - "set_probes", + resolved, probes=probes, test=test, commit=commit, @@ -310,8 +313,11 @@ def delete_probes( """ # pylint: disable=undefined-variable + resolved = salt.utils.napalm.template_path(napalm_device, "delete_probes") + if resolved is None: + return salt.utils.napalm.template_not_available("delete_probes", napalm_device) return __salt__["net.load_template"]( - "delete_probes", + resolved, probes=probes, test=test, commit=commit, @@ -367,8 +373,13 @@ def schedule_probes( """ # pylint: disable=undefined-variable + resolved = salt.utils.napalm.template_path(napalm_device, "schedule_probes") + if resolved is None: + return salt.utils.napalm.template_not_available( + "schedule_probes", napalm_device + ) return __salt__["net.load_template"]( - "schedule_probes", + resolved, probes=probes, test=test, commit=commit, diff --git a/salt/modules/napalm_snmp.py b/salt/modules/napalm_snmp.py index ee6ca24b38ba..5d090b90c70c 100644 --- a/salt/modules/napalm_snmp.py +++ b/salt/modules/napalm_snmp.py @@ -129,7 +129,14 @@ def remove_config( salt '*' snmp.remove_config community='abcd' """ - dic = {"template_name": "delete_snmp_config", "test": test, "commit": commit} + resolved = salt.utils.napalm.template_path( + napalm_device, "delete_snmp_config" # pylint: disable=undefined-variable + ) + if resolved is None: + return salt.utils.napalm.template_not_available( + "delete_snmp_config", napalm_device # pylint: disable=undefined-variable + ) + dic = {"template_name": resolved, "test": test, "commit": commit} if chassis_id: dic["chassis_id"] = chassis_id @@ -206,7 +213,14 @@ def update_config( True """ - dic = {"template_name": "snmp_config", "test": test, "commit": commit} + resolved = salt.utils.napalm.template_path( + napalm_device, "snmp_config" # pylint: disable=undefined-variable + ) + if resolved is None: + return salt.utils.napalm.template_not_available( + "snmp_config", napalm_device # pylint: disable=undefined-variable + ) + dic = {"template_name": resolved, "test": test, "commit": commit} if chassis_id: dic["chassis_id"] = chassis_id diff --git a/salt/utils/napalm.py b/salt/utils/napalm.py index 9fc15e45fdb1..cc75bd547196 100644 --- a/salt/utils/napalm.py +++ b/salt/utils/napalm.py @@ -16,7 +16,9 @@ import copy import importlib +import inspect import logging +import os.path import traceback from functools import wraps @@ -100,6 +102,67 @@ def virtual(opts, virtualname, filename): ) +def template_path(napalm_device, template_name): + """ + Return the absolute path to a NAPALM-shipped Jinja template (e.g. + ``set_ntp_peers``) for the driver backing this proxy, or ``None`` if the + driver does not ship one. + + NAPALM keeps these config templates in a ``templates`` directory next to + each driver module and resolves them by walking the driver class MRO + (concrete driver first, then its bases). ``net.load_template`` used to route + bare template names into NAPALM's own renderer, but that path was removed in + the Sodium release; resolving the template to an absolute path lets the + still-supported Salt rendering pipeline render it instead. + """ + driver = napalm_device.get("DRIVER") if napalm_device else None + if driver is None: + return None + for klass in type(driver).__mro__: + try: + module_file = inspect.getfile(klass) + except (TypeError, OSError): + # Built-in types (e.g. ``object``) raise TypeError; classes without + # an on-disk source (``__main__``, frozen) raise OSError. Neither + # can ship a template dir, so move on. + continue + candidate = os.path.join( + os.path.dirname(module_file), "templates", f"{template_name}.j2" + ) + if os.path.isfile(candidate): + return candidate + return None + + +def template_not_available(template_name, napalm_device): + """ + Standard failure payload returned by the NAPALM config helpers when the + driver backing this proxy does not ship ``template_name`` (e.g. the ``ios`` + driver has no user templates). Mirrors the shape of ``net.load_template``'s + return so callers and states handle it uniformly. + + Returning here short-circuits ``net.load_template``, which would otherwise be + responsible for closing the per-call connection a non-always-alive proxy / + minion opened (``proxy_napalm_wrap`` only closes on ``force_reconnect``). So + close it here in that mode to avoid leaking the session. + """ + driver_name = napalm_device.get("DRIVER_NAME") if napalm_device else None + opts = napalm_device.get("__opts__") if napalm_device else None + if opts and not_always_alive(opts) and napalm_device.get("CLOSE", True): + try: + napalm_device["DRIVER"].close() + except Exception: # pylint: disable=broad-except + log.debug("Failed to close the connection", exc_info=True) + return { + "result": False, + "out": None, + "comment": ( + f"The '{template_name}' template is not available for the" + f" '{driver_name}' driver." + ), + } + + def call(napalm_device, method, *args, **kwargs): """ Calls arbitrary methods from the network driver instance. diff --git a/tests/pytests/unit/modules/napalm/test_ntp.py b/tests/pytests/unit/modules/napalm/test_ntp.py index 1f146603e041..4404abe0b4ab 100644 --- a/tests/pytests/unit/modules/napalm/test_ntp.py +++ b/tests/pytests/unit/modules/napalm/test_ntp.py @@ -9,89 +9,93 @@ from tests.support.mock import MagicMock, patch -def mock_net_load_template(template, *args, **kwargs): - if template == "set_ntp_peers" or template == "delete_ntp_peers": - assert "1.2.3.4" in kwargs["peers"] - if template == "set_ntp_servers" or template == "delete_ntp_servers": - assert "2.2.3.4" in kwargs["servers"] - - @pytest.fixture def configure_loader_modules(): - module_globals = { - "__salt__": { - "config.get": MagicMock( - return_value={"test": {"driver": "test", "key": "2orgk34kgk34g"}} - ), - "file.file_exists": napalm_test_support.true, - "file.join": napalm_test_support.join, - "file.get_managed": napalm_test_support.get_managed_file, - "random.hash": napalm_test_support.random_hash, - "net.load_template": mock_net_load_template, - } - } - - return {napalm_ntp: module_globals} - - -def test_peers(): - with patch( - "salt.utils.napalm.get_device", - MagicMock(return_value=napalm_test_support.MockNapalmDevice()), - ): - ret = napalm_ntp.peers() - assert "172.17.17.1" in ret["out"] + return {napalm_ntp: {"__salt__": {"config.get": MagicMock(return_value={})}}} -def test_servers(): - with patch( - "salt.utils.napalm.get_device", - MagicMock(return_value=napalm_test_support.MockNapalmDevice()), - ): - ret = napalm_ntp.servers() - assert "172.17.17.1" in ret["out"] - - -def test_stats(): - with patch( +def _mock_device(): + return patch( "salt.utils.napalm.get_device", MagicMock(return_value=napalm_test_support.MockNapalmDevice()), - ): - ret = napalm_ntp.stats() - assert ret["out"][0]["reachability"] == 377 + ) -def test_set_peers(): - with patch( - "salt.utils.napalm.get_device", - MagicMock(return_value=napalm_test_support.MockNapalmDevice()), - ): - ret = napalm_ntp.set_peers("1.2.3.4", "5.6.7.8") - assert ret is None +# --- read pass-throughs (unchanged behaviour) ------------------------------- -def test_set_servers(): - with patch( - "salt.utils.napalm.get_device", - MagicMock(return_value=napalm_test_support.MockNapalmDevice()), - ): - ret = napalm_ntp.set_servers("2.2.3.4", "6.6.7.8") - assert ret is None +def test_peers(): + with _mock_device(): + assert "172.17.17.1" in napalm_ntp.peers()["out"] -def test_delete_servers(): - with patch( - "salt.utils.napalm.get_device", - MagicMock(return_value=napalm_test_support.MockNapalmDevice()), - ): - ret = napalm_ntp.delete_servers("2.2.3.4", "6.6.7.8") - assert ret is None +def test_servers(): + with _mock_device(): + assert "172.17.17.1" in napalm_ntp.servers()["out"] -def test_delete_peers(): - with patch( - "salt.utils.napalm.get_device", - MagicMock(return_value=napalm_test_support.MockNapalmDevice()), +def test_stats(): + with _mock_device(): + assert napalm_ntp.stats()["out"][0]["reachability"] == 377 + + +# --- config writers: route onto the driver's resolved template (#62170) ----- + +RESOLVED = "/opt/napalm/junos/templates/tpl.j2" + + +def _route(func, *args): + """Run a config writer with template resolution + net.load_template mocked.""" + device = napalm_test_support.MockNapalmDevice() + load_template = MagicMock(return_value={"result": True, "comment": "", "out": None}) + tpath = MagicMock(return_value=RESOLVED) + with patch("salt.utils.napalm.get_device", MagicMock(return_value=device)), patch( + "salt.utils.napalm.template_path", tpath + ), patch.dict(napalm_ntp.__salt__, {"net.load_template": load_template}): + ret = func(*args, test=True, commit=False) + return ret, tpath, load_template, device + + +@pytest.mark.parametrize( + "func_name, template_name, arg_key, values", + [ + ("set_peers", "set_ntp_peers", "peers", ("1.2.3.4", "5.6.7.8")), + ("set_servers", "set_ntp_servers", "servers", ("2.2.3.4", "6.6.7.8")), + ("delete_peers", "delete_ntp_peers", "peers", ("1.2.3.4", "5.6.7.8")), + ("delete_servers", "delete_ntp_servers", "servers", ("2.2.3.4", "6.6.7.8")), + ], +) +def test_writer_routes_resolved_template(func_name, template_name, arg_key, values): + ret, tpath, load_template, device = _route(getattr(napalm_ntp, func_name), *values) + assert ret == {"result": True, "comment": "", "out": None} + # Correct template name requested (guards a set/delete or peers/servers mixup). + assert tpath.call_args[0][1] == template_name + load_template.assert_called_once() + args, kwargs = load_template.call_args + # Absolute resolved path passed, not the bare name. + assert args[0] == RESOLVED + assert values[0] in kwargs[arg_key] + assert kwargs["test"] is True + assert kwargs["commit"] is False + # The open proxy device is threaded through by identity (guards =None / wrong). + assert kwargs["inherit_napalm_device"] is device + + +@pytest.mark.parametrize( + "func_name, template_name", + [ + ("set_peers", "set_ntp_peers"), + ("set_servers", "set_ntp_servers"), + ("delete_peers", "delete_ntp_peers"), + ("delete_servers", "delete_ntp_servers"), + ], +) +def test_writer_no_template_for_driver(func_name, template_name): + with _mock_device(), patch( + "salt.utils.napalm.template_path", MagicMock(return_value=None) ): - ret = napalm_ntp.delete_peers("1.2.3.4", "5.6.7.8") - assert ret is None + ret = getattr(napalm_ntp, func_name)("1.2.3.4") + assert ret["result"] is False + # Exact quoted name (guards a set/delete literal swap in the failure branch). + assert f"'{template_name}'" in ret["comment"] + assert "not available" in ret["comment"] diff --git a/tests/pytests/unit/modules/napalm/test_probes.py b/tests/pytests/unit/modules/napalm/test_probes.py new file mode 100644 index 000000000000..3333b56ed5d0 --- /dev/null +++ b/tests/pytests/unit/modules/napalm/test_probes.py @@ -0,0 +1,94 @@ +""" +Unit tests for the napalm_probes execution module. +""" + +import pytest + +import salt.modules.napalm_probes as napalm_probes +import tests.support.napalm as napalm_test_support +from tests.support.mock import MagicMock, patch + + +@pytest.fixture +def configure_loader_modules(): + return {napalm_probes: {"__salt__": {"config.get": MagicMock(return_value={})}}} + + +def _mock_device(): + return patch( + "salt.utils.napalm.get_device", + MagicMock(return_value=napalm_test_support.MockNapalmDevice()), + ) + + +# --- read pass-throughs ----------------------------------------------------- + + +def test_config(): + with _mock_device(): + ret = napalm_probes.config() + assert ret["result"] is True + assert ret["out"] == napalm_test_support.TEST_PROBES_CONFIG.copy() + + +def test_results(): + with _mock_device(): + ret = napalm_probes.results() + assert ret["result"] is True + assert ret["out"] == napalm_test_support.TEST_PROBES_RESULTS.copy() + + +# --- config writers: route onto the driver's resolved template (#62170) ----- + +RESOLVED = "/opt/napalm/junos/templates/tpl.j2" +PROBES = {"new_probe": {"new_test1": {}}} + + +def _route(func): + device = napalm_test_support.MockNapalmDevice() + load_template = MagicMock(return_value={"result": True, "comment": "", "out": None}) + tpath = MagicMock(return_value=RESOLVED) + with patch("salt.utils.napalm.get_device", MagicMock(return_value=device)), patch( + "salt.utils.napalm.template_path", tpath + ), patch.dict(napalm_probes.__salt__, {"net.load_template": load_template}): + ret = func(PROBES, test=True, commit=False) + return ret, tpath, load_template, device + + +@pytest.mark.parametrize( + "func_name, template_name", + [ + ("set_probes", "set_probes"), + ("delete_probes", "delete_probes"), + ("schedule_probes", "schedule_probes"), + ], +) +def test_writer_routes_resolved_template(func_name, template_name): + ret, tpath, load_template, device = _route(getattr(napalm_probes, func_name)) + assert ret == {"result": True, "comment": "", "out": None} + assert tpath.call_args[0][1] == template_name + load_template.assert_called_once() + args, kwargs = load_template.call_args + assert args[0] == RESOLVED + assert kwargs["probes"] == PROBES + assert kwargs["test"] is True + assert kwargs["commit"] is False + assert kwargs["inherit_napalm_device"] is device + + +@pytest.mark.parametrize( + "func_name, template_name", + [ + ("set_probes", "set_probes"), + ("delete_probes", "delete_probes"), + ("schedule_probes", "schedule_probes"), + ], +) +def test_writer_no_template_for_driver(func_name, template_name): + with _mock_device(), patch( + "salt.utils.napalm.template_path", MagicMock(return_value=None) + ): + ret = getattr(napalm_probes, func_name)(PROBES) + assert ret["result"] is False + assert f"'{template_name}'" in ret["comment"] + assert "not available" in ret["comment"] diff --git a/tests/pytests/unit/modules/napalm/test_snmp.py b/tests/pytests/unit/modules/napalm/test_snmp.py index 5166372ca68c..f26547015a78 100644 --- a/tests/pytests/unit/modules/napalm/test_snmp.py +++ b/tests/pytests/unit/modules/napalm/test_snmp.py @@ -4,7 +4,6 @@ import pytest -import salt.modules.napalm_network as napalm_network import salt.modules.napalm_snmp as napalm_snmp import tests.support.napalm as napalm_test_support from tests.support.mock import MagicMock, patch @@ -12,44 +11,75 @@ @pytest.fixture def configure_loader_modules(): - module_globals = { - "__salt__": { - "config.get": MagicMock( - return_value={"test": {"driver": "test", "key": "2orgk34kgk34g"}} - ), - "file.file_exists": napalm_test_support.true, - "file.join": napalm_test_support.join, - "file.get_managed": napalm_test_support.get_managed_file, - "random.hash": napalm_test_support.random_hash, - "net.load_template": napalm_network.load_template, - } - } - - return {napalm_snmp: module_globals, napalm_network: module_globals} + return {napalm_snmp: {"__salt__": {"config.get": MagicMock(return_value={})}}} -def test_config(): - with patch( +def _mock_device(): + return patch( "salt.utils.napalm.get_device", MagicMock(return_value=napalm_test_support.MockNapalmDevice()), - ): + ) + + +def test_config(): + with _mock_device(): ret = napalm_snmp.config() assert ret["out"] == napalm_test_support.TEST_SNMP_INFO.copy() -def test_remove_config(): - with patch( - "salt.utils.napalm.get_device", - MagicMock(return_value=napalm_test_support.MockNapalmDevice()), - ): - ret = napalm_snmp.remove_config("1.2.3.4") - assert ret["result"] is False +# --- config writers: route onto the driver's resolved template (#62170) ----- +RESOLVED = "/opt/napalm/junos/templates/tpl.j2" -def test_update_config(): - with patch( - "salt.utils.napalm.get_device", - MagicMock(return_value=napalm_test_support.MockNapalmDevice()), + +def _route(func, **kwargs): + device = napalm_test_support.MockNapalmDevice() + load_template = MagicMock(return_value={"result": True, "comment": "", "out": None}) + tpath = MagicMock(return_value=RESOLVED) + with patch("salt.utils.napalm.get_device", MagicMock(return_value=device)), patch( + "salt.utils.napalm.template_path", tpath + ), patch.dict(napalm_snmp.__salt__, {"net.load_template": load_template}): + ret = func(test=True, commit=False, **kwargs) + return ret, tpath, load_template, device + + +@pytest.mark.parametrize( + "func_name, template_name", + [ + ("update_config", "snmp_config"), + ("remove_config", "delete_snmp_config"), + ], +) +def test_writer_routes_resolved_template(func_name, template_name): + ret, tpath, load_template, device = _route( + getattr(napalm_snmp, func_name), location="Greenwich, UK" + ) + assert ret == {"result": True, "comment": "", "out": None} + assert tpath.call_args[0][1] == template_name + load_template.assert_called_once() + _args, kwargs = load_template.call_args + # snmp builds a dict and calls net.load_template(**dic), so the resolved + # path arrives as the template_name kwarg, not positionally. + assert kwargs["template_name"] == RESOLVED + assert kwargs["location"] == "Greenwich, UK" + assert kwargs["test"] is True + assert kwargs["commit"] is False + assert kwargs["inherit_napalm_device"] is device + + +@pytest.mark.parametrize( + "func_name, template_name", + [ + ("update_config", "snmp_config"), + ("remove_config", "delete_snmp_config"), + ], +) +def test_writer_no_template_for_driver(func_name, template_name): + with _mock_device(), patch( + "salt.utils.napalm.template_path", MagicMock(return_value=None) ): - ret = napalm_snmp.update_config("1.2.3.4") - assert ret["result"] is False + ret = getattr(napalm_snmp, func_name)(location="x") + assert ret["result"] is False + # Exact quoted name: "'snmp_config'" must not match "'delete_snmp_config'". + assert f"'{template_name}'" in ret["comment"] + assert "not available" in ret["comment"] diff --git a/tests/pytests/unit/utils/test_napalm.py b/tests/pytests/unit/utils/test_napalm.py new file mode 100644 index 000000000000..37640a0cffe2 --- /dev/null +++ b/tests/pytests/unit/utils/test_napalm.py @@ -0,0 +1,163 @@ +""" +Unit tests for salt.utils.napalm helpers. +""" + +import salt.utils.napalm as napalm_utils +from tests.support.mock import MagicMock, patch + + +class _BaseDriver: + pass + + +class _ConcreteDriver(_BaseDriver): + pass + + +def _getfile_map(mapping): + """ + Build an ``inspect.getfile`` replacement that returns a distinct path per + class and raises (like the real one) for anything not in the map -- notably + ``object``, so the resolver's exception-continue is genuinely exercised. + """ + + def fake_getfile(klass): + try: + return mapping[klass] + except KeyError: + raise TypeError(f"{klass!r} is a built-in class") + + return fake_getfile + + +def _ship(directory, template_name): + tpl_dir = directory / "templates" + tpl_dir.mkdir(parents=True) + (tpl_dir / f"{template_name}.j2").write_text("system { }") + return tpl_dir / f"{template_name}.j2" + + +def test_template_path_walks_mro_to_base(tmp_path): + """ + Templates can be inherited: the concrete driver ships none but a base class + does. The resolver must walk the MRO (concrete -> base) and skip ``object`` + (which raises from getfile) rather than stopping at the first class. + """ + (tmp_path / "concrete").mkdir() + base_tpl = _ship(tmp_path / "base", "set_ntp_peers") + + device = {"DRIVER": _ConcreteDriver()} + getfile = _getfile_map( + { + _ConcreteDriver: str(tmp_path / "concrete" / "driver.py"), + _BaseDriver: str(tmp_path / "base" / "base.py"), + } + ) + with patch("salt.utils.napalm.inspect.getfile", side_effect=getfile): + resolved = napalm_utils.template_path(device, "set_ntp_peers") + assert resolved == str(base_tpl) + + +def test_template_path_prefers_concrete_over_base(tmp_path): + """ + When both the concrete driver and a base class ship the same template, the + concrete override wins -- the walk must be concrete-first, not reversed. + """ + concrete_tpl = _ship(tmp_path / "concrete", "set_ntp_peers") + _ship(tmp_path / "base", "set_ntp_peers") + + device = {"DRIVER": _ConcreteDriver()} + getfile = _getfile_map( + { + _ConcreteDriver: str(tmp_path / "concrete" / "driver.py"), + _BaseDriver: str(tmp_path / "base" / "base.py"), + } + ) + with patch("salt.utils.napalm.inspect.getfile", side_effect=getfile): + resolved = napalm_utils.template_path(device, "set_ntp_peers") + assert resolved == str(concrete_tpl) + + +def test_template_path_skips_oserror(tmp_path): + """ + ``inspect.getfile`` raises ``OSError`` for classes with no on-disk source + (frozen / ``__main__``); that class must be skipped, not propagated. + """ + base_tpl = _ship(tmp_path / "base", "set_ntp_peers") + + def getfile(klass): + if klass is _BaseDriver: + return str(tmp_path / "base" / "base.py") + raise OSError("source code not available") # _ConcreteDriver + object + + device = {"DRIVER": _ConcreteDriver()} + with patch("salt.utils.napalm.inspect.getfile", side_effect=getfile): + resolved = napalm_utils.template_path(device, "set_ntp_peers") + assert resolved == str(base_tpl) + + +def test_template_path_missing_returns_none(tmp_path): + """ + Drivers that ship no matching template anywhere in the MRO (e.g. ios has no + user templates) resolve to ``None``, and a device with no / an empty / + a missing ``DRIVER`` is handled too. + """ + (tmp_path / "concrete").mkdir() + (tmp_path / "base").mkdir() + device = {"DRIVER": _ConcreteDriver()} + getfile = _getfile_map( + { + _ConcreteDriver: str(tmp_path / "concrete" / "driver.py"), + _BaseDriver: str(tmp_path / "base" / "base.py"), + } + ) + with patch("salt.utils.napalm.inspect.getfile", side_effect=getfile): + assert napalm_utils.template_path(device, "set_ntp_peers") is None + # A truthy device whose DRIVER key is absent -> None (not a KeyError). + assert napalm_utils.template_path({"NOT_DRIVER": object()}, "x") is None + # No device / driver at all is handled too. + assert napalm_utils.template_path({}, "x") is None + assert napalm_utils.template_path(None, "x") is None + + +def test_template_not_available_shape(): + """ + The failure payload mirrors net.load_template's shape and names the driver. + """ + ret = napalm_utils.template_not_available("set_ntp_peers", {"DRIVER_NAME": "ios"}) + assert ret["result"] is False + assert ret["out"] is None + assert ( + ret["comment"] + == "The 'set_ntp_peers' template is not available for the 'ios' driver." + ) + # Tolerates a missing / None device without raising. + assert napalm_utils.template_not_available("x", None)["result"] is False + + +def test_template_not_available_closes_non_always_alive(): + """ + Because it short-circuits net.load_template, the failure path must close the + per-call connection a non-always-alive proxy / minion opened. + """ + driver = MagicMock() + device = {"DRIVER": driver, "DRIVER_NAME": "junos", "__opts__": {"id": "sw01"}} + with patch("salt.utils.napalm.not_always_alive", MagicMock(return_value=True)): + napalm_utils.template_not_available("set_ntp_peers", device) + driver.close.assert_called_once() + + +def test_template_not_available_leaves_always_alive_open(): + """An always-alive proxy's persistent session must NOT be closed here.""" + driver = MagicMock() + device = {"DRIVER": driver, "DRIVER_NAME": "junos", "__opts__": {"id": "sw01"}} + with patch("salt.utils.napalm.not_always_alive", MagicMock(return_value=False)): + napalm_utils.template_not_available("set_ntp_peers", device) + driver.close.assert_not_called() + # CLOSE explicitly False also suppresses the close. + with patch("salt.utils.napalm.not_always_alive", MagicMock(return_value=True)): + napalm_utils.template_not_available( + "set_ntp_peers", + {"DRIVER": driver, "__opts__": {"id": "sw01"}, "CLOSE": False}, + ) + driver.close.assert_not_called() diff --git a/tests/unit/modules/test_napalm_probes.py b/tests/unit/modules/test_napalm_probes.py deleted file mode 100644 index 6eaa80b83104..000000000000 --- a/tests/unit/modules/test_napalm_probes.py +++ /dev/null @@ -1,90 +0,0 @@ -""" - :codeauthor: :email:`Anthony Shaw ` -""" - -import salt.modules.napalm_probes as napalm_probes -import tests.support.napalm as napalm_test_support -from tests.support.mixins import LoaderModuleMockMixin -from tests.support.mock import MagicMock, patch -from tests.support.unit import TestCase - - -class NapalmProbesModuleTestCase(TestCase, LoaderModuleMockMixin): - @classmethod - def setUpClass(cls): - cls._test_probes = { - "new_probe": { - "new_test1": { - "probe_type": "icmp-ping", - "target": "192.168.0.1", - "source": "192.168.0.2", - "probe_count": 13, - "test_interval": 3, - } - } - } - cls._test_delete_probes = { - "existing_probe": {"existing_test1": {}, "existing_test2": {}} - } - cls._test_schedule_probes = { - "test_probe": {"existing_test1": {}, "existing_test2": {}} - } - - @classmethod - def tearDownClass(cls): - cls._test_probes = cls._test_delete_probes = cls._test_schedule_probes = None - - def setup_loader_modules(self): - patcher = patch( - "salt.utils.napalm.get_device", - MagicMock(return_value=napalm_test_support.MockNapalmDevice()), - ) - patcher.start() - self.addCleanup(patcher.stop) - - def mock_net_load(template, *args, **kwargs): - if template == "set_probes": - assert kwargs["probes"] == self._test_probes.copy() - return napalm_test_support.TEST_TERM_CONFIG.copy() - if template == "delete_probes": - assert kwargs["probes"] == self._test_delete_probes.copy() - return napalm_test_support.TEST_TERM_CONFIG.copy() - if template == "schedule_probes": - assert kwargs["probes"] == self._test_schedule_probes.copy() - return napalm_test_support.TEST_TERM_CONFIG.copy() - raise ValueError(f"incorrect template {template}") - - module_globals = { - "__salt__": { - "config.get": MagicMock( - return_value={"test": {"driver": "test", "key": "2orgk34kgk34g"}} - ), - "file.file_exists": napalm_test_support.true, - "file.join": napalm_test_support.join, - "file.get_managed": napalm_test_support.get_managed_file, - "random.hash": napalm_test_support.random_hash, - "net.load_template": mock_net_load, - } - } - - return {napalm_probes: module_globals} - - def test_probes_config(self): - ret = napalm_probes.config() - assert ret["out"] == napalm_test_support.TEST_PROBES_CONFIG.copy() - - def test_probes_results(self): - ret = napalm_probes.results() - assert ret["out"] == napalm_test_support.TEST_PROBES_RESULTS.copy() - - def test_set_probes(self): - ret = napalm_probes.set_probes(self._test_probes.copy()) - assert ret["result"] is True - - def test_delete_probes(self): - ret = napalm_probes.delete_probes(self._test_delete_probes.copy()) - assert ret["result"] is True - - def test_schedule_probes(self): - ret = napalm_probes.schedule_probes(self._test_schedule_probes.copy()) - assert ret["result"] is True