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() #