From 6cff40aa3b98ecdc2382f3cee051bd249467c94b Mon Sep 17 00:00:00 2001 From: Tyler Woods Date: Mon, 15 Jun 2026 17:16:33 +0000 Subject: [PATCH 1/2] Add firewall rule helpers and API error details --- README.md | 35 +++++ tests/test_api_error_details.py | 101 ++++++++++++ tests/test_firewall_rules.py | 154 +++++++++++++++++++ tests/test_package_smoke.py | 9 +- unifi_controller_api/__init__.py | 4 +- unifi_controller_api/api_client.py | 127 ++++++++++++++- unifi_controller_api/exceptions.py | 90 ++++++++++- unifi_controller_api/models/__init__.py | 2 + unifi_controller_api/models/firewall_rule.py | 62 ++++++++ 9 files changed, 577 insertions(+), 7 deletions(-) create mode 100644 tests/test_api_error_details.py create mode 100644 tests/test_firewall_rules.py create mode 100644 unifi_controller_api/models/firewall_rule.py diff --git a/README.md b/README.md index ce91a99..7d21b7c 100644 --- a/README.md +++ b/README.md @@ -67,6 +67,7 @@ except Exception as e: # events = controller.get_unifi_site_event(site_name) # rogue_aps = controller.get_unifi_site_rogueap(site_name) # networks = controller.get_unifi_site_networkconf(site_name) +# firewall_rules = controller.get_unifi_site_firewallrule(site_name) # report = controller.devices_report(site_names=['site1', 'site2']) # Exporting data: @@ -93,6 +94,39 @@ except Exception as e: --- +## Firewall Rules + +Firewall rule support follows the same raw/typed pattern as the rest of the client: + +```python +rules = controller.get_unifi_site_firewallrule("default", raw=False) +for rule in rules: + print(rule.name, rule.ruleset, rule.action, rule.protocol, rule.dst_port) +``` + +Mutating helpers are available for callers that deliberately want to manage rules: + +```python +controller.create_unifi_site_firewallrule( + "default", + { + "name": "Block example traffic", + "enabled": True, + "action": "drop", + "ruleset": "WAN_OUT", + "protocol": "tcp", + "dst_port": "25", + }, +) +``` + +> **Caution:** Firewall rule endpoints are private UniFi Network APIs. Valid fields, +> rule index conventions, and controller-side validation can vary across UniFi +> Network versions. Prefer read-only listing first, keep writes explicit, and inspect +> `UnifiAPIError.response_json` when the controller rejects a payload. + +--- + ## Data Models The library automatically maps JSON API responses to Python data classes located in `unifi_controller_api.models`. Key models include: @@ -103,6 +137,7 @@ The library automatically maps JSON API responses to Python data classes located * `UnifiClient`: Represents a connected client (wired or wireless). * `UnifiWlanConf`: Represents a Wireless LAN configuration. * `UnifiNetworkConf`: Represents a Network configuration. +* `UnifiFirewallRule`: Represents a firewall rule. * `UnifiAlarm`: Represents a controller alarm. * `UnifiEvent`: Represents a controller event. * `UnifiRogueAp`: Represents a detected rogue access point. diff --git a/tests/test_api_error_details.py b/tests/test_api_error_details.py new file mode 100644 index 0000000..0c97289 --- /dev/null +++ b/tests/test_api_error_details.py @@ -0,0 +1,101 @@ +import requests +import pytest + +from unifi_controller_api import UnifiController +from unifi_controller_api.exceptions import UnifiAPIError + + +class ErrorResponse: + status_code = 400 + text = '{"meta":{"rc":"error","msg":"api.err.InvalidValue"},"data":[]}' + + def json(self): + return {"meta": {"rc": "error", "msg": "api.err.InvalidValue"}, "data": []} + + def raise_for_status(self): + raise requests.HTTPError("400 Client Error", response=self) # type: ignore[arg-type] + + +class NonJsonErrorResponse: + status_code = 502 + text = "bad gateway" + + def json(self): + raise ValueError("not json") + + def raise_for_status(self): + raise requests.HTTPError("502 Server Error", response=self) # type: ignore[arg-type] + + +class NoResponseSession: + cookies = {} + + def request(self, method, url, **kwargs): + raise requests.ConnectionError("connection failed") + + +class ErrorSession: + cookies = {} + + def __init__(self, response): + self.response = response + + def request(self, method, url, **kwargs): + return self.response + + +def make_controller(session): + controller = UnifiController.__new__(UnifiController) + controller.controller_url = "https://controller.example" + controller.original_controller_url = "https://controller.example" + controller.is_udm_pro = False + controller.session = session + controller.verify_ssl = True + controller.auth_retry_enabled = False + controller.auth_retry_count = 1 + controller.auth_retry_delay = 0.1 # type: ignore[assignment] + controller.request_timeout = None + return controller + + +def test_unifi_api_error_preserves_json_response_details(): + controller = make_controller(ErrorSession(ErrorResponse())) + + with pytest.raises(UnifiAPIError) as excinfo: + controller._invoke_api_call("POST", "https://controller.example/api/test", {}) + + err = excinfo.value + assert err.method == "POST" + assert err.url == "https://controller.example/api/test" + assert err.status_code == 400 + assert err.response_json is not None + assert err.response_json["meta"]["msg"] == "api.err.InvalidValue" + assert "api.err.InvalidValue" in str(err) + + +def test_unifi_api_error_preserves_non_json_response_text(): + controller = make_controller(ErrorSession(NonJsonErrorResponse())) + + with pytest.raises(UnifiAPIError) as excinfo: + controller._invoke_api_call("PUT", "https://controller.example/api/test", {}) + + err = excinfo.value + assert err.method == "PUT" + assert err.status_code == 502 + assert err.response_text == "bad gateway" + assert err.response_json is None + assert "bad gateway" in str(err) + + +def test_unifi_api_error_handles_request_failure_without_response(): + controller = make_controller(NoResponseSession()) + + with pytest.raises(UnifiAPIError) as excinfo: + controller._invoke_api_call("DELETE", "https://controller.example/api/test") + + err = excinfo.value + assert err.method == "DELETE" + assert err.url == "https://controller.example/api/test" + assert err.status_code is None + assert err.response_json is None + assert "connection failed" in str(err) diff --git a/tests/test_firewall_rules.py b/tests/test_firewall_rules.py new file mode 100644 index 0000000..fefa9f2 --- /dev/null +++ b/tests/test_firewall_rules.py @@ -0,0 +1,154 @@ +from unifi_controller_api import UnifiController, UnifiFirewallRule + + +class FakeResponse: + status_code = 200 + text = "" + + def __init__(self, payload): + self._payload = payload + + def json(self): + return self._payload + + def raise_for_status(self): + return None + + +class FirewallSession: + cookies = {} + + def __init__(self): + self.get_calls = [] + self.request_calls = [] + self.get_payload = { + "meta": {"rc": "ok"}, + "data": [ + { + "_id": "rule-1", + "name": "Allow DNS", + "enabled": True, + "action": "accept", + "ruleset": "LAN_OUT", + "rule_index": 20000, + "protocol": "tcp_udp", + "dst_port": "53", + "undocumented_field": "preserved", + } + ], + } + + def get(self, url, **kwargs): + self.get_calls.append((url, kwargs)) + return FakeResponse(self.get_payload) + + def request(self, method, url, **kwargs): + self.request_calls.append((method, url, kwargs)) + payload = kwargs.get("json") or {} + return FakeResponse( + { + "meta": {"rc": "ok"}, + "data": [{"_id": "rule-2", **payload}], + } + ) + + +def make_controller(session=None): + controller = UnifiController.__new__(UnifiController) + controller.controller_url = "https://controller.example/proxy/network" + controller.original_controller_url = "https://controller.example" + controller.is_udm_pro = False + controller.session = session or FirewallSession() # type: ignore[assignment] + controller.verify_ssl = True + controller.auth_retry_enabled = False + controller.auth_retry_count = 1 + controller.auth_retry_delay = 0.1 # type: ignore[assignment] + controller.request_timeout = None + return controller + + +def test_get_firewall_rules_returns_raw_data_by_default(): + session = FirewallSession() + controller = make_controller(session) + + rules = controller.get_unifi_site_firewallrule("default") + + assert isinstance(rules[0], dict) + assert rules[0]["name"] == "Allow DNS" + assert session.get_calls[0][0] == ( + "https://controller.example/proxy/network/api/s/default/rest/firewallrule" + ) + + +def test_get_firewall_rules_maps_typed_model_and_preserves_extra_fields(): + controller = make_controller(FirewallSession()) + + rules = controller.get_unifi_site_firewallrule("default", raw=False) + + assert isinstance(rules[0], UnifiFirewallRule) + assert rules[0].name == "Allow DNS" + assert rules[0].dst_port == "53" + assert rules[0]._extra_fields["undocumented_field"] == "preserved" + + +def test_get_firewall_rule_by_id_uses_rule_specific_endpoint(): + session = FirewallSession() + controller = make_controller(session) + + controller.get_unifi_site_firewallrule("default", firewall_rule_id="rule-1") + + assert session.get_calls[0][0].endswith( + "/api/s/default/rest/firewallrule/rule-1" + ) + + +def test_create_firewall_rule_posts_payload_and_maps_response(): + session = FirewallSession() + controller = make_controller(session) + + rules = controller.create_unifi_site_firewallrule( + "default", + {"name": "Block SMTP", "action": "drop"}, + raw=False, + protocol="tcp", + dst_port="25", + ) + + method, url, kwargs = session.request_calls[0] + assert method == "POST" + assert url.endswith("/api/s/default/rest/firewallrule") + assert kwargs["json"] == { + "name": "Block SMTP", + "action": "drop", + "protocol": "tcp", + "dst_port": "25", + } + assert isinstance(rules[0], UnifiFirewallRule) + assert rules[0]._id == "rule-2" + assert rules[0].dst_port == "25" + + +def test_update_firewall_rule_puts_payload_to_rule_endpoint(): + session = FirewallSession() + controller = make_controller(session) + + controller.update_unifi_site_firewallrule( + "default", "rule-1", {"enabled": False} + ) + + method, url, kwargs = session.request_calls[0] + assert method == "PUT" + assert url.endswith("/api/s/default/rest/firewallrule/rule-1") + assert kwargs["json"] == {"enabled": False} + + +def test_delete_firewall_rule_uses_delete_method(): + session = FirewallSession() + controller = make_controller(session) + + controller.delete_unifi_site_firewallrule("default", "rule-1") + + method, url, kwargs = session.request_calls[0] + assert method == "DELETE" + assert url.endswith("/api/s/default/rest/firewallrule/rule-1") + assert "json" not in kwargs diff --git a/tests/test_package_smoke.py b/tests/test_package_smoke.py index 1d7dcb1..474ea5e 100644 --- a/tests/test_package_smoke.py +++ b/tests/test_package_smoke.py @@ -3,7 +3,13 @@ from importlib import resources import unifi_controller_api as api -from unifi_controller_api.models import UnifiDevice, UnifiHealth, UnifiPortConf, UnifiSite +from unifi_controller_api.models import ( + UnifiDevice, + UnifiFirewallRule, + UnifiHealth, + UnifiPortConf, + UnifiSite, +) def test_public_imports_are_available(): @@ -12,6 +18,7 @@ def test_public_imports_are_available(): assert UnifiDevice is not None assert UnifiHealth is not None assert UnifiPortConf is not None + assert api.UnifiFirewallRule is UnifiFirewallRule def test_packaged_device_model_database_is_valid_json(): diff --git a/unifi_controller_api/__init__.py b/unifi_controller_api/__init__.py index 128f9b1..948c45f 100644 --- a/unifi_controller_api/__init__.py +++ b/unifi_controller_api/__init__.py @@ -8,7 +8,8 @@ from .api_client import UnifiController from .models import ( UnifiSite, UnifiDevice, LLDPEntry, UnifiClient, - UnifiEvent, UnifiAlarm, UnifiWlanConf, UnifiRogueAp, UnifiNetworkConf + UnifiEvent, UnifiAlarm, UnifiWlanConf, UnifiRogueAp, UnifiNetworkConf, + UnifiFirewallRule ) from .export import export_csv, export_json, to_dict_list from .exceptions import ( @@ -30,6 +31,7 @@ "UnifiWlanConf", "UnifiRogueAp", "UnifiNetworkConf", + "UnifiFirewallRule", "export_csv", "export_json", "to_dict_list", diff --git a/unifi_controller_api/api_client.py b/unifi_controller_api/api_client.py index a6f63a8..dbd9c52 100644 --- a/unifi_controller_api/api_client.py +++ b/unifi_controller_api/api_client.py @@ -17,6 +17,7 @@ from .models.wlanconf import UnifiWlanConf from .models.rogueap import UnifiRogueAp from .models.networkconf import UnifiNetworkConf +from .models.firewall_rule import UnifiFirewallRule from .models.health import UnifiHealth, UnifiSubsystemHealth from .models.portconf import UnifiPortConf from .logging import get_logger @@ -337,7 +338,12 @@ def _invoke_api_call( if not isinstance(e, UnifiAuthenticationError): error_msg = f"API {method} request to {url} failed: {str(e)}" logger.error(error_msg) - raise UnifiAPIError(error_msg) from e + raise UnifiAPIError( + error_msg, + method=method, + url=url, + response=getattr(e, "response", None), + ) from e else: raise @@ -458,7 +464,12 @@ def invoke_get_rest_api_call(self, url, headers=None, timeout=None): if not isinstance(e, UnifiAuthenticationError): error_msg = f"API GET request to {url} failed: {str(e)}" logger.error(error_msg) - raise UnifiAPIError(error_msg) from e + raise UnifiAPIError( + error_msg, + method="GET", + url=url, + response=getattr(e, "response", None), + ) from e raise def _process_api_response( @@ -487,7 +498,13 @@ def _process_api_response( if meta.get("rc") == "error": error_msg = meta.get("msg") or f"API request to {uri} failed" logger.warning(f"UniFi API error for {uri}: {error_msg}") - raise UnifiAPIError(error_msg) + raise UnifiAPIError( + error_msg, + method="GET", + url=uri, + response=response, + response_json=raw_data, + ) raw_results = raw_data.get("data", []) if "data" not in raw_data: @@ -1081,6 +1098,110 @@ def get_unifi_site_networkconf( logger.debug("Returning raw network configuration data.") return raw_results + def _map_firewall_rules( + self, raw_results: List[Dict[str, Any]], raw: bool + ) -> Union[List[Dict[str, Any]], List[UnifiFirewallRule]]: + if raw: + logger.debug("Returning raw firewall rule data.") + return raw_results + + firewall_rules = [] + for rule_data in raw_results: + try: + model_fields, extra_fields = map_api_data_to_model( + rule_data, UnifiFirewallRule + ) + rule = UnifiFirewallRule(**model_fields) + rule._extra_fields = extra_fields + firewall_rules.append(rule) + except Exception as e: + logger.error( + f"Error creating UnifiFirewallRule model from data: {rule_data}. Error: {e}" + ) + logger.debug( + f"Returning {len(firewall_rules)} mapped UnifiFirewallRule objects." + ) + return firewall_rules + + def get_unifi_site_firewallrule( + self, site_name: str, firewall_rule_id: Optional[str] = None, raw: bool = True + ) -> Union[List[Dict[str, Any]], List[UnifiFirewallRule]]: + """Get firewall rules for a UniFi site. + + Retrieves firewall rules from ``/api/s/{site_name}/rest/firewallrule``. + If ``firewall_rule_id`` is provided, fetches a single rule by ID. This + method uses UniFi's private API; returned fields and valid values can + vary by controller version. + """ + uri_suffix = f"/{firewall_rule_id}" if firewall_rule_id else "" + uri = f"{self.controller_url}/api/s/{site_name}/rest/firewallrule{uri_suffix}" + logger.info( + f"Fetching firewall rule(s) for site '{site_name}' from {uri}" + ) + response = self.invoke_get_rest_api_call(uri) + raw_results = self._process_api_response(response, uri) + return self._map_firewall_rules(raw_results, raw) + + def create_unifi_site_firewallrule( + self, + site_name: str, + rule_data: Optional[Dict[str, Any]] = None, + raw: bool = True, + **rule_fields: Any, + ) -> Union[List[Dict[str, Any]], List[UnifiFirewallRule]]: + """Create a firewall rule for a UniFi site. + + This is a mutating operation. Pass a dictionary of UniFi firewall rule + fields via ``rule_data`` and/or keyword fields. Keyword fields override + duplicate keys from ``rule_data``. + """ + payload = dict(rule_data or {}) + payload.update(rule_fields) + uri = f"{self.controller_url}/api/s/{site_name}/rest/firewallrule" + logger.info(f"Creating firewall rule for site '{site_name}' at {uri}") + response = self._invoke_api_call("POST", uri, json_payload=payload) + raw_results = self._process_api_response(response, uri) + return self._map_firewall_rules(raw_results, raw) + + def update_unifi_site_firewallrule( + self, + site_name: str, + firewall_rule_id: str, + rule_data: Optional[Dict[str, Any]] = None, + raw: bool = True, + **rule_fields: Any, + ) -> Union[List[Dict[str, Any]], List[UnifiFirewallRule]]: + """Update a firewall rule for a UniFi site. + + This is a mutating operation. Pass the rule ID and a dictionary of + fields via ``rule_data`` and/or keyword fields. Keyword fields override + duplicate keys from ``rule_data``. + """ + payload = dict(rule_data or {}) + payload.update(rule_fields) + uri = f"{self.controller_url}/api/s/{site_name}/rest/firewallrule/{firewall_rule_id}" + logger.info( + f"Updating firewall rule '{firewall_rule_id}' for site '{site_name}' at {uri}" + ) + response = self._invoke_api_call("PUT", uri, json_payload=payload) + raw_results = self._process_api_response(response, uri) + return self._map_firewall_rules(raw_results, raw) + + def delete_unifi_site_firewallrule( + self, site_name: str, firewall_rule_id: str + ) -> List[Dict[str, Any]]: + """Delete a firewall rule for a UniFi site. + + This is a mutating operation. Returns the controller's raw ``data`` + payload for consistency with other low-level operations. + """ + uri = f"{self.controller_url}/api/s/{site_name}/rest/firewallrule/{firewall_rule_id}" + logger.info( + f"Deleting firewall rule '{firewall_rule_id}' for site '{site_name}' at {uri}" + ) + response = self._invoke_api_call("DELETE", uri) + return self._process_api_response(response, uri) + def normalize_mac(self, mac_address): """ Normalize MAC address to colon-separated format. diff --git a/unifi_controller_api/exceptions.py b/unifi_controller_api/exceptions.py index 095be6e..5e9cf06 100644 --- a/unifi_controller_api/exceptions.py +++ b/unifi_controller_api/exceptions.py @@ -1,3 +1,9 @@ +from __future__ import annotations + +import json +from typing import Any, Optional + + class UnifiControllerError(Exception): """Base exception for UnifiController errors.""" @@ -11,9 +17,89 @@ class UnifiAuthenticationError(UnifiControllerError): class UnifiAPIError(UnifiControllerError): - """Raised when an API call to the UniFi Controller fails.""" + """Raised when an API call to the UniFi Controller fails. - pass + The UniFi private API often returns useful structured JSON for failures + (for example ``meta.msg`` and validation details). Preserve those details + so callers do not need to bypass the client to diagnose controller errors. + """ + + def __init__( + self, + message: str, + *, + method: Optional[str] = None, + url: Optional[str] = None, + status_code: Optional[int] = None, + response_text: Optional[str] = None, + response_json: Optional[Any] = None, + response: Optional[Any] = None, + ) -> None: + self.method = method + self.url = url + self.status_code = status_code + self.response_text = response_text + self.response_json = response_json + + if response is not None: + self.status_code = status_code if status_code is not None else getattr( + response, "status_code", None + ) + if self.response_text is None: + self.response_text = getattr(response, "text", None) + if self.response_json is None: + try: + self.response_json = response.json() + except (TypeError, ValueError, AttributeError): + self.response_json = None + + detail = self._extract_controller_message() + parts = [message] + context = self._format_context() + if context: + parts.append(context) + if detail and detail not in message: + parts.append(f"controller_message={detail}") + elif self.response_text and self.response_json is None: + text = self.response_text.strip() + if text and text not in message: + parts.append(f"response_text={text[:500]}") + + super().__init__("; ".join(parts)) + + def _format_context(self) -> str: + values = [] + if self.method: + values.append(f"method={self.method.upper()}") + if self.url: + values.append(f"url={self.url}") + if self.status_code is not None: + values.append(f"status_code={self.status_code}") + return ", ".join(values) + + def _extract_controller_message(self) -> Optional[str]: + if not isinstance(self.response_json, dict): + return None + + meta = self.response_json.get("meta") + if isinstance(meta, dict): + msg = meta.get("msg") + if msg: + return str(msg) + + # Some UniFi validation failures put details in data entries. Keep this + # compact so exception strings remain readable while structured data is + # still available on ``response_json`` for callers that need more. + data = self.response_json.get("data") + if isinstance(data, list) and data: + first = data[0] + if isinstance(first, dict): + msg = first.get("msg") or first.get("validationError") + if msg: + if isinstance(msg, (dict, list)): + return json.dumps(msg, sort_keys=True) + return str(msg) + return None class UnifiDataError(UnifiControllerError): diff --git a/unifi_controller_api/models/__init__.py b/unifi_controller_api/models/__init__.py index 34a41e5..54f6585 100644 --- a/unifi_controller_api/models/__init__.py +++ b/unifi_controller_api/models/__init__.py @@ -40,6 +40,7 @@ from .wlanconf import UnifiWlanConf from .rogueap import UnifiRogueAp from .networkconf import UnifiNetworkConf +from .firewall_rule import UnifiFirewallRule from .health import UnifiHealth, UnifiSubsystemHealth from .portconf import UnifiPortConf @@ -53,6 +54,7 @@ "UnifiWlanConf", "UnifiRogueAp", "UnifiNetworkConf", + "UnifiFirewallRule", "UnifiHealth", "UnifiSubsystemHealth", "UnifiPortConf" diff --git a/unifi_controller_api/models/firewall_rule.py b/unifi_controller_api/models/firewall_rule.py new file mode 100644 index 0000000..3b5d3b2 --- /dev/null +++ b/unifi_controller_api/models/firewall_rule.py @@ -0,0 +1,62 @@ +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional, Union + + +@dataclass +class UnifiFirewallRule: + """Represents a UniFi firewall rule. + + UniFi Network firewall rule payloads vary by controller version. Common + fields are modeled directly and any unrecognized fields are preserved in + ``_extra_fields`` when mapping API responses with ``raw=False``. + """ + + _id: Optional[str] = None + name: Optional[str] = None + site_id: Optional[str] = None + enabled: Optional[bool] = None + action: Optional[str] = None + ruleset: Optional[str] = None + rule_index: Optional[Union[int, str]] = None + protocol: Optional[str] = None + protocol_v6: Optional[str] = None + dst_port: Optional[str] = None + src_port: Optional[str] = None + dst_address: Optional[str] = None + src_address: Optional[str] = None + dst_address_ipv6: Optional[str] = None + src_address_ipv6: Optional[str] = None + dst_networkconf_id: Optional[str] = None + src_networkconf_id: Optional[str] = None + dst_networkconf_type: Optional[str] = None + src_networkconf_type: Optional[str] = None + dst_firewallgroup_ids: Optional[List[str]] = None + src_firewallgroup_ids: Optional[List[str]] = None + src_mac_address: Optional[str] = None + logging: Optional[bool] = None + state_established: Optional[bool] = None + state_invalid: Optional[bool] = None + state_new: Optional[bool] = None + state_related: Optional[bool] = None + ipsec: Optional[str] = None + icmp_typename: Optional[str] = None + icmpv6_typename: Optional[str] = None + setting_preference: Optional[str] = None + attr_hidden: Optional[bool] = None + attr_hidden_id: Optional[str] = None + attr_no_delete: Optional[bool] = None + attr_no_edit: Optional[bool] = None + + _extra_fields: Dict[str, Any] = field(default_factory=dict, repr=False) + + def to_dict(self) -> Dict[str, Any]: + """Convert the dataclass instance to a dictionary for API payloads.""" + data = { + k: v + for k, v in self.__dict__.items() + if not k.startswith("_") and v is not None + } + if self._id is not None: + data["_id"] = self._id + data.update(self._extra_fields) + return data From 360ff3d567692c7c4a1b806ec0d5b683061b4c3c Mon Sep 17 00:00:00 2001 From: Tyler Woods Date: Mon, 15 Jun 2026 17:24:58 +0000 Subject: [PATCH 2/2] Preserve method on controller error payloads --- tests/test_api_response_handling.py | 13 +++++++++++++ unifi_controller_api/api_client.py | 11 ++++++----- 2 files changed, 19 insertions(+), 5 deletions(-) diff --git a/tests/test_api_response_handling.py b/tests/test_api_response_handling.py index cad1405..319fa08 100644 --- a/tests/test_api_response_handling.py +++ b/tests/test_api_response_handling.py @@ -33,3 +33,16 @@ def test_process_api_response_rejects_missing_data_key(): def test_process_api_response_raises_api_error_for_unifi_error_payload(): with pytest.raises(UnifiAPIError, match="api.err.Invalid"): process({"meta": {"rc": "error", "msg": "api.err.Invalid"}, "data": []}) + + +def test_process_api_response_preserves_request_method_for_error_payload(): + controller = UnifiController.__new__(UnifiController) + response = cast( + Any, + FakeResponse({"meta": {"rc": "error", "msg": "api.err.Invalid"}, "data": []}), + ) + + with pytest.raises(UnifiAPIError) as excinfo: + controller._process_api_response(response, "/api/test", method="POST") + + assert excinfo.value.method == "POST" diff --git a/unifi_controller_api/api_client.py b/unifi_controller_api/api_client.py index dbd9c52..750f48d 100644 --- a/unifi_controller_api/api_client.py +++ b/unifi_controller_api/api_client.py @@ -473,7 +473,7 @@ def invoke_get_rest_api_call(self, url, headers=None, timeout=None): raise def _process_api_response( - self, response: Optional[requests.Response], uri: str + self, response: Optional[requests.Response], uri: str, method: str = "GET" ) -> List[Dict[str, Any]]: """ Process API response and handle common error cases. @@ -481,6 +481,7 @@ def _process_api_response( Args: response: Response from API call uri: URI that was called + method: HTTP method used for the request, for error context Returns: List of data items from the response @@ -500,7 +501,7 @@ def _process_api_response( logger.warning(f"UniFi API error for {uri}: {error_msg}") raise UnifiAPIError( error_msg, - method="GET", + method=method, url=uri, response=response, response_json=raw_data, @@ -1160,7 +1161,7 @@ def create_unifi_site_firewallrule( uri = f"{self.controller_url}/api/s/{site_name}/rest/firewallrule" logger.info(f"Creating firewall rule for site '{site_name}' at {uri}") response = self._invoke_api_call("POST", uri, json_payload=payload) - raw_results = self._process_api_response(response, uri) + raw_results = self._process_api_response(response, uri, method="POST") return self._map_firewall_rules(raw_results, raw) def update_unifi_site_firewallrule( @@ -1184,7 +1185,7 @@ def update_unifi_site_firewallrule( f"Updating firewall rule '{firewall_rule_id}' for site '{site_name}' at {uri}" ) response = self._invoke_api_call("PUT", uri, json_payload=payload) - raw_results = self._process_api_response(response, uri) + raw_results = self._process_api_response(response, uri, method="PUT") return self._map_firewall_rules(raw_results, raw) def delete_unifi_site_firewallrule( @@ -1200,7 +1201,7 @@ def delete_unifi_site_firewallrule( f"Deleting firewall rule '{firewall_rule_id}' for site '{site_name}' at {uri}" ) response = self._invoke_api_call("DELETE", uri) - return self._process_api_response(response, uri) + return self._process_api_response(response, uri, method="DELETE") def normalize_mac(self, mac_address): """