feat: use api_token for PortaSwitch Session/login on MR128+

Author: yakimenko73

app/bss/adapters/portaswitch/adapter.py

@@ -1,5 +1,6 @@
 import logging
 import re
+import uuid
 from datetime import datetime, timedelta, UTC
 from typing import Final, Iterator, Optional, Dict, List
 
@@ -162,8 +163,11 @@ def authenticate(self, user: UserInfo, password: str = None) -> SessionInfo:
             if self._portaswitch_settings.ALLOWED_ADDONS:
                 self._check_allowed_addons(account_info)
 
-            token = account_info["password"] if self._is_portaswitch_version_with_token() else None
-            session_data = self._account_api.login(account_info["login"], account_info["password"], token)
+            if self._is_portaswitch_version_with_token():
+                token = self._get_or_create_api_token(account_info)
+                session_data = self._account_api.login(account_info["login"], token=token)
+            else:
+                session_data = self._account_api.login(account_info["login"], account_info["password"], token=account_info["password"])
 
             return SessionInfo(
                 user_id=UserId(str(account_info["i_account"])),
@@ -1420,12 +1424,24 @@ def _get_all_accounts_by_customer(self, i_customer: int) -> list[dict]:
 
         return all_accounts
 
+    def _get_or_create_api_token(self, account_info: dict) -> str:
+        """Return the account's api_token, creating and persisting one if absent."""
+        api_token = account_info.get("api_token")
+        if not api_token:
+            api_token = str(uuid.uuid4())
+            self._admin_api.update_account(account_info["i_account"], api_token=api_token)
+            logging.info(f"Created new api_token for i_account={account_info['i_account']}")
+        return api_token
+
     def _emulate_account_login(self, i_account: str) -> dict:
         """Emulate a login for a PortaSwitch account."""
         account_info = self._admin_api.get_account_info(i_account=i_account).get("account_info")
 
-        token = account_info["password"] if self._is_portaswitch_version_with_token() else None
-        return self._account_api.login(account_info["login"], account_info["password"], token)
+        if self._is_portaswitch_version_with_token():
+            token = self._get_or_create_api_token(account_info)
+            return self._account_api.login(account_info["login"], token=token)
+        else:
+            return self._account_api.login(account_info["login"], account_info["password"], token=account_info["password"])
 
     def _is_portaswitch_version_with_token(self) -> bool:
         """Check if the actual version of PortaSwitch is a version with token support."""
@@ -1434,4 +1450,4 @@ def _is_portaswitch_version_with_token(self) -> bool:
         expected_portaswitch_mr_with_token_support = [int(d) for d in
                                                       re.findall(r'\d+', PORTASWITCH_VERSION_WITH_TOKEN)]
 
-        return actual_portaswitch_mr < expected_portaswitch_mr_with_token_support
+        return actual_portaswitch_mr >= expected_portaswitch_mr_with_token_support

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

@@ -100,19 +100,22 @@ def decode_response(self, response: requests.models.Response) -> Union[dict, tup
 
         raise ValueError("Not expected response")
 
-    def login(self, login: str, password: str, token: str = None) -> dict:
-        """Performs an account login by its login and password.
+    def login(self, login: str, password: str = None, token: str = None) -> dict:
+        """Performs an account login by its login, password and/or token.
 
         Parameters:
             :login (str): The login of the account.
-            :password (str): The password of the account.
-            :token (str): The token of the account.
+            :password (str): The password of the account. Not required when token is provided.
+            :token (str): The api_token of the account.
 
         Returns:
             :(dict): The API method execution result.
 
         """
-        params = {"login": login, "password": password}
+        params = {"login": login}
+
+        if password:
+            params["password"] = password
 
         if token:
             params["token"] = token

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

@@ -156,10 +156,32 @@ def get_account_info(self, **params) -> dict:
             params={
                 "without_service_features": 1,
                 "detailed_info": 1,
+                "get_auth_info": 1,
                 **params
             }
         )
 
+    def update_account(self, i_account: int, **account_info_fields) -> dict:
+        """Updates account fields for the given i_account.
+
+        Parameters:
+            i_account (int): The internal PortaSwitch account identifier.
+            **account_info_fields: Fields to update inside account_info (e.g. api_token="...").
+
+        Returns:
+            dict: The API method execution result.
+        """
+        return self._send_request(
+            module="Account",
+            method="update_account",
+            params={
+                "account_info": {
+                    "i_account": i_account,
+                    **account_info_fields,
+                },
+            },
+        )
+
     def get_env_info(self) -> dict:
         """Returns PortaSwitch environment info.
         Returns: