diff --git a/napalm_opnsense/opnsense.py b/napalm_opnsense/opnsense.py
index d1c70ff..9ef9e04 100644
--- a/napalm_opnsense/opnsense.py
+++ b/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)
diff --git a/tests/unit/test_driver.py b/tests/unit/test_driver.py
index 1729961..506d009 100644
--- a/tests/unit/test_driver.py
+++ b/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: ""
+ driver._get_text = lambda path: ""
result = driver.get_config()
assert result["candidate"] == ""
def test_candidate_contains_staged_routes(self, driver):
- driver._get = lambda path: ""
+ driver._get_text = lambda path: ""
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: ""
+ driver._get_text = lambda path: ""
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 = '\ngw\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()
#