Implemented auto refresh for OpenShift API token

esiclient/tests/unit/v1/cluster/test_openshift.py

@@ -39,45 +39,69 @@ class TestCallAssistedInstallerAPI(TestCase):
     @mock.patch('requests.patch', autospec=True)
     @mock.patch('requests.get', autospec=True)
     @mock.patch('requests.post', autospec=True)
-    def test_call_assisted_installer_api_post(self, mock_post,
+    @mock.patch(
+        'esiclient.v1.cluster.openshift.check_and_refresh_token',
+        autospec=True)
+    def test_call_assisted_installer_api_post(self, mock_cart,
+                                              mock_post,
                                               mock_get, mock_patch):
         mock_post.return_value = MockResponse()
 
+        mock_cart.return_value = 'access_token'
+
         openshift.call_assisted_installer_api('test', 'post')
 
         mock_post.assert_called_once_with(
             openshift.BASE_ASSISTED_INSTALLER_URL + "test",
-            headers={}, json=None)
+            headers={
+                'Authorization': 'Bearer access_token'
+            }, json=None)
         mock_get.assert_not_called
         mock_patch.assert_not_called
 
     @mock.patch('requests.patch', autospec=True)
     @mock.patch('requests.get', autospec=True)
     @mock.patch('requests.post', autospec=True)
-    def test_call_assisted_installer_api_get(self, mock_post,
+    @mock.patch(
+        'esiclient.v1.cluster.openshift.check_and_refresh_token',
+        autospec=True)
+    def test_call_assisted_installer_api_get(self, mock_cart,
+                                             mock_post,
                                              mock_get, mock_patch):
         mock_get.return_value = MockResponse()
 
+        mock_cart.return_value = 'access_token'
+
         openshift.call_assisted_installer_api('test', 'get')
 
         mock_get.assert_called_once_with(
             openshift.BASE_ASSISTED_INSTALLER_URL + "test",
-            headers={})
+            headers={
+                'Authorization': 'Bearer access_token'
+            })
         mock_post.assert_not_called
         mock_patch.assert_not_called
 
     @mock.patch('requests.patch', autospec=True)
     @mock.patch('requests.get', autospec=True)
     @mock.patch('requests.post', autospec=True)
-    def test_call_assisted_installer_api_patch(self, mock_post,
+    @mock.patch(
+        'esiclient.v1.cluster.openshift.check_and_refresh_token',
+        autospec=True)
+    def test_call_assisted_installer_api_patch(self, mock_cart,
+                                               mock_post,
                                                mock_get, mock_patch):
         mock_patch.return_value = MockResponse()
 
+        mock_cart.return_value = 'access_token'
+
         openshift.call_assisted_installer_api('test', 'patch')
 
         mock_patch.assert_called_once_with(
             openshift.BASE_ASSISTED_INSTALLER_URL + "test",
-            headers={}, json=None)
+            headers={
+                'Authorization': 'Bearer access_token'
+            }, json=None)
         mock_get.assert_not_called
         mock_post.assert_not_called
 

esiclient/v1/cluster/openshift.py

@@ -10,6 +10,7 @@
 #   License for the specific language governing permissions and limitations
 #   under the License.
 
+import asyncio
 import json
 import logging
 import os
@@ -26,8 +27,45 @@
 BASE_ASSISTED_INSTALLER_URL = \
     'https://api.openshift.com/api/assisted-install/v2/'
 
+AUTHENTICATION_URL = \
+    'https://sso.redhat.com/auth/realms/redhat-external/ \
+        protocol/openid-connect/token'
+
+
+async def check_and_refresh_token():
+    access_token = os.getenv('API_TOKEN', '')
+    offline_token = os.getenv('OFFLINE_TOKEN', '')
+
+    if access_token == '':
+        raise OSAIException('API_TOKEN not set in environment')
+    if offline_token == '':
+        raise OSAIException('OFFLINE_TOKEN not set in environment')
+
+    token_check_url = BASE_ASSISTED_INSTALLER_URL + 'component-versions'
+    auth_headers = {'Authorization': 'Bearer ' + access_token}
+    check_response = requests.get(token_check_url, headers=auth_headers)
+
+    if check_response.status_code != 200:
+        payload = {
+            'grant_type': 'refresh_token',
+            'client_id': 'cloud-services',
+            'refresh_token': offline_token
+        }
+        refresh_response = requests.post(AUTHENTICATION_URL, data=payload)
+        if refresh_response.status_code == 200:
+            response_json = refresh_response.json()
+            access_token = response_json['access_token']
+            os.environ['API_TOKEN'] = access_token
+            return access_token
+        else:
+            raise OSAIException('Failed to refresh token')
+    else:
+        return access_token
+
 
 def call_assisted_installer_api(url, method, headers={}, data=None):
+    access_token = asyncio.run(check_and_refresh_token())
+    headers['Authorization'] = 'Bearer ' + access_token
     full_url = BASE_ASSISTED_INSTALLER_URL + url
     if method == 'post':
         response = requests.post(full_url, headers=headers, json=data)
@@ -93,8 +131,8 @@ def _print_failure_message(self, exception, cluster_config_file,
         if message:
             print(message)
         print("* %s" % str(exception))
-        if isinstance(exception, OSAIException):
-            print("* YOU MAY NEED TO REFRESH YOUR OPENSHIFT API TOKEN")
+        # if isinstance(exception, OSAIException):
+        #     print("* YOU MAY NEED TO REFRESH YOUR OPENSHIFT API TOKEN")
         print("Run this command to continue installation:")
         print("* %s" % command)
         return