[2024] Code quality fixes (#266)

This pull request includes various changes to improve the `sagemcom_api`
codebase. The most important changes include adding flexibility in
initializing the `Device` class, modifying the `EncryptionMethod` enum
class to ensure uniqueness of values, and removing unused imports and
code blocks in the `sagemcom_api/client.py` file.

Main functionality changes:

*
[`sagemcom_api/models.py`](diffhunk://#diff-1f5ad46f1c332379d35609bca9ea626894e97d6e40c47cfb5cbe83d2c4258c34L106-R129):
Added flexibility in initializing the `Device` class, removed unused
attributes, modified attribute values, and updated type annotations.
[[1]](diffhunk://#diff-1f5ad46f1c332379d35609bca9ea626894e97d6e40c47cfb5cbe83d2c4258c34L106-R129)
[[2]](diffhunk://#diff-1f5ad46f1c332379d35609bca9ea626894e97d6e40c47cfb5cbe83d2c4258c34R51-R58)
[[3]](diffhunk://#diff-1f5ad46f1c332379d35609bca9ea626894e97d6e40c47cfb5cbe83d2c4258c34R8)
[[4]](diffhunk://#diff-1f5ad46f1c332379d35609bca9ea626894e97d6e40c47cfb5cbe83d2c4258c34R71)
[[5]](diffhunk://#diff-1f5ad46f1c332379d35609bca9ea626894e97d6e40c47cfb5cbe83d2c4258c34L97)
[[6]](diffhunk://#diff-1f5ad46f1c332379d35609bca9ea626894e97d6e40c47cfb5cbe83d2c4258c34L80-L82)
[[7]](diffhunk://#diff-1f5ad46f1c332379d35609bca9ea626894e97d6e40c47cfb5cbe83d2c4258c34R154-R161)
*
[`sagemcom_api/enums.py`](diffhunk://#diff-696d324cfc32d09ae617a7597df251392c35ae1c07be0c0eb30e87d4142937c8R2-R14):
Modified the `EncryptionMethod` enum class to use the `StrEnum` class
and ensure uniqueness of values.
*
[`sagemcom_api/client.py`](diffhunk://#diff-052218051094ff1cc2c8352cd555cfb9eb02bf261eb5e635cf73b798809306a5L10-L17):
Removed unused imports, modified headers and parameters, updated type
annotations, added comments, and removed code blocks.
[[1]](diffhunk://#diff-052218051094ff1cc2c8352cd555cfb9eb02bf261eb5e635cf73b798809306a5L10-L17)
[[2]](diffhunk://#diff-052218051094ff1cc2c8352cd555cfb9eb02bf261eb5e635cf73b798809306a5L84-R88)
[[3]](diffhunk://#diff-052218051094ff1cc2c8352cd555cfb9eb02bf261eb5e635cf73b798809306a5L340-R340)
[[4]](diffhunk://#diff-052218051094ff1cc2c8352cd555cfb9eb02bf261eb5e635cf73b798809306a5L215-R216)
[[5]](diffhunk://#diff-052218051094ff1cc2c8352cd555cfb9eb02bf261eb5e635cf73b798809306a5L328-R327)
[[6]](diffhunk://#diff-052218051094ff1cc2c8352cd555cfb9eb02bf261eb5e635cf73b798809306a5L365-R364)
[[7]](diffhunk://#diff-052218051094ff1cc2c8352cd555cfb9eb02bf261eb5e635cf73b798809306a5L392-R391)
[[8]](diffhunk://#diff-052218051094ff1cc2c8352cd555cfb9eb02bf261eb5e635cf73b798809306a5L193)
[[9]](diffhunk://#diff-052218051094ff1cc2c8352cd555cfb9eb02bf261eb5e635cf73b798809306a5L381-R380)
[[10]](diffhunk://#diff-052218051094ff1cc2c8352cd555cfb9eb02bf261eb5e635cf73b798809306a5L354-R353)
[[11]](diffhunk://#diff-052218051094ff1cc2c8352cd555cfb9eb02bf261eb5e635cf73b798809306a5L282-R283)
[[12]](diffhunk://#diff-052218051094ff1cc2c8352cd555cfb9eb02bf261eb5e635cf73b798809306a5L295-R296)
[[13]](diffhunk://#diff-052218051094ff1cc2c8352cd555cfb9eb02bf261eb5e635cf73b798809306a5L96-R100)
[[14]](diffhunk://#diff-052218051094ff1cc2c8352cd555cfb9eb02bf261eb5e635cf73b798809306a5L308-R315)
[[15]](diffhunk://#diff-052218051094ff1cc2c8352cd555cfb9eb02bf261eb5e635cf73b798809306a5R44-R59)

Other changes:

*
[`pyproject.toml`](diffhunk://#diff-50c86b7ed8ac2cf95bd48334961bf0530cdc77b5a56f852c5c61b89d735fd711R23):
Added `backports-strenum` package as a dependency for Python versions
lower than 3.11.
*
[`sagemcom_api/__init__.py`](diffhunk://#diff-43f61fbb6f1f76bed40f65c0d68f581efef61f6684219f1afc7363a60be0f336L2):
Removed the unused `__version__` variable.
*
[`setup.cfg`](diffhunk://#diff-fa602a8a75dc9dcc92261bac5f533c2a85e34fcceaff63b3a3a81d9acde2fc52R1-R5):
Added ignore rule for `E501` error code in `flake8` linting.
*
[`tests/test_sagemcom.py`](diffhunk://#diff-0f905eca51ada7d34dc2d2978c6c41945cf4a8179cb58bcb446219f97d335aaeL2-L8):
Removed the `test_version` function as it was no longer needed.
*
[`sagemcom_api/exceptions.py`](diffhunk://#diff-63c847110513149341a9f7e94d412bfb7c67d79df07a321774a9a2f3329ca34fL8-L58):
Removed `pass` statements from several exception classes.

.gitattributes

@@ -1,4 +1,4 @@
 # Ensure Docker script files uses LF to support Docker for Windows.
 # Ensure "git config --global core.autocrlf input" before you clone
 *     text eol=lf
-*.py  whitespace=error
+*.py  whitespace=error

pyproject.toml

@@ -20,6 +20,7 @@ packages = [
 python = ">=3.9,<4.0"
 aiohttp = "^3.7.3"
 pyhumps = "^3.0.2"
+backports-strenum = { version = "^1.3.1", python = "<3.11" }
 
 [tool.poetry.dev-dependencies]
 pytest = "^7.1"

sagemcom_api/__init__.py

@@ -1,2 +1 @@
 """Package to communicate with Sagemcom F@st internal APIs."""
-__version__ = "1.0.1"

sagemcom_api/client.py

@@ -7,14 +7,12 @@
 import math
 import random
 from types import TracebackType
-from typing import Dict, List, Optional, Type
 import urllib.parse
 
 from aiohttp import ClientSession, ClientTimeout
 from aiohttp.connector import TCPConnector
 import humps
 
-from . import __version__
 from .const import (
     API_ENDPOINT,
     DEFAULT_TIMEOUT,
@@ -43,18 +41,22 @@
 from .models import Device, DeviceInfo, PortMapping
 
 
+# pylint: disable=too-many-instance-attributes
 class SagemcomClient:
     """Client to communicate with the Sagemcom API."""
 
+    _auth_key: str | None
+
+    # pylint: disable=too-many-arguments
     def __init__(
         self,
-        host,
-        username,
-        password,
-        authentication_method,
-        session: ClientSession = None,
-        ssl=False,
-        verify_ssl=True,
+        host: str,
+        username: str,
+        password: str,
+        authentication_method: EncryptionMethod,
+        session: ClientSession | None = None,
+        ssl: bool | None = False,
+        verify_ssl: bool | None = True,
     ):
         """
         Create a SagemCom client.
@@ -81,9 +83,9 @@ def __init__(
             session
             if session
             else ClientSession(
-                headers={"User-Agent": f"{DEFAULT_USER_AGENT}/{__version__}"},
+                headers={"User-Agent": f"{DEFAULT_USER_AGENT}"},
                 timeout=ClientTimeout(DEFAULT_TIMEOUT),
-                connector=TCPConnector(ssl=verify_ssl),
+                connector=TCPConnector(verify_ssl=verify_ssl if verify_ssl else True),
             )
         )
 
@@ -93,9 +95,9 @@ async def __aenter__(self) -> SagemcomClient:
 
     async def __aexit__(
         self,
-        exc_type: Optional[Type[BaseException]],
-        exc_value: Optional[BaseException],
-        traceback: Optional[TracebackType],
+        exc_type: type[BaseException] | None,
+        exc_value: BaseException | None,
+        traceback: TracebackType | None,
     ) -> None:
         """Close session on exit."""
         await self.close()
@@ -190,7 +192,6 @@ async def __api_request_async(self, actions, priority=False):
         async with self.session.post(
             api_host, data="req=" + json.dumps(payload, separators=(",", ":"))
         ) as response:
-
             if response.status == 400:
                 result = await response.text()
                 raise BadRequestException(result)
@@ -212,7 +213,7 @@ async def __api_request_async(self, actions, priority=False):
 
                 # Error in one of the actions
                 if error["description"] == XMO_REQUEST_ACTION_ERR:
-
+                    # pylint:disable=fixme
                     # TODO How to support multiple actions + error handling?
                     actions = result["reply"]["actions"]
                     for action in actions:
@@ -279,8 +280,8 @@ async def login(self):
             self._session_id = data["id"]
             self._server_nonce = data["nonce"]
             return True
-        else:
-            raise UnauthorizedException(data)
+
+        raise UnauthorizedException(data)
 
     async def logout(self):
         """Log out of the Sagemcom F@st device."""
@@ -292,9 +293,7 @@ async def logout(self):
         self._server_nonce = ""
         self._request_id = -1
 
-    async def get_value_by_xpath(
-        self, xpath: str, options: Optional[Dict] = {}
-    ) -> Dict:
+    async def get_value_by_xpath(self, xpath: str, options: dict | None = None) -> dict:
         """
         Retrieve raw value from router using XPath.
 
@@ -305,15 +304,15 @@ async def get_value_by_xpath(
             "id": 0,
             "method": "getValue",
             "xpath": urllib.parse.quote(xpath),
-            "options": options,
+            "options": options if options else {},
         }
 
         response = await self.__api_request_async([actions], False)
         data = self.__get_response_value(response)
 
         return data
 
-    async def get_values_by_xpaths(self, xpaths, options: Optional[Dict] = {}) -> Dict:
+    async def get_values_by_xpaths(self, xpaths, options: dict | None = None) -> dict:
         """
         Retrieve raw values from router using XPath.
 
@@ -325,7 +324,7 @@ async def get_values_by_xpaths(self, xpaths, options: Optional[Dict] = {}) -> Di
                 "id": i,
                 "method": "getValue",
                 "xpath": urllib.parse.quote(xpath),
-                "options": options,
+                "options": options if options else {},
             }
             for i, xpath in enumerate(xpaths.values())
         ]
@@ -337,8 +336,8 @@ async def get_values_by_xpaths(self, xpaths, options: Optional[Dict] = {}) -> Di
         return data
 
     async def set_value_by_xpath(
-        self, xpath: str, value: str, options: Optional[Dict] = {}
-    ) -> Dict:
+        self, xpath: str, value: str, options: dict | None = None
+    ) -> dict:
         """
         Retrieve raw value from router using XPath.
 
@@ -351,7 +350,7 @@ async def set_value_by_xpath(
             "method": "setValue",
             "xpath": urllib.parse.quote(xpath),
             "parameters": {"value": str(value)},
-            "options": options,
+            "options": options if options else {},
         }
 
         response = await self.__api_request_async([actions], False)
@@ -362,7 +361,7 @@ async def get_device_info(self) -> DeviceInfo:
         """Retrieve information about Sagemcom F@st device."""
         try:
             data = await self.get_value_by_xpath("Device/DeviceInfo")
-            return DeviceInfo(**data.get("device_info"))
+            return DeviceInfo(**data["device_info"])
         except UnknownPathException:
             data = await self.get_values_by_xpaths(
                 {
@@ -378,7 +377,7 @@ async def get_device_info(self) -> DeviceInfo:
 
         return DeviceInfo(**data)
 
-    async def get_hosts(self, only_active: Optional[bool] = False) -> List[Device]:
+    async def get_hosts(self, only_active: bool | None = False) -> list[Device]:
         """Retrieve hosts connected to Sagemcom F@st device."""
         data = await self.get_value_by_xpath("Device/Hosts/Hosts")
         devices = [Device(**d) for d in data]
@@ -389,7 +388,7 @@ async def get_hosts(self, only_active: Optional[bool] = False) -> List[Device]:
 
         return devices
 
-    async def get_port_mappings(self) -> List[PortMapping]:
+    async def get_port_mappings(self) -> list[PortMapping]:
         """Retrieve configured Port Mappings on Sagemcom F@st device."""
         data = await self.get_value_by_xpath("Device/NAT/PortMappings")
         port_mappings = [PortMapping(**p) for p in data]

sagemcom_api/enums.py

@@ -1,9 +1,17 @@
 """Enums for the Sagemcom F@st client."""
+from enum import unique
+import sys
 
-from enum import Enum
+# Since we support Python versions lower than 3.11, we use
+# a backport for StrEnum when needed.
+if sys.version_info >= (3, 11):
+    from enum import StrEnum
+else:
+    from backports.strenum import StrEnum
 
 
-class EncryptionMethod(Enum):
+@unique
+class EncryptionMethod(StrEnum):
     """Encryption method defining the password hash."""
 
     MD5 = "MD5"

sagemcom_api/exceptions.py

@@ -5,54 +5,35 @@
 class AccessRestrictionException(Exception):
     """Raised when current user has access restrictions."""
 
-    pass
-
 
 class AuthenticationException(Exception):
     """Raised when authentication is not correct."""
 
-    pass
-
 
 class LoginTimeoutException(Exception):
     """Raised when a timeout is encountered during login."""
 
-    pass
-
 
 class NonWritableParameterException(Exception):
     """Raised when provided parameter is not writable."""
 
-    pass
-
 
 class UnknownPathException(Exception):
     """Raised when provided path does not exist."""
 
-    pass
-
 
 class MaximumSessionCountException(Exception):
     """Raised when the maximum session count is reached."""
 
-    pass
 
-
-# Exceptions provided by this library
-# TODO Validate our own errors
+# Broad exceptions provided by this library
 class BadRequestException(Exception):
     """TODO."""
 
-    pass
-
 
 class UnauthorizedException(Exception):
     """TODO."""
 
-    pass
-
 
 class UnknownException(Exception):
     """TODO."""
-
-    pass