feat(portaswitch): support API token-based login with version compatibility check

Author: yakimenko73

app/bss/adapters/portaswitch/adapter.py

@@ -1,4 +1,5 @@
 import logging
+import re
 from datetime import datetime, timedelta, UTC
 from typing import Final, Iterator, Optional, Dict, List
 
@@ -69,6 +70,8 @@
 )
 from .utils import generate_otp_id, extract_fault_code, generate_hash_dictionary
 
+PORTASWITCH_VERSION_WITH_TOKEN: Final[str] = "128"
+
 
 class PortaSwitchAdapter(BSSAdapter):
     """Bridges WebTrit Core with PortaSwitch APIs.
@@ -158,7 +161,14 @@ def authenticate(self, user: UserInfo, password: str = None) -> SessionInfo:
             if self._portaswitch_settings.ALLOWED_ADDONS:
                 self._check_allowed_addons(account_info)
 
-            session_data = self._account_api.login(account_info["login"], account_info["password"])
+            version = self._admin_api.get_version()
+            actual_portaswitch_mr = [int(d) for d in re.findall(r'\d+', version)]
+            expected_portaswitch_mr_with_token_support = [int(d) for d in
+                                                          re.findall(r'\d+', PORTASWITCH_VERSION_WITH_TOKEN)]
+
+            token = account_info["password"]
+            session_data = self._account_api.login(account_info["login"], account_info["password"],
+                                                   token if actual_portaswitch_mr <= expected_portaswitch_mr_with_token_support else None)
 
             return SessionInfo(
                 user_id=UserId(str(account_info["i_account"])),

app/bss/adapters/portaswitch/api/account.py

@@ -100,20 +100,24 @@ def decode_response(self, response: requests.models.Response) -> Union[dict, tup
 
         raise ValueError("Not expected response")
 
-    def login(self, login: str, password: str) -> dict:
+    def login(self, login: str, password: str, token: str = None) -> dict:
         """Performs an account login by its login and password.
 
         Parameters:
             :login (str): The login of the account.
             :password (str): The password of the account.
+            :token (str): The token of the account.
 
         Returns:
             :(dict): The API method execution result.
 
         """
-        return self.__send_request(
-            module="Session", method="login", params={"login": login, "password": password}
-        )
+        params = {"login": login, "password": password}
+
+        if token:
+            params["token"] = token
+
+        return self.__send_request(**params)
 
     def logout(self, access_token: str) -> dict:
         """Performs an account logout.

app/bss/adapters/portaswitch/api/admin.py

@@ -1,4 +1,5 @@
 import logging
+from typing import Optional
 
 from bss.adapters.portaswitch.config import PortaSwitchSettings
 from bss.adapters.portaswitch.types import PortaSwitchAdminUser
@@ -168,6 +169,15 @@ def get_env_info(self) -> dict:
 
         return response.get("env_info", dict())
 
+    def get_version(self) -> Optional[str]:
+        """Returns PortaSwitch version.
+        Returns:
+            str: The API method execution result that contains a PortaSwitch version.
+        """
+        response = self._send_request(module="Generic", method="get_version", params={})
+
+        return response.get("version")
+
     def _send_request(self, module: str, method: str, params: dict, turn_off_login: bool = False):
         """Sends the Porta-Billing API method by means of HTTP POST request.