url mmoda validator and token arg adding (#171)

* url mmoda validator and token arg adding

* correct function name

* chjecking mmoda url

* test

* reading token from context dict

* adapting test

* better url updating

nb2workflow/nbadapter.py

@@ -28,7 +28,7 @@
 import scrapbook as sb
 import nbformat
 from nbconvert import HTMLExporter
-from urllib.parse import urlparse
+from urllib.parse import urlparse, parse_qs, urlencode, urlunparse
 
 from . import logstash
 
@@ -37,8 +37,9 @@
 from nb2workflow import workflows
 from nb2workflow.logging_setup import setup_logging
 from nb2workflow.json import CustomJSONEncoder
+from nb2workflow.url_helper import is_mmoda_url
+from nb2workflow.semantics import understand_comment_references
 
-from nb2workflow.semantics import understand_comment_references, oda_ontology_prefix
 from git import Repo, InvalidGitRepositoryError, GitCommandError
 
 import logging
@@ -487,6 +488,8 @@ def _execute(self, parameters, progress_bar=True, log_output=True, inplace=False
             tmpdir =os.path.dirname(os.path.realpath(self.notebook_fn))
             logger.info("executing inplace, no tmpdir is input dir: %s", tmpdir)
 
+        r = self.handle_url_params(parameters, tmpdir, context=context)
+
         if len(context) > 0:
             self._pass_context(tmpdir, context)
 
@@ -495,8 +498,6 @@ def _execute(self, parameters, progress_bar=True, log_output=True, inplace=False
         self.inject_output_gathering()
         exceptions = []
 
-        r = self.extract_files_locally(parameters, tmpdir)
-
         if len(r['exceptions']) > 0:
             exceptions.extend(r['exceptions'])
 
@@ -641,7 +642,7 @@ def download_file(self, file_url, tmpdir):
 
         return file_name
 
-    def extract_files_locally(self, parameters, tmpdir):
+    def handle_url_params(self, parameters, tmpdir, context={}):
         adapted_parameters = copy.deepcopy(parameters)
         exceptions = []
         for input_par_name, input_par_obj in self.input_parameters.items():
@@ -651,6 +652,18 @@ def extract_files_locally(self, parameters, tmpdir):
                 if arg_par_value is None:
                     arg_par_value = input_par_obj['default_value']
                 if validators.url(arg_par_value):
+
+                    if is_mmoda_url(arg_par_value):
+                        token = context.get('token', None)
+                        if token is not None:
+                            logger.debug(f'adding token to the url: {arg_par_value}')
+                            url_parts = urlparse(adapted_parameters[input_par_name])
+                            url_args = parse_qs(url_parts.query)
+                            url_args['token'] = [token]  # the values in the dictionary need to be lists
+                            new_url_parts = url_parts._replace(query=urlencode(url_args, doseq=True))
+                            adapted_parameters[input_par_name] = urlunparse(new_url_parts)
+
+
                     logger.debug(f'download {arg_par_value}')
                     try:
                         file_name = self.download_file(arg_par_value, tmpdir)

nb2workflow/url_helper.py

@@ -0,0 +1,8 @@
+from urllib.parse import urlparse
+
+
+def is_mmoda_url(url):
+    parsed_url = urlparse(url)
+    if 'mmoda' in parsed_url.path:
+        return True
+    return False

tests/test_input_types.py

@@ -30,6 +30,15 @@ def test_posix_download_file_with_arg_wrong_url(client):
                                        "https://fits.gsfc.nasa.gov/samples/aaaaaa.fits. This might be related "
                                        "to an invalid url, please check the input provided')")
 
+def test_posix_download_file_mmoda_url(client):
+    r = client.get('/api/v1.0/get/testposixpath', query_string={
+        'fits_file_path': 'https://www.astro.unige.ch/mmoda/test.fits',
+        '_token': 'test_token'
+    })
+    assert r.json['exceptions'][0] == ("Exception('An issue occurred when attempting to getting the file size at the url "
+                                       "https://www.astro.unige.ch/mmoda/test.fits. This might be related to an "
+                                       "invalid url, please check the input provided')")
+
 def test_boolean_default(client):
     r = client.get('/api/v1.0/get/testbool')
     assert r.json['output']['output'] == 'boolean True'