Fix crash/mutation/log bugs in the shared napalm utils and proxy

changelog/69796.fixed.md

@@ -0,0 +1,5 @@
+Fixed three bugs in the shared NAPALM support code. ``salt.utils.napalm.get_device_opts``
+no longer crashes on ``optional_args: null`` and no longer mutates the caller's
+opts/pillar; ``force_reconnect`` no longer raises ``KeyError: 'proxy'`` on a
+straight (non-proxy) NAPALM minion; and the NAPALM proxy's shutdown error log no
+longer renders the port as a tuple.

salt/proxy/napalm.py

@@ -292,7 +292,7 @@ def shutdown(opts):
         port = (
             __context__["napalm_device"]["network_device"]
             .get("OPTIONAL_ARGS", {})
-            .get("port"),
+            .get("port")
         )
         log.error(
             "Cannot close connection with %s%s! Please check error: %s",

salt/utils/napalm.py

@@ -291,7 +291,13 @@ def get_device_opts(opts, salt_obj=None):
         or ""
     )
     network_device["TIMEOUT"] = device_dict.get("timeout", 60)
-    network_device["OPTIONAL_ARGS"] = device_dict.get("optional_args", {})
+    # ``or {}`` (not a ``get`` default) so an explicit ``optional_args: null`` in
+    # the config yields a dict rather than None; deepcopy so the config_lock /
+    # keepalive injected below are not written back into the caller's live
+    # opts / pillar structure.
+    network_device["OPTIONAL_ARGS"] = copy.deepcopy(
+        device_dict.get("optional_args") or {}
+    )
     network_device["ALWAYS_ALIVE"] = device_dict.get("always_alive", True)
     network_device["PROVIDER"] = device_dict.get("provider")
     network_device["UP"] = False
@@ -382,11 +388,16 @@ def func_wrapper(*args, **kwargs):
         force_reconnect = kwargs.get("force_reconnect", False)
         if force_reconnect:
             log.debug("Usage of reconnect force detected")
-            log.debug("Opts before merging")
-            log.debug(opts["proxy"])
-            opts["proxy"].update(**kwargs)
-            log.debug("Opts after merging")
-            log.debug(opts["proxy"])
+            # The credential override is merged into opts['proxy'] for the
+            # always-alive proxy path below. A straight minion has no 'proxy'
+            # key (this raised KeyError) and picks the override up from
+            # clean_kwargs further down, so only touch opts['proxy'] for a proxy.
+            if is_proxy(opts):
+                log.debug("Opts before merging")
+                log.debug(opts["proxy"])
+                opts["proxy"].update(**kwargs)
+                log.debug("Opts after merging")
+                log.debug(opts["proxy"])
         if is_proxy(opts) and always_alive:
             # if it is running in a NAPALM Proxy and it's using the default
             # always alive behaviour, will get the cached copy of the network

tests/pytests/unit/proxy/test_napalm.py

@@ -311,3 +311,30 @@ def test_call(test_opts):
     with patch.dict(napalm_proxy.__context__, mock_context):
         ret = napalm_proxy.call("get_arp_table")
         assert ret == {"result": False, "comment": "Not initialised yet", "out": None}
+
+
+def _shutdown_port_log(optional_args):
+    """Run shutdown() with a failing close() and return the port fragment that
+    was passed to the error log (index 2 of the log.error args)."""
+    driver = MagicMock()
+    driver.close.side_effect = Exception("boom")
+    network_device = {
+        "DRIVER": driver,
+        "UP": True,
+        "HOSTNAME": "core05.nrt02",
+        "OPTIONAL_ARGS": optional_args,
+    }
+    mock_context = {"napalm_device": {"network_device": network_device}}
+    with patch.dict(napalm_proxy.__context__, mock_context), patch(
+        "salt.proxy.napalm.log"
+    ) as mock_log:
+        ret = napalm_proxy.shutdown({})
+    assert ret is True
+    return mock_log.error.call_args[0][2]
+
+
+def test_shutdown_logs_scalar_port_on_close_failure():
+    # The trailing comma made ``port`` a 1-tuple, so the error log rendered
+    # ``:(830,)`` / ``:(None,)`` instead of ``:830`` / ``""``.
+    assert _shutdown_port_log({"port": 830}) == ":830"
+    assert _shutdown_port_log({}) == ""

tests/pytests/unit/utils/test_napalm.py

@@ -0,0 +1,57 @@
+"""
+Unit tests for salt.utils.napalm.
+"""
+
+import types
+
+import salt.utils.napalm as napalm_utils
+from tests.support.mock import MagicMock, patch
+
+
+def test_get_device_opts_null_optional_args():
+    # ``optional_args: null`` in the config yields None from ``.get(..., {})``
+    # (the default only applies to a missing key), which then crashed the
+    # ``"config_lock" not in ...`` membership test.
+    opts = {"napalm": {"driver": "junos", "optional_args": None}}
+    device = napalm_utils.get_device_opts(opts)
+    assert isinstance(device["OPTIONAL_ARGS"], dict)
+    assert device["OPTIONAL_ARGS"]["config_lock"] is False
+
+
+def test_get_device_opts_does_not_mutate_caller():
+    # The injected config_lock / keepalive must land in a copy, not in the
+    # caller's live opts / pillar ``optional_args`` dict.
+    optional_args = {"port": 830}
+    opts = {"napalm": {"driver": "junos", "optional_args": optional_args}}
+    napalm_utils.get_device_opts(opts)
+    assert optional_args == {"port": 830}
+
+
+def _wrapped_with_opts(opts):
+    """A ``proxy_napalm_wrap``-decorated function whose module globals carry the
+    given ``__opts__`` (so we can drive the wrapper without a real minion)."""
+
+    def _fn(*args, **kwargs):
+        return "ok"
+
+    func_globals = {
+        "__opts__": opts,
+        "__proxy__": {},
+        "__salt__": {"config.get": MagicMock(return_value={"driver": "junos"})},
+    }
+    fn = types.FunctionType(_fn.__code__, func_globals, "_fn")
+    return napalm_utils.proxy_napalm_wrap(fn)
+
+
+def test_proxy_napalm_wrap_force_reconnect_straight_minion():
+    # force_reconnect on a straight (non-proxy) minion has no ``opts['proxy']``;
+    # the wrapper must not blindly do ``opts['proxy'].update(...)`` (KeyError).
+    # The override reaches the device through clean_kwargs on the straight path.
+    opts = {"napalm": {"driver": "junos"}}
+    wrapped = _wrapped_with_opts(opts)
+    get_device = MagicMock(return_value={"DRIVER": MagicMock()})
+    with patch("salt.utils.napalm.get_device", get_device):
+        result = wrapped(force_reconnect=True)
+    assert result == "ok"
+    get_device.assert_called_once()
+    assert get_device.call_args[0][0]["napalm"].get("force_reconnect") is True