Merge pull request 'fix: get_config reads the backup as text, not JSON' (#12) from fix/get-config-xml into master

napalm_opnsense/opnsense.py

@@ -172,6 +172,22 @@ class OPNsenseDriver(OPNsensePingMixin, FirewallDriver):
         response.raise_for_status()
         return response.json()
 
+    def _get_text(self, path: str) -> str:
+        """Perform a GET request and return the body as text.
+
+        For the few endpoints that answer with a file rather than JSON, such
+        as the configuration backup.
+
+        :raises ConnectionClosedException: if called before :meth:`open`.
+        :raises RequestException: on HTTP-level errors.
+        """
+        if self.session is None:
+            raise ConnectionClosedException("Not connected – call open() first.")
+        url = self.base_url.rstrip("/") + path
+        response = self.session.get(url, timeout=self.timeout)
+        response.raise_for_status()
+        return response.text
+
     def _post(self, path: str, data: dict[str, Any] | None = None) -> dict[str, Any]:
         """Perform a POST request against the OPNsense REST API.
 
@@ -835,22 +851,19 @@ class OPNsenseDriver(OPNsensePingMixin, FirewallDriver):
         the JSON-serialised staged routes when a candidate has been loaded,
         or an empty string otherwise.
 
-        Calls ``GET /api/core/backup/download/this``.
+        Calls ``GET /api/core/backup/download/this``, which answers with
+        ``config.xml`` itself rather than JSON. A failed read raises: an empty
+        ``running`` would claim the device has no configuration and hide why
+        the read failed, e.g. an API key without the backup privilege.
         """
         configs: dict[str, str] = {"running": "", "startup": "", "candidate": ""}
 
         if retrieve in ("all", "running", "startup"):
-            try:
-                response = self._get("/api/core/backup/download/this")
-                xml_text: str = (
-                    response if isinstance(response, str) else str(response)
-                )
-                if retrieve in ("all", "running"):
-                    configs["running"] = xml_text
-                if retrieve in ("all", "startup"):
-                    configs["startup"] = xml_text
-            except Exception as exc:
-                logger.warning("Failed to fetch running/startup config: %s", exc)
+            xml_text = self._get_text("/api/core/backup/download/this")
+            if retrieve in ("all", "running"):
+                configs["running"] = xml_text
+            if retrieve in ("all", "startup"):
+                configs["startup"] = xml_text
 
         if retrieve in ("all", "candidate") and self._candidate_config is not None:
             configs["candidate"] = json.dumps(self._candidate_config, indent=2)

tests/unit/test_driver.py

@@ -2,6 +2,7 @@
 
 import json
 import pytest
+import requests
 from unittest.mock import MagicMock, patch
 
 from napalm_opnsense.opnsense import OPNsenseDriver
@@ -1656,24 +1657,74 @@ class TestRollback:
 
 class TestGetConfigCandidate:
     def test_candidate_empty_without_staged_config(self, driver):
-        driver._get = lambda path: "<opnsense/>"
+        driver._get_text = lambda path: "<opnsense/>"
         result = driver.get_config()
         assert result["candidate"] == ""
 
     def test_candidate_contains_staged_routes(self, driver):
-        driver._get = lambda path: "<opnsense/>"
+        driver._get_text = lambda path: "<opnsense/>"
         driver.load_merge_candidate(config=VALID_ROUTES_CONFIG)
         result = driver.get_config()
         assert "WAN_GW" in result["candidate"]
 
     def test_candidate_is_valid_json(self, driver):
-        driver._get = lambda path: "<opnsense/>"
+        driver._get_text = lambda path: "<opnsense/>"
         driver.load_merge_candidate(config=VALID_ROUTES_CONFIG)
         result = driver.get_config()
         parsed = json.loads(result["candidate"])
         assert isinstance(parsed, list)
 
 
+# ---------------------------------------------------------------------------
+# get_config() — running/startup
+#
+# /api/core/backup/download/this answers with config.xml itself, not JSON.
+# These go through the real session path: a stubbed _get hid for months that
+# the getter called .json() on XML and always returned an empty "running".
+# ---------------------------------------------------------------------------
+
+CONFIG_XML = '<?xml version="1.0"?>\n<opnsense><system><hostname>gw</hostname></system></opnsense>\n'
+
+
+def _make_xml_response(text):
+    mock_resp = MagicMock()
+    mock_resp.text = text
+    mock_resp.json.side_effect = json.JSONDecodeError("Expecting value", text, 0)
+    mock_resp.raise_for_status.return_value = None
+    return mock_resp
+
+
+class TestGetConfigRunning:
+    def test_running_is_the_xml_the_device_sent(self, driver):
+        driver.session.get.return_value = _make_xml_response(CONFIG_XML)
+        result = driver.get_config(retrieve="running")
+        assert result["running"] == CONFIG_XML
+
+    def test_startup_mirrors_running(self, driver):
+        driver.session.get.return_value = _make_xml_response(CONFIG_XML)
+        result = driver.get_config()
+        assert result["startup"] == result["running"] == CONFIG_XML
+
+    def test_reads_the_backup_download_endpoint(self, driver):
+        driver.session.get.return_value = _make_xml_response(CONFIG_XML)
+        driver.get_config(retrieve="running")
+        url = driver.session.get.call_args.args[0]
+        assert url.endswith("/api/core/backup/download/this")
+
+    def test_failed_read_raises_instead_of_returning_empty(self, driver):
+        # An empty "running" says "the device has no configuration"; the
+        # caller needs the reason, e.g. a key without backup privilege.
+        resp = MagicMock()
+        resp.raise_for_status.side_effect = requests.HTTPError("403 Forbidden")
+        driver.session.get.return_value = resp
+        with pytest.raises(requests.HTTPError, match="403"):
+            driver.get_config(retrieve="running")
+
+    def test_candidate_only_does_not_touch_the_device(self, driver):
+        driver.get_config(retrieve="candidate")
+        driver.session.get.assert_not_called()
+
+
 # ---------------------------------------------------------------------------
 # create_dhcp_reservation()
 #