add the possibility to refresh token when necessary

synology_srm/http.py

@@ -49,22 +49,23 @@ def _login(self):
             method='Login',
             version=2,
             params=params,
-            authorized=False
+            restricted=False
         )
 
         self.sid = response['sid']
 
     def call(self, path: str, api: str, method: str,
         version: int = 1, params: dict = {},
-        authorized: bool = True):
+        restricted: bool = True, retried: bool = False):
         """Performs an HTTP call to the Synology API."""
         url = '{}/{}'.format(
             self._get_base_url(),
             path
         )
 
-        if authorized and self.sid is None:
-            self._login()
+        if restricted:
+            if self.sid is None:
+                self._login()
             params['_sid'] = self.sid
 
         params['api'] = api
@@ -92,6 +93,28 @@ def call(self, path: str, api: str, method: str,
 
         if not data['success']:
             code = data['error']['code']
+            # 106 Session timeout
+            # 107 Session interrupted by duplicate login
+            if code == 106 or code == 107:
+                if not restricted or retried:
+                    # We should stop here if:
+                    #  1 - We are on a public route, no need to retry the login
+                    #  2 - We already retried the route
+                    raise SynologyHttpException(
+                        code,
+                        "Session timeout even when trying to refresh the token"
+                    )
+                self._login()
+                # Retry the current request why a new token
+                return self.call(
+                    path=path,
+                    api=api,
+                    method=method,
+                    version=version,
+                    params=params,
+                    restricted=True,
+                    retried=True
+                )
             if code == 400:
                 raise SynologyIncorrectPasswordException(
                     400,

tests/test_login.py

@@ -49,6 +49,86 @@ def test_successful_login(self, m):
         self.assertEqual(self.http.sid, sid)
         self.http.sid = None
 
+    @requests_mock.Mocker()
+    def test_refresh_sid_login(self, m):
+        m.get('{}/auth.cgi'.format(self.http._get_base_url()), [
+            {
+                'json': {
+                    'data': {
+                        'sid': 'sid_one'
+                    },
+                    'success': True
+                }
+            },
+            {
+                'json': {
+                    'data': {
+                        'sid': 'sid_two'
+                    },
+                    'success': True
+                }
+            }
+        ])
+
+        m.get('{}/entry.cgi'.format(self.http._get_base_url()), [
+            {
+                'json': {
+                    'error': {
+                        'code': 106
+                    },
+                    'success': False
+                }
+            },
+            {
+                'json': {
+                    'data': {
+                         'devices': []
+                    },
+                    'success': True
+                }
+            }
+        ])
+
+        self.http._login()
+        self.assertEqual(self.http.sid, 'sid_one')
+
+        # Try another API to force the SID renewal
+        devices = self.client.mesh.network_wifidevice()
+
+        self.assertEqual(self.http.sid, 'sid_two')
+        self.assertEqual(devices, [])
+
+    @requests_mock.Mocker()
+    def test_loop_redirect(self, m):
+        m.get('{}/auth.cgi'.format(self.http._get_base_url()), json={
+            'data': {
+                'sid': 'sid'
+            },
+            'success': True
+        })
+
+        m.get('{}/entry.cgi'.format(self.http._get_base_url()), [
+            {
+                'json': {
+                    'error': {
+                        'code': 106
+                    },
+                    'success': False
+                }
+            },
+            {
+                'json': {
+                    'error': {
+                        'code': 106
+                    },
+                    'success': False
+                }
+            }
+        ])
+
+        with self.assertRaises(synology_srm.SynologyHttpException) as cm:
+            devices = self.client.mesh.network_wifidevice()
+
     @requests_mock.Mocker()
     def test_login_or_password_incorrect(self, m):
         m.get('{}/auth.cgi'.format(self.http._get_base_url()), json={