Add confirm_access method

Author: jaladh-singhal

firefly_client/fc_utils.py

@@ -64,8 +64,8 @@ def ensure3(val, name):
     return ret
 
 
-# actions from Firefly
 ACTION_DICT = {
+    # actions from Firefly JS
     'ShowFits': 'ImagePlotCntlr.PlotImage',
     'AddExtension': 'ExternalAccessCntlr/extensionAdd',
     'FetchTable': 'table.fetch',
@@ -95,6 +95,8 @@ def ensure3(val, name):
     'ShowHiPS': 'ImagePlotCntlr.PlotHiPS',
     'ShowImageOrHiPS': 'ImagePlotCntlr.plotHiPSOrImage',
     'ImagelineBasedFootprint': 'DrawLayerCntlr.ImageLineBasedFP.imagelineBasedFPCreate',
+
+    # actions from jupyter_firefly_extensions
     'StartLabWindow': 'StartLabWindow',
     'StartBrowserTab': 'StartBrowserTab'
 }

firefly_client/firefly_client.py

@@ -241,6 +241,7 @@ def __init__(self, url, channel, html_file=_def_html_file, token=None, viewer_ov
         token and ssl and self.session.headers.update(self.auth_headers)
         not ssl and token and warn('token ignored: should be None when url starts with http://')
         self.firefly_viewer = FireflyClient.get_viewer_mode(html_file,viewer_override)
+        FireflyClient.confirm_access(url, token)
         debug('new instance: %s' % url)
 
     def _lab_env_tab_start(self, tab_type, html_file):
@@ -273,6 +274,19 @@ def get_viewer_mode(html_file, viewer_override):
                 return UNKNOWN
         else:
             return FireflyClient.SLATE_VIEWER if html_file == 'slate.html' else FireflyClient.TRIVIEW_VIEWER
+        
+    @staticmethod
+    def confirm_access(url, token):
+        headers = {'Authorization': f'Bearer {token}'} if token else None
+        healthz_url = url + ('healthz' if url.endswith('/') else '/healthz')
+        response = requests.get(healthz_url, headers=headers, allow_redirects=False)
+        if response.status_code == 200:
+            return True
+        else:
+            raise ValueError(
+                f"Access to {url} failed with status code {response.status_code}.\n"
+                f"GET Headers: {response.headers}"
+            )
 
     def _send_url_as_get(self, url):
         return self.call_response(self.session.get(url, headers=self.header_from_ws))