Add REST API support with backward-compatible auto detection#454
Add REST API support with backward-compatible auto detection#454espipj wants to merge 13 commits intoiMicknl:mainfrom
Conversation
|
Apologies for the delay, @espipj! Happy to take a better look at this soon, but I might need to first fix our CI/CD... |
48dbdfd to
833f12b
Compare
There was a problem hiding this comment.
Pull request overview
Adds support for newer Sagemcom firmwares exposing /api/v1/* REST endpoints while maintaining compatibility with the legacy /cgi/json-req (XPath) API, including an AUTO mode that attempts to pick the right API at runtime.
Changes:
- Introduced
ApiMode(AUTO/LEGACY/REST) and extendedSagemcomClientto select/track the active API mode. - Implemented REST flows for
login(),logout(),get_device_info(),get_hosts(), andreboot(), while guarding XPath methods as legacy-only. - Added/updated unit tests and README documentation for the new API modes and REST parsing.
Reviewed changes
Copilot reviewed 4 out of 4 changed files in this pull request and generated 5 comments.
| File | Description |
|---|---|
sagemcom_api/client.py |
Adds REST request/parse paths, API mode selection, and legacy-only guards. |
sagemcom_api/enums.py |
Adds ApiMode and EncryptionMethod.NONE. |
tests/unit/test_client_basic.py |
Adds tests for AUTO fallback to REST, REST host parsing, and REST reboot. |
README.md |
Documents API mode behavior and updates quickstart usage. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| payload = urllib.parse.urlencode(data or {}) | ||
| request_headers = {"Content-Type": "application/x-www-form-urlencoded"} | ||
|
|
||
| async with self.session.request(method, url, data=payload, headers=request_headers) as response: |
There was a problem hiding this comment.
__rest_request() always sends a form-encoded body via data=payload, even for GET requests (e.g., /api/v1/device, /api/v1/home, /api/v1/hosts). Some servers reject or ignore GET bodies; this can break REST mode. Consider using params=data (and omit the Content-Type header) for GET/HEAD, and only send form-encoded data for methods that actually support a request body (POST/PUT/PATCH).
| payload = urllib.parse.urlencode(data or {}) | |
| request_headers = {"Content-Type": "application/x-www-form-urlencoded"} | |
| async with self.session.request(method, url, data=payload, headers=request_headers) as response: | |
| request_method = method.upper() | |
| request_kwargs: dict[str, Any] = {} | |
| if request_method in {"GET", "HEAD"}: | |
| if data: | |
| request_kwargs["params"] = data | |
| elif request_method in {"POST", "PUT", "PATCH"}: | |
| request_kwargs["data"] = urllib.parse.urlencode(data or {}) | |
| request_kwargs["headers"] = { | |
| "Content-Type": "application/x-www-form-urlencoded" | |
| } | |
| elif data: | |
| request_kwargs["data"] = urllib.parse.urlencode(data) | |
| request_kwargs["headers"] = { | |
| "Content-Type": "application/x-www-form-urlencoded" | |
| } | |
| async with self.session.request(request_method, url, **request_kwargs) as response: |
| result = await response.text() | ||
| if response.status in (401, 403): | ||
| raise UnauthorizedException(result) | ||
|
|
||
| if response.status == 404: | ||
| raise UnsupportedHostException(result) | ||
|
|
||
| if response.status == 400: | ||
| raise AuthenticationException(result) | ||
|
|
There was a problem hiding this comment.
__rest_request() maps HTTP 400 to AuthenticationException, but 400 is generally a bad request (and your own tests show /api/v1/home can return 400 for non-auth reasons). This can mislead callers and makes error handling inconsistent with the legacy path. Consider raising BadRequestException (or a REST-specific exception) for 400, and reserve AuthenticationException/UnauthorizedException for 401/403-like cases.
| async def get_encryption_method(self) -> EncryptionMethod: | ||
| """Determine which encryption method to use for authentication and set it directly.""" | ||
| if self.api_mode == ApiMode.REST: | ||
| return EncryptionMethod.NONE | ||
|
|
||
| if self.api_mode == ApiMode.AUTO and await self.__probe_rest_availability(): | ||
| return EncryptionMethod.NONE | ||
|
|
There was a problem hiding this comment.
In AUTO mode, get_encryption_method() returns EncryptionMethod.NONE as soon as __probe_rest_availability() succeeds. That probe only proves REST login works, not that the legacy JSON-REQ API is unavailable; on dual-stack firmware this would incorrectly select NONE, and EncryptionMethod.NONE is not compatible with legacy hashing (it results in sending the raw password). Consider removing this early return, or only returning NONE when legacy probing fails with the same conditions used by __should_fallback_to_rest() (i.e., legacy endpoint appears unavailable).
| device = data[0].get("device", {}) | ||
| return DeviceInfo( | ||
| mac_address=device.get("wan_mac_address"), | ||
| serial_number=device.get("serialnumber"), | ||
| model_name=device.get("modelname"), | ||
| model_number=device.get("modelname"), | ||
| product_class=device.get("modelname"), | ||
| software_version=device.get("running", {}).get("version"), |
There was a problem hiding this comment.
REST get_device_info() assumes data[0] is a dict and calls .get(...) unconditionally. If the endpoint returns an empty list or a list with a non-dict element, this will raise AttributeError instead of the intended UnknownException. Add a type check for data[0] (and for the nested device object) before accessing .get().
| device = data[0].get("device", {}) | |
| return DeviceInfo( | |
| mac_address=device.get("wan_mac_address"), | |
| serial_number=device.get("serialnumber"), | |
| model_name=device.get("modelname"), | |
| model_number=device.get("modelname"), | |
| product_class=device.get("modelname"), | |
| software_version=device.get("running", {}).get("version"), | |
| first_item = data[0] | |
| if not isinstance(first_item, dict): | |
| raise UnknownException("Invalid response from /api/v1/device") | |
| device = first_item.get("device", {}) | |
| if not isinstance(device, dict): | |
| raise UnknownException("Invalid response from /api/v1/device") | |
| running = device.get("running", {}) | |
| if not isinstance(running, dict): | |
| running = {} | |
| return DeviceInfo( | |
| mac_address=device.get("wan_mac_address"), | |
| serial_number=device.get("serialnumber"), | |
| model_name=device.get("modelname"), | |
| model_number=device.get("modelname"), | |
| product_class=device.get("modelname"), | |
| software_version=running.get("version"), |
| @pytest.mark.asyncio | ||
| async def test_default_session_accepts_ip_cookies(): | ||
| """Default aiohttp session should accept cookies from IP hosts.""" | ||
| client = SagemcomClient( | ||
| host="192.168.1.1", | ||
| username="admin", | ||
| password="admin", | ||
| authentication_method=EncryptionMethod.MD5, | ||
| ) | ||
| try: | ||
| assert getattr(client.session.cookie_jar, "_unsafe", False) is True | ||
| finally: | ||
| await client.close() |
There was a problem hiding this comment.
This test asserts client.session.cookie_jar._unsafe (a private aiohttp attribute). That makes the test brittle across aiohttp versions. Prefer asserting behavior instead (e.g., set a cookie for an IP-based URL via cookie_jar.update_cookies(...) and assert it is retained), or mock ClientSession creation and assert CookieJar(unsafe=True) was passed.
Summary
This PR adds support for newer Sagemcom firmware that exposes
/api/v1/*endpoints, while preserving compatibility with legacy/cgi/json-reqfirmware.What Changed
ApiModeenum:ApiMode.AUTO(default)ApiMode.LEGACYApiMode.RESTSagemcomClientwithapi_modeparameter.login():LEGACY: force legacy flowREST: force REST flowAUTO: try legacy first, fallback to REST when legacy endpoint appears unavailable.login()viaPOST /api/v1/login(form-encoded)logout()viaPOST /api/v1/logoutget_device_info()viaGET /api/v1/deviceget_hosts()viaGET /api/v1/homeNotImplementedErrorin REST mode.Backward Compatibility
AUTO, so legacy users do not need to change code.Tests
AUTOfallback to REST on legacy 503 HTML response.get_hosts()parsing (wireless + ethernet devices).Documentation
ApiModeusage in quickstart