fix: Token re-authentication and night light model bug (#409)

* fix: data types

* Fix tests

* Add re-authentication logic for expired tokens

* Add Tests

---------

Co-authored-by: cdnninja <[email protected]>

Author: webdjoe

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "pyvesync"
-version = "3.1.1"
+version = "3.1.2"
 description = "pyvesync is a library to manage Etekcity Devices, Cosori Air Fryers, and Levoit Air Purifiers run on the VeSync app."
 readme = "README.md"
 requires-python = ">=3.11"

src/pyvesync/auth.py

@@ -145,6 +145,15 @@ def set_credentials(
         self._country_code = country_code.upper()
         self._current_region = region
 
+    async def reauthenticate(self) -> bool:
+        """Re-authenticate using stored username and password.
+
+        Returns:
+            True if re-authentication successful, False otherwise
+        """
+        self.clear_credentials()
+        return await self.login()
+
     async def load_credentials_from_file(
         self, file_path: str | Path | None = None
     ) -> bool:

src/pyvesync/const.py

@@ -34,6 +34,7 @@
 
 from pyvesync.utils.enum_utils import IntEnumMixin
 
+MAX_API_REAUTH_RETRIES = 3
 DEFAULT_LANGUAGE = 'en'
 API_BASE_URL = None  # Global URL (non-EU regions): "https://smartapi.vesync.com"
 # If device is out of reach, the cloud api sends a timeout response after 7 seconds,

src/pyvesync/devices/vesyncpurifier.py

@@ -146,7 +146,7 @@ def _set_purifier_state(self, result: PurifierCoreDetailsResult) -> None:
             self.state.pm25 = result.air_quality_value
             self.state.set_air_quality_level(result.air_quality)
         if result.night_light is not None:
-            self.state.nightlight_status = DeviceStatus.from_bool(result.night_light)
+            self.state.nightlight_status = DeviceStatus(result.night_light)
 
     async def get_details(self) -> None:
         r_dict = await self.call_bypassv2_api('getPurifierStatus')

src/pyvesync/models/purifier_models.py

@@ -124,7 +124,7 @@ class PurifierCoreDetailsResult(InnerPurifierBaseResult):
     configuration: PurifierCoreDetailsConfig | None = None
     extension: dict | None = None
     air_quality_value: int | None = None
-    night_light: bool | None = None
+    night_light: str | None = None
     fan_rotate: str | None = None
 
 

src/pyvesync/utils/errors.py

@@ -799,9 +799,11 @@ def __init__(self, msg: str) -> None:
 class VeSyncTokenError(VeSyncError):
     """Exception raised for VeSync API authentication errors."""
 
-    def __init__(self) -> None:
+    def __init__(self, msg: str | None = None) -> None:
         """Initialize the exception with a message."""
-        super().__init__('Token expired or invalid - please re-authenticate with login()')
+        super().__init__(
+            f'Token expired or invalid - {msg if msg else "Re-authentication required"}'
+        )
 
 
 class VeSyncServerError(VeSyncError):
@@ -857,8 +859,6 @@ def raise_api_errors(error_info: ResponseInfo) -> None:
             raise VeSyncRateLimitError
         case ErrorTypes.AUTHENTICATION:
             raise VeSyncLoginError(error_info.message)
-        case ErrorTypes.TOKEN_ERROR:
-            raise VeSyncTokenError
         case ErrorTypes.SERVER_ERROR:
             msg = (
                 f'{error_info.message} - '

src/pyvesync/vesync.py

@@ -16,6 +16,7 @@
 from pyvesync.const import (
     DEFAULT_REGION,
     DEFAULT_TZ,
+    MAX_API_REAUTH_RETRIES,
     REGION_API_MAP,
     STATUS_OK,
 )
@@ -34,6 +35,7 @@
     VeSyncAPIStatusCodeError,
     VeSyncError,
     VeSyncServerError,
+    VeSyncTokenError,
     raise_api_errors,
 )
 from pyvesync.utils.helpers import Helpers
@@ -47,6 +49,7 @@ class VeSync:  # pylint: disable=function-redefined
 
     __slots__ = (
         '__weakref__',
+        '_api_attempts',
         '_auth',
         '_close_session',
         '_debug',
@@ -130,6 +133,7 @@ class is instantiated, call `await manager.login()` to log in to VeSync servers,
                 Object to store device state information
         """
         self.session = session
+        self._api_attempts = 0
         self._close_session = False
         self._debug = debug
         self.redact = redact
@@ -381,6 +385,7 @@ async def login(self) -> bool:  # pylint: disable=W9006 # pylint mult docstring
             VeSyncAPIResponseError: If API response is invalid.
             VeSyncServerError: If server returns an error.
         """
+        self.enabled = False
         success = await self._auth.login()
         if success:
             self.enabled = True
@@ -427,6 +432,24 @@ async def __aexit__(self, *exec_info: object) -> None:
             return
         logger.debug('Session not closed, exiting context manager')
 
+    async def _reauthenticate(self) -> bool:
+        """Re-authenticate using stored username and password.
+
+        Returns:
+            True if re-authentication successful, False otherwise
+        """
+        self.enabled = False
+        self._api_attempts += 1
+        if self._api_attempts >= MAX_API_REAUTH_RETRIES:
+            logger.error('Max API re-authentication attempts reached')
+            raise VeSyncTokenError
+        success = await self.auth.reauthenticate()
+        if success:
+            self.enabled = True
+            self._api_attempts = 0
+            return True
+        return await self.auth.reauthenticate()
+
     async def async_call_api(
         self,
         api: str,
@@ -463,6 +486,7 @@ async def async_call_api(
         if self.session is None:
             self.session = ClientSession()
             self._close_session = True
+        self._api_attempts += 1
         response = None
         status_code = None
         if isinstance(json_object, DataClassORJSONMixin):
@@ -494,6 +518,13 @@ async def async_call_api(
                 resp_dict = Helpers.try_json_loads(resp_bytes)
                 if isinstance(resp_dict, dict):
                     error_info = ErrorCodes.get_error_info(resp_dict.get('code'))
+                    if error_info.error_type == ErrorTypes.TOKEN_ERROR:
+                        if await self._reauthenticate():
+                            self.enabled = True
+                            return await self.async_call_api(
+                                api, method, json_object, headers
+                            )
+                        raise VeSyncTokenError
                     if resp_dict.get('msg') is not None:
                         error_info.message = f'{error_info.message} ({resp_dict["msg"]})'
                         raise_api_errors(error_info)

src/tests/call_json.py

@@ -96,6 +96,15 @@
 }
 
 
+def response_body(code: int = 0, msg: str | None = 'Success') -> dict[str, Any]:
+    """Return base response dictionary."""
+    return {
+        'code': code,
+        'traceId': TestDefaults.trace_id,
+        'msg': msg
+    }
+
+
 class LoginResponses:
     GET_TOKEN_RESPONSE_SUCCESS = {
         "traceId": TestDefaults.trace_id,

src/tests/call_json_purifiers.py

@@ -87,7 +87,7 @@ class PurifierDefaults:
           "level": PurifierDefaults.fan_level,
           "display": bool(PurifierDefaults.display),
           "child_lock": bool(PurifierDefaults.child_lock),
-          "night_light": bool(PurifierDefaults.night_light),
+          "night_light": PurifierDefaults.night_light.value,
           "configuration": {
             "display": bool(PurifierDefaults.display),
             "display_forever": PurifierDefaults.display_forever

src/tests/test_x_vesync_api_responses.py

@@ -8,36 +8,32 @@
 
 from pyvesync import VeSync
 from pyvesync.utils.errors import VeSyncRateLimitError, VeSyncServerError
+
 from pyvesync.const import API_BASE_URL_US
 from pyvesync.utils.errors import (
     VeSyncAPIStatusCodeError
     )
+import call_json
 from defaults import TestDefaults
 from aiohttp_mocker import AiohttpMockSession
 
+
 DEFAULT_ENDPOINT = '/endpoint'
 DEFAULT_POST_DATA = {'key': 'value'}
-
-
-def response_dict(code, msg):
-    """Return a response dictionary."""
-    return {'code': code, 'msg': msg}
-
-
 PARAM_ARGS = "endpoint, method, resp_bytes, resp_status"
 
 # Successful API calls should return the response in bytes and a 200 status code
-SUCCESS_RESP = response_dict(0, 'Success')
+SUCCESS_RESP = call_json.response_body(0, 'Success')
 
 
 # Rate limit errors should raise an exception in `async_call_api`
 RATE_LIMIT_CODE = -11003000
-RATE_LIMIT_RESP = response_dict(RATE_LIMIT_CODE, "Rate limit exceeded")
+RATE_LIMIT_RESP = call_json.response_body(RATE_LIMIT_CODE, "Rate limit exceeded")
 
 
 # Server errors should raise an exception in `async_call_api`
 SERVER_ERROR = -11102000
-SERVER_ERROR_RESP = response_dict(SERVER_ERROR, "Server error")
+SERVER_ERROR_RESP = call_json.response_body(SERVER_ERROR, "Server error")
 
 
 # Status code errors should raise an exception in `async_call_api`
@@ -53,7 +49,7 @@ def response_dict(code, msg):
 # Device errors should return the response and a 200 status code
 # with no exception thrown by `async_call_api`
 DEVICE_ERROR_CODE = -11901000
-DEVICE_ERROR_RESP = response_dict(DEVICE_ERROR_CODE, "Device error")
+DEVICE_ERROR_RESP = call_json.response_body(DEVICE_ERROR_CODE, "Device error")
 
 
 class TestApiFunc:
@@ -75,10 +71,6 @@ def setup(self, caplog):
         self.caplog = caplog
         self.caplog.set_level(logging.DEBUG)
         self.loop = asyncio.new_event_loop()
-        # self.mock_api = self.mock_api_call.start()
-        # self.mock_api.return_value.ok = True
-        # self.mock = aioresponses()
-        # self.mock.start()
         self.mock = MagicMock()
         self.manager = VeSync('EMAIL', 'PASSWORD')
         self.manager.verbose = True
@@ -116,7 +108,7 @@ def test_api_success(self, mock):
     @patch("pyvesync.vesync.ClientSession")
     def test_api_rate_limit(self, mock):
         """Test rate limit error - raises `VeSyncRateLimitError` from `VeSync.async_call_api`."""
-        rate_limit_resp = response_dict(RATE_LIMIT_CODE, "Rate limit exceeded")
+        rate_limit_resp = call_json.response_body(RATE_LIMIT_CODE, "Rate limit exceeded")
         mock.return_value.request.return_value = AiohttpMockSession(
             method="post",
             url=API_BASE_URL_US + DEFAULT_ENDPOINT,