NXSAPCR-1091: Added functionality to download proprietary data

 - added corresponding tests

Author: javier-ballester

astroquery/esa/xmm_newton/core.py

@@ -12,6 +12,7 @@
 
 """
 import re
+from getpass import getpass
 from ...utils.tap.core import TapPlus
 from ...query import BaseQuery
 import shutil
@@ -21,7 +22,7 @@
 import os
 
 from astropy.io import fits
-from . import conf
+from . import conf, config
 from astroquery import log
 from astropy.coordinates import SkyCoord
 from ...exceptions import LoginError
@@ -41,14 +42,13 @@ def __init__(self, tap_handler=None):
         super(XMMNewtonClass, self).__init__()
 
         if tap_handler is None:
-            self._tap = TapPlus(url="https://nxsa.esac.esa.int"
-                                    "/tap-server/tap")
+            self._tap = TapPlus(url="https://nxsa.esac.esa.int/tap-server/tap")
         else:
             self._tap = tap_handler
         self._rmf_ftp = str("http://sasdev-xmm.esac.esa.int/pub/ccf/constituents/extras/responses/")
 
     def download_data(self, observation_id, *, filename=None, verbose=False,
-                      cache=True, **kwargs):
+                      cache=True, prop=False, username=None, password=None, **kwargs):
         """
         Download data from XMM-Newton
 
@@ -100,42 +100,38 @@ def download_data(self, observation_id, *, filename=None, verbose=False,
         None if not verbose. It downloads the observation indicated
         If verbose returns the filename
         """
-        if filename is not None:
-            filename = os.path.splitext(filename)[0]
+        """
+        Here we change the log level so that it is above 20, this is to stop a log.debug in query.py. this debug
+        reveals the url being sent which in turn reveals the users username and password
+        """
+        previouslevel = log.getEffectiveLevel()
+        log.setLevel(50)
 
-        link = self.data_aio_url + "obsno=" + observation_id
+        # create url to access the aio
+        link = self._create_link(observation_id, **kwargs)
 
-        link = link + "".join("&{0}={1}".format(key, val)
-                              for key, val in kwargs.items())
+        # If the user wants to access proprietary data, ask them for there credentials
+        if prop:
+            username, password = self._get_username_and_password(credentials_file)
+            link = f"{link}&AIOUSER={username}&AIOPWD={password}"
 
         if verbose:
             log.info(link)
 
-        # we can cache this HEAD request - the _download_file one will check
-        # the file size and will never cache
-        response = self._request('HEAD', link, save=False, cache=cache)
-
-        # Get original extension
-        if 'Content-Type' in response.headers and 'text' not in response.headers['Content-Type']:
-            _, params = cgi.parse_header(response.headers['Content-Disposition'])
-        else:
-            if response.status_code == 401:
-                error = "Data protected by proprietary rights. Please check your credentials"
-                raise LoginError(error)
-        response.raise_for_status()
-
+        # get response of created url
+        params = self._request_link(link, cache)
         r_filename = params["filename"]
         suffixes = Path(r_filename).suffixes
+        print(suffixes)
 
-        if filename is None:
-            filename = observation_id
-
-        filename += "".join(suffixes)
+        # get desired filename
+        filename = self._create_filename(filename, observation_id, suffixes)
 
         self._download_file(link, filename, head_safe=True, cache=cache)
 
         if verbose:
-            log.info("Wrote {0} to {1}".format(link, filename))
+            log.info(f"Wrote {link} to {filename}")
+        log.setLevel(previouslevel)
 
     def get_postcard(self, observation_id, *, image_type="OBS_EPIC",
                      filename=None, verbose=False):
@@ -186,12 +182,12 @@ def get_postcard(self, observation_id, *, image_type="OBS_EPIC",
         else:
             filename = observation_id + ".png"
 
-        log.info("Copying file to {0}...".format(filename))
+        log.info(f"Copying file to {filename}...")
 
         shutil.move(local_filepath, filename)
 
         if verbose:
-            log.info("Wrote {0} to {1}".format(link, filename))
+            log.info(f"Wrote {link} to {filename}")
 
         return filename
 
@@ -273,14 +269,53 @@ def get_columns(self, table_name, *, only_names=True, verbose=False):
                 break
 
         if columns is None:
-            raise ValueError("table name specified is not found in "
-                             "XSA TAP service")
+            raise ValueError("table name specified is not found in XSA TAP service")
 
         if only_names:
             return [c.name for c in columns]
         else:
             return columns
 
+    def _create_link(self, observation_id, **kwargs):
+        link = f"{self.data_aio_url}obsno={observation_id}"
+        link = link + "".join("&{0}={1}".format(key, val)
+                              for key, val in kwargs.items())
+        return link
+
+    def _request_link(self, link, cache):
+        # we can cache this HEAD request - the _download_file one will check
+        # the file size and will never cache
+        response = self._request('HEAD', link, save=False, cache=cache)
+        # Get original extension
+        if 'Content-Type' in response.headers and 'text' not in response.headers['Content-Type']:
+            _, params = cgi.parse_header(response.headers['Content-Disposition'])
+        elif response.status_code == 401:
+            error = "Data protected by proprietary rights. Please check your credentials"
+            raise LoginError(error)
+        elif 'Content-Type' not in response.headers:
+            error = "Incorrect credentials"
+            raise LoginError(error)
+        response.raise_for_status()
+        return params
+
+    def _get_username_and_password(self, credentials_file):
+        if credentials_file != None:
+            self.configuration.read(credentials_file)
+            username = self.configuration.get('user', 'username')
+            password = self.configuration.get('user', 'password')
+        else:
+            username = input("Username: ")
+            password = getpass("Password: ")
+        return username, password
+
+    def _create_filename(self, filename, observation_id, suffixes):
+        if filename is not None:
+            filename = os.path.splitext(filename)[0]
+        else:
+            filename = observation_id
+        filename += "".join(suffixes)
+        return filename
+
     def _parse_filename(self, filename):
         """Parses the file's name of a product
 
@@ -572,9 +607,9 @@ def get_epic_metadata(self, *, target_name=None,
             Tables containing the metadata of the target
         """
         if not target_name and not coordinates:
-                raise Exception("Input parameters needed, "
-                                "please provide the name "
-                                "or the coordinates of the target")
+            raise Exception("Input parameters needed, "
+                            "please provide the name "
+                            "or the coordinates of the target")
 
         epic_source = {"table": "xsa.v_epic_source",
                        "column": "epic_source_equatorial_spoint"}
@@ -600,29 +635,29 @@ def get_epic_metadata(self, *, target_name=None,
         query_fmt = ("select {} from {} "
                      "where 1=contains({}, circle('ICRS', {}, {}, {}));")
         epic_source_table = self.query_xsa_tap(query_fmt.format(cols,
-                                               epic_source["table"],
-                                               epic_source["column"],
-                                               c.ra.degree,
-                                               c.dec.degree,
-                                               radius))
+                                                                epic_source["table"],
+                                                                epic_source["column"],
+                                                                c.ra.degree,
+                                                                c.dec.degree,
+                                                                radius))
         cat_4xmm_table = self.query_xsa_tap(query_fmt.format(cols,
-                                            cat_4xmm["table"],
-                                            cat_4xmm["column"],
-                                            c.ra.degree,
-                                            c.dec.degree,
-                                            radius))
+                                                             cat_4xmm["table"],
+                                                             cat_4xmm["column"],
+                                                             c.ra.degree,
+                                                             c.dec.degree,
+                                                             radius))
         stack_4xmm_table = self.query_xsa_tap(query_fmt.format(cols,
-                                              stack_4xmm["table"],
-                                              stack_4xmm["column"],
-                                              c.ra.degree,
-                                              c.dec.degree,
-                                              radius))
+                                                               stack_4xmm["table"],
+                                                               stack_4xmm["column"],
+                                                               c.ra.degree,
+                                                               c.dec.degree,
+                                                               radius))
         slew_source_table = self.query_xsa_tap(query_fmt.format(cols,
-                                               slew_source["table"],
-                                               slew_source["column"],
-                                               c.ra.degree,
-                                               c.dec.degree,
-                                               radius))
+                                                                slew_source["table"],
+                                                                slew_source["column"],
+                                                                c.ra.degree,
+                                                                c.dec.degree,
+                                                                radius))
         return epic_source_table, cat_4xmm_table, stack_4xmm_table, slew_source_table
 
     def get_epic_lightcurve(self, filename, source_number, *,

astroquery/esa/xmm_newton/tests/my_config.ini

@@ -0,0 +1,3 @@
+[user]
+username=test
+password=test

astroquery/esa/xmm_newton/tests/setup_package.py

@@ -19,6 +19,7 @@
 def get_package_data():
     paths = [os.path.join('data', '*.tar'),
              os.path.join('data', '*.xml'),
+             os.path.join('my_config.ini')
              ]  # etc, add other extensions
     # you can also enlist files individually by names
     # finally construct and return a dict for the sub module

astroquery/esa/xmm_newton/tests/test_xmm_newton.py

@@ -9,33 +9,40 @@
 
 Created on 4 Sept. 2019
 """
+from unittest.mock import patch
 
 import pytest
-
-import sys
 import tarfile
 import os
 import errno
 import shutil
-from astropy.coordinates import SkyCoord
-from astropy.utils.diff import report_diff_values
-from astroquery.utils.tap.core import TapPlus
 
 from ..core import XMMNewtonClass
 from ..tests.dummy_tap_handler import DummyXMMNewtonTapHandler
 from ..tests.dummy_handler import DummyHandler
-from fileinput import filename
-from tarfile import is_tarfile
+from astroquery.exceptions import LoginError
 
 
-class TestXMMNewton():
+class mockResponse():
+    headers = {'Date': 'Wed, 24 Nov 2021 13:43:50 GMT',
+               'Server': 'Apache/2.4.6 (Red Hat Enterprise Linux) OpenSSL/1.0.2k-fips',
+               'Content-Disposition': 'inline; filename="0560181401.tar.gz"',
+               'Content-Type': 'application/x-gzip',
+               'Content-Length': '6590874', 'Connection': 'close'}
+    status_code = 400
+
+    @staticmethod
+    def raise_for_status():
+        pass
 
+
+class TestXMMNewton():
     def get_dummy_tap_handler(self):
-        parameterst = {'query': "select top 10 * from v_public_observations",
-                       'output_file': "test2.vot",
-                       'output_format': "votable",
-                       'verbose': False}
-        dummyTapHandler = DummyXMMNewtonTapHandler("launch_job", parameterst)
+        parameters = {'query': "select top 10 * from v_public_observations",
+                      'output_file': "test2.vot",
+                      'output_format': "votable",
+                      'verbose': False}
+        dummyTapHandler = DummyXMMNewtonTapHandler("launch_job", parameters)
         return dummyTapHandler
 
     def test_query_xsa_tap(self):
@@ -235,7 +242,7 @@ def _create_tar(self, tarname, files):
                             os.makedirs(os.path.join(ob_name, ftype))
                         except OSError as exc:
                             if exc.errno == errno.EEXIST and \
-                              os.path.isdir(os.path.join(ob_name, ftype)):
+                                    os.path.isdir(os.path.join(ob_name, ftype)):
                                 pass
                             else:
                                 raise
@@ -255,7 +262,7 @@ def _create_tar_lightcurves(self, tarname, files):
                             os.makedirs(os.path.join(ob_name, ftype))
                         except OSError as exc:
                             if exc.errno == errno.EEXIST and \
-                              os.path.isdir(os.path.join(ob_name, ftype)):
+                                    os.path.isdir(os.path.join(ob_name, ftype)):
                                 pass
                             else:
                                 raise
@@ -377,7 +384,7 @@ def test_get_epic_images(self):
         xsa = XMMNewtonClass(self.get_dummy_tap_handler())
         res = xsa.get_epic_images(_tarname, band=[], instrument=[],
                                   get_detmask=True, get_exposure_map=True)
-        assert len(res) == 6     # Number of different bands
+        assert len(res) == 6  # Number of different bands
         assert len(res[1]) == 9  # Number of different inst within band 1
         assert len(res[2]) == 9  # Number of different inst within band 2
         assert len(res[3]) == 9  # Number of different inst within band 3
@@ -510,3 +517,50 @@ def test_get_epic_lightcurve_invalid_source_number(self, capsys):
                        % (_tarname, _invalid_source_number,
                           _default_instrument))
         os.remove(_tarname)
+
+    def test_create_link(self):
+        xsa = XMMNewtonClass(self.get_dummy_tap_handler())
+        link = xsa._create_link("0560181401")
+        assert link == "https://nxsa.esac.esa.int/nxsa-sl/servlet/data-action-aio?obsno=0560181401"
+
+    @patch('astroquery.query.BaseQuery._request')
+    def test_request_link(self, mock_request):
+        xsa = XMMNewtonClass(self.get_dummy_tap_handler())
+        mock_request.return_value = mockResponse
+        params = xsa._request_link("https://nxsa.esac.esa.int/nxsa-sl/servlet/data-action-aio?obsno=0560181401", None)
+        assert params == {'filename': '0560181401.tar.gz'}
+
+    @pytest.mark.xfail(raises=LoginError)
+    @patch('astroquery.query.BaseQuery._request')
+    def test_request_link_protected(self, mock_request):
+        xsa = XMMNewtonClass(self.get_dummy_tap_handler())
+        dummyclass = mockResponse
+        dummyclass.headers = {}
+        mock_request.return_value = dummyclass
+        xsa._request_link("https://nxsa.esac.esa.int/nxsa-sl/servlet/data-action-aio?obsno=0560181401", None)
+
+    @pytest.mark.xfail(raises=LoginError)
+    @patch('astroquery.query.BaseQuery._request')
+    def test_request_link_incorrect_credentials(self, mock_request):
+        xsa = XMMNewtonClass(self.get_dummy_tap_handler())
+        dummyclass = mockResponse
+        dummyclass.headers = {}
+        dummyclass.status_code = 10
+        mock_request.return_value = dummyclass
+        xsa._request_link("https://nxsa.esac.esa.int/nxsa-sl/servlet/data-action-aio?obsno=0560181401", None)
+
+    def test_get_username_and_password(self):
+        xsa = XMMNewtonClass(self.get_dummy_tap_handler())
+        username, password = xsa._get_username_and_password("astroquery/esa/xmm_newton/tests/my_config.ini")
+        assert username == "test"
+        assert password == "test"
+
+    def test_create_filename_None(self):
+        xsa = XMMNewtonClass(self.get_dummy_tap_handler())
+        filename = xsa._create_filename(None, "0560181401", ['.tar', '.gz'])
+        assert filename == "0560181401.tar.gz"
+
+    def test_create_filename_Not_None(self):
+        xsa = XMMNewtonClass(self.get_dummy_tap_handler())
+        filename = xsa._create_filename("Test", "0560181401", ['.tar', '.gz'])
+        assert filename == "Test.tar.gz"