files: # Licensed under a 3-clause BSD style license - see LICENSE.rst
  """
  ===============
  eJWST TAP tests
  ===============
  
  European Space Astronomy Centre (ESAC)
  European Space Agency (ESA)
  
  """
  import os
  import shutil
  from pathlib import Path
  from unittest.mock import MagicMock
  
  import astropy.units as u
  import numpy as np
  import pytest
  from astropy import units
  from astropy.coordinates.sky_coordinate import SkyCoord
  from astropy.table import Table
  from astropy.units import Quantity
  from astroquery.exceptions import TableParseError
  
  from astroquery.esa.jwst import JwstClass
  from astroquery.esa.jwst.tests.DummyTapHandler import DummyTapHandler
  from astroquery.ipac.ned import Ned
  from astroquery.simbad import Simbad
  from astroquery.utils.tap.conn.tests.DummyConnHandler import DummyConnHandler
  from astroquery.utils.tap.conn.tests.DummyResponse import DummyResponse
  from astroquery.utils.tap.core import TapPlus
  from astroquery.vizier import Vizier
  
  from astroquery.esa.jwst import conf
  
  
  JOB_DATA = (Path(__file__).with_name("data") / "job_1.vot").read_text()
  
  
  def data_path(filename):
      data_dir = os.path.join(os.path.dirname(__file__), 'data')
      return os.path.join(data_dir, filename)
  
  
  def get_plane_id_mock(url, *args, **kwargs):
      return ['00000000-0000-0000-879d-ae91fa2f43e2'], 3
  
  
  @pytest.fixture(autouse=True)
  def plane_id_request(request):
      mp = request.getfixturevalue("monkeypatch")
      mp.setattr(JwstClass, '_get_plane_id', get_plane_id_mock)
      return mp
  
  
  def get_associated_planes_mock(url, *args, **kwargs):
      if kwargs.get("max_cal_level") == 2:
          return "('00000000-0000-0000-879d-ae91fa2f43e2')"
      else:
          return planeids
  
  
  @pytest.fixture(autouse=True)
  def associated_planes_request(request):
      mp = request.getfixturevalue("monkeypatch")
      mp.setattr(JwstClass, '_get_associated_planes', get_associated_planes_mock)
      return mp
  
  
  def get_product_mock(params, *args, **kwargs):
      if ('file_name' in kwargs and kwargs.get('file_name') == 'file_name_id'):
          return "00000000-0000-0000-8740-65e2827c9895"
      else:
          return "jw00617023001_02102_00001_nrcb4_uncal.fits"
  
  
  @pytest.fixture(autouse=True)
  def get_product_request(request):
      if 'noautofixt' in request.keywords:
          return
      mp = request.getfixturevalue("monkeypatch")
      mp.setattr(JwstClass, '_query_get_product', get_product_mock)
      return mp
  
  
  planeids = "('00000000-0000-0000-879d-ae91fa2f43e2', '00000000-0000-0000-9852-a9fa8c63f7ef')"
  
  
  class TestTap:
  
      def test_load_tables(self):
          dummyTapHandler = DummyTapHandler()
          tap = JwstClass(tap_plus_handler=dummyTapHandler, show_messages=False)
          # default parameters
          parameters = {}
          parameters['only_names'] = False
          parameters['include_shared_tables'] = False
          parameters['verbose'] = False
          tap.load_tables()
          dummyTapHandler.check_call('load_tables', parameters)
          # test with parameters
          dummyTapHandler.reset()
          parameters = {}
          parameters['only_names'] = True
          parameters['include_shared_tables'] = True
          parameters['verbose'] = True
          tap.load_tables(only_names=True, include_shared_tables=True, verbose=True)
          dummyTapHandler.check_call('load_tables', parameters)
  
      def test_load_table(self):
          dummyTapHandler = DummyTapHandler()
          tap = JwstClass(tap_plus_handler=dummyTapHandler, show_messages=False)
          # default parameters
          parameters = {}
          parameters['table'] = 'table'
          parameters['verbose'] = False
          tap.load_table('table')
          dummyTapHandler.check_call('load_table', parameters)
          # test with parameters
          dummyTapHandler.reset()
          parameters = {}
          parameters['table'] = 'table'
          parameters['verbose'] = True
          tap.load_table('table', verbose=True)
          dummyTapHandler.check_call('load_table', parameters)
  
      def test_launch_sync_job(self):
          dummyTapHandler = DummyTapHandler()
          tap = JwstClass(tap_plus_handler=dummyTapHandler, show_messages=False)
          query = "query"
          # default parameters
          parameters = {}
          parameters['query'] = query
          parameters['name'] = None
          parameters['output_file'] = None
          parameters['output_format'] = 'votable'
          parameters['verbose'] = False
          parameters['dump_to_file'] = False
          parameters['upload_resource'] = None
          parameters['upload_table_name'] = None
          tap.launch_job(query)
          dummyTapHandler.check_call('launch_job', parameters)
          # test with parameters
          dummyTapHandler.reset()
          name = 'name'
          output_file = 'output'
          output_format = 'format'
          verbose = True
          dump_to_file = True
          upload_resource = 'upload_res'
          upload_table_name = 'upload_table'
          parameters['query'] = query
          parameters['name'] = name
          parameters['output_file'] = output_file
          parameters['output_format'] = output_format
          parameters['verbose'] = verbose
          parameters['dump_to_file'] = dump_to_file
          parameters['upload_resource'] = upload_resource
          parameters['upload_table_name'] = upload_table_name
          tap.launch_job(query,
                         name=name,
                         output_file=output_file,
                         output_format=output_format,
                         verbose=verbose,
                         dump_to_file=dump_to_file,
                         upload_resource=upload_resource,
                         upload_table_name=upload_table_name)
          dummyTapHandler.check_call('launch_job', parameters)
  
      def test_launch_async_job(self):
          dummyTapHandler = DummyTapHandler()
          tap = JwstClass(tap_plus_handler=dummyTapHandler, show_messages=False)
          query = "query"
          # default parameters
          parameters = {}
          parameters['query'] = query
          parameters['name'] = None
          parameters['output_file'] = None
          parameters['output_format'] = 'votable'
          parameters['verbose'] = False
          parameters['dump_to_file'] = False
          parameters['background'] = False
          parameters['upload_resource'] = None
          parameters['upload_table_name'] = None
          tap.launch_job(query, async_job=True)
          dummyTapHandler.check_call('launch_job_async', parameters)
          # test with parameters
          dummyTapHandler.reset()
          name = 'name'
          output_file = 'output'
          output_format = 'format'
          verbose = True
          dump_to_file = True
          background = True
          upload_resource = 'upload_res'
          upload_table_name = 'upload_table'
          parameters['query'] = query
          parameters['name'] = name
          parameters['output_file'] = output_file
          parameters['output_format'] = output_format
          parameters['verbose'] = verbose
          parameters['dump_to_file'] = dump_to_file
          parameters['background'] = background
          parameters['upload_resource'] = upload_resource
          parameters['upload_table_name'] = upload_table_name
          tap.launch_job(query,
                         name=name,
                         output_file=output_file,
                         output_format=output_format,
                         verbose=verbose,
                         dump_to_file=dump_to_file,
                         background=background,
                         upload_resource=upload_resource,
                         upload_table_name=upload_table_name,
                         async_job=True)
          dummyTapHandler.check_call('launch_job_async', parameters)
  
      def test_list_async_jobs(self):
          dummyTapHandler = DummyTapHandler()
          tap = JwstClass(tap_plus_handler=dummyTapHandler, show_messages=False)
          # default parameters
          parameters = {}
          parameters['verbose'] = False
          tap.list_async_jobs()
          dummyTapHandler.check_call('list_async_jobs', parameters)
          # test with parameters
          dummyTapHandler.reset()
          parameters['verbose'] = True
          tap.list_async_jobs(verbose=True)
          dummyTapHandler.check_call('list_async_jobs', parameters)
  
      def test_query_region(self):
          connHandler = DummyConnHandler()
          tapplus = TapPlus("http://test:1111/tap", connhandler=connHandler)
          tap = JwstClass(tap_plus_handler=tapplus, show_messages=False)
  
          # Launch response: we use default response because the
          # query contains decimals
          responseLaunchJob = DummyResponse(200)
          responseLaunchJob.set_data(method='POST', body=JOB_DATA)
          # The query contains decimals: force default response
          connHandler.set_default_response(responseLaunchJob)
          sc = SkyCoord(ra=29.0, dec=15.0, unit=(u.degree, u.degree),
                        frame='icrs')
          with pytest.raises(ValueError) as err:
              tap.query_region(sc)
          assert "Missing required argument: 'width'" in err.value.args[0]
  
          width = Quantity(12, u.deg)
          height = Quantity(10, u.deg)
  
          with pytest.raises(ValueError) as err:
              tap.query_region(sc, width=width)
          assert "Missing required argument: 'height'" in err.value.args[0]
  
          assert (isinstance(tap.query_region(sc, width=width, height=height), Table))
  
          # Test observation_id argument
          with pytest.raises(ValueError) as err:
              tap.query_region(sc, width=width, height=height, observation_id=1)
          assert "observation_id must be string" in err.value.args[0]
  
          assert (isinstance(tap.query_region(sc, width=width, height=height, observation_id="observation"), Table))
          # raise ValueError
  
          # Test cal_level argument
          with pytest.raises(ValueError) as err:
              tap.query_region(sc, width=width, height=height, cal_level='a')
          assert "cal_level must be either 'Top' or an integer" in err.value.args[0]
  
          assert (isinstance(tap.query_region(sc, width=width, height=height, cal_level='Top'), Table))
          assert (isinstance(tap.query_region(sc, width=width, height=height, cal_level=1), Table))
  
          # Test only_public
          with pytest.raises(ValueError) as err:
              tap.query_region(sc, width=width, height=height, only_public='a')
          assert "only_public must be boolean" in err.value.args[0]
  
          assert (isinstance(tap.query_region(sc, width=width, height=height, only_public=True), Table))
  
          # Test dataproduct_type argument
          with pytest.raises(ValueError) as err:
              tap.query_region(sc, width=width, height=height, prod_type=1)
          assert "prod_type must be string" in err.value.args[0]
  
          with pytest.raises(ValueError) as err:
              tap.query_region(sc, width=width, height=height, prod_type='a')
          assert "prod_type must be one of: " in err.value.args[0]
  
          assert (isinstance(tap.query_region(sc, width=width, height=height, prod_type='image'), Table))
  
          # Test instrument_name argument
          with pytest.raises(ValueError) as err:
              tap.query_region(sc, width=width, height=height, instrument_name=1)
          assert "instrument_name must be string" in err.value.args[0]
  
          assert (isinstance(tap.query_region(sc, width=width, height=height, instrument_name='NIRCAM'), Table))
  
          with pytest.raises(ValueError) as err:
              tap.query_region(sc, width=width, height=height,
                               instrument_name='a')
          assert "instrument_name must be one of: " in err.value.args[0]
  
          # Test filter_name argument
          with pytest.raises(ValueError) as err:
              tap.query_region(sc, width=width, height=height, filter_name=1)
          assert "filter_name must be string" in err.value.args[0]
  
          assert (isinstance(tap.query_region(sc, width=width, height=height, filter_name='filter'), Table))
  
          # Test proposal_id argument
          with pytest.raises(ValueError) as err:
              tap.query_region(sc, width=width, height=height, proposal_id=123)
          assert "proposal_id must be string" in err.value.args[0]
  
          assert (isinstance(tap.query_region(sc, width=width, height=height, proposal_id='123'), Table))
  
          table = tap.query_region(sc, width=width, height=height)
          assert len(table) == 3, f"Wrong job results (num rows). Expected: {3}, found {len(table)}"
          self.__check_results_column(table,
                                      'alpha',
                                      'alpha',
                                      None,
                                      np.float64)
          self.__check_results_column(table,
                                      'delta',
                                      'delta',
                                      None,
                                      np.float64)
          self.__check_results_column(table,
                                      'source_id',
                                      'source_id',
                                      None,
                                      object)
          self.__check_results_column(table,
                                      'table1_oid',
                                      'table1_oid',
                                      None,
                                      np.int32)
          # by radius
          radius = Quantity(1, u.deg)
          table = tap.query_region(sc, radius=radius)
          assert len(table) == 3, f"Wrong job results (num rows). Expected: {3}, found {len(table)}"
          self.__check_results_column(table,
                                      'alpha',
                                      'alpha',
                                      None,
                                      np.float64)
          self.__check_results_column(table,
                                      'delta',
                                      'delta',
                                      None,
                                      np.float64)
          self.__check_results_column(table,
                                      'source_id',
                                      'source_id',
                                      None,
                                      object)
          self.__check_results_column(table,
                                      'table1_oid',
                                      'table1_oid',
                                      None,
                                      np.int32)
  
      def test_query_region_async(self):
          connHandler = DummyConnHandler()
          tapplus = TapPlus("http://test:1111/tap", connhandler=connHandler)
          tap = JwstClass(tap_plus_handler=tapplus, show_messages=False)
          jobid = '12345'
          # Launch response
          responseLaunchJob = DummyResponse(303)
          # list of list (httplib implementation for headers in response)
          launchResponseHeaders = [['location', 'http://test:1111/tap/async/' + jobid]]
          responseLaunchJob.set_data(method='POST', headers=launchResponseHeaders)
          connHandler.set_default_response(responseLaunchJob)
          # Phase response
          responsePhase = DummyResponse(200)
          responsePhase.set_data(method='GET', body="COMPLETED")
          req = "async/" + jobid + "/phase"
          connHandler.set_response(req, responsePhase)
          # Results response
          responseResultsJob = DummyResponse(200)
          responseResultsJob.set_data(method='GET', body=JOB_DATA)
          req = "async/" + jobid + "/results/result"
          connHandler.set_response(req, responseResultsJob)
          sc = SkyCoord(ra=29.0, dec=15.0, unit=(u.degree, u.degree),
                        frame='icrs')
          width = Quantity(12, u.deg)
          height = Quantity(10, u.deg)
          table = tap.query_region(sc, width=width, height=height, async_job=True)
          assert len(table) == 3, f"Wrong job results (num rows). Expected: {3}, found {len(table)}"
          self.__check_results_column(table,
                                      'alpha',
                                      'alpha',
                                      None,
                                      np.float64)
          self.__check_results_column(table,
                                      'delta',
                                      'delta',
                                      None,
                                      np.float64)
          self.__check_results_column(table,
                                      'source_id',
                                      'source_id',
                                      None,
                                      object)
          self.__check_results_column(table,
                                      'table1_oid',
                                      'table1_oid',
                                      None,
                                      np.int32)
          # by radius
          radius = Quantity(1, u.deg)
          table = tap.query_region(sc, radius=radius, async_job=True)
          assert len(table) == 3, f"Wrong job results (num rows). Expected: {3}, found {len(table)}"
          self.__check_results_column(table,
                                      'alpha',
                                      'alpha',
                                      None,
                                      np.float64)
          self.__check_results_column(table,
                                      'delta',
                                      'delta',
                                      None,
                                      np.float64)
          self.__check_results_column(table,
                                      'source_id',
                                      'source_id',
                                      None,
                                      object)
          self.__check_results_column(table,
                                      'table1_oid',
                                      'table1_oid',
                                      None,
                                      np.int32)
  
      def test_cone_search_sync(self):
          connHandler = DummyConnHandler()
          tapplus = TapPlus("http://test:1111/tap", connhandler=connHandler)
          tap = JwstClass(tap_plus_handler=tapplus, show_messages=False)
          # Launch response: we use default response because the
          # query contains decimals
          responseLaunchJob = DummyResponse(200)
          responseLaunchJob.set_data(method='POST', body=JOB_DATA)
          ra = 19.0
          dec = 20.0
          sc = SkyCoord(ra=ra, dec=dec, unit=(u.degree, u.degree), frame='icrs')
          radius = Quantity(1.0, u.deg)
          connHandler.set_default_response(responseLaunchJob)
          job = tap.cone_search(sc, radius)
          assert job is not None, "Expected a valid job"
          assert job.async_ is False, "Expected a synchronous job"
          assert job.get_phase() == 'COMPLETED', f"Wrong job phase. Expected: {'COMPLETED'}, found {job.get_phase()}"
          assert job.failed is False, "Wrong job status (set Failed = True)"
          # results
          results = job.get_results()
          assert len(results) == 3, f"Wrong job results (num rows). Expected: {3}, found {len(results)}"
          self.__check_results_column(results,
                                      'alpha',
                                      'alpha',
                                      None,
                                      np.float64)
          self.__check_results_column(results,
                                      'delta',
                                      'delta',
                                      None,
                                      np.float64)
          self.__check_results_column(results,
                                      'source_id',
                                      'source_id',
                                      None,
                                      object)
          self.__check_results_column(results,
                                      'table1_oid',
                                      'table1_oid',
                                      None,
                                      np.int32)
  
          # Test observation_id argument
          with pytest.raises(ValueError) as err:
              tap.cone_search(sc, radius, observation_id=1)
          assert "observation_id must be string" in err.value.args[0]
  
          # Test cal_level argument
          with pytest.raises(ValueError) as err:
              tap.cone_search(sc, radius, cal_level='a')
          assert "cal_level must be either 'Top' or an integer" in err.value.args[0]
  
          # Test only_public
          with pytest.raises(ValueError) as err:
              tap.cone_search(sc, radius, only_public='a')
          assert "only_public must be boolean" in err.value.args[0]
  
          # Test dataproduct_type argument
          with pytest.raises(ValueError) as err:
              tap.cone_search(sc, radius, prod_type=1)
          assert "prod_type must be string" in err.value.args[0]
  
          with pytest.raises(ValueError) as err:
              tap.cone_search(sc, radius, prod_type='a')
          assert "prod_type must be one of: " in err.value.args[0]
  
          # Test instrument_name argument
          with pytest.raises(ValueError) as err:
              tap.cone_search(sc, radius, instrument_name=1)
          assert "instrument_name must be string" in err.value.args[0]
  
          with pytest.raises(ValueError) as err:
              tap.cone_search(sc, radius, instrument_name='a')
          assert "instrument_name must be one of: " in err.value.args[0]
  
          # Test filter_name argument
          with pytest.raises(ValueError) as err:
              tap.cone_search(sc, radius, filter_name=1)
          assert "filter_name must be string" in err.value.args[0]
  
          # Test proposal_id argument
          with pytest.raises(ValueError) as err:
              tap.cone_search(sc, radius, proposal_id=123)
          assert "proposal_id must be string" in err.value.args[0]
  
      def test_cone_search_async(self):
          connHandler = DummyConnHandler()
          tapplus = TapPlus("http://test:1111/tap", connhandler=connHandler)
          tap = JwstClass(tap_plus_handler=tapplus, show_messages=False)
          jobid = '12345'
          # Launch response
          responseLaunchJob = DummyResponse(303)
          # list of list (httplib implementation for headers in response)
          launchResponseHeaders = [['location', 'http://test:1111/tap/async/' + jobid]]
          responseLaunchJob.set_data(method='POST', headers=launchResponseHeaders)
          ra = 19
          dec = 20
          sc = SkyCoord(ra=ra, dec=dec, unit=(u.degree, u.degree), frame='icrs')
          radius = Quantity(1.0, u.deg)
          connHandler.set_default_response(responseLaunchJob)
          # Phase response
          responsePhase = DummyResponse(200)
          responsePhase.set_data(method='GET', body="COMPLETED")
          req = "async/" + jobid + "/phase"
          connHandler.set_response(req, responsePhase)
          # Results response
          responseResultsJob = DummyResponse(200)
          responseResultsJob.set_data(method='GET', body=JOB_DATA)
          req = "async/" + jobid + "/results/result"
          connHandler.set_response(req, responseResultsJob)
          job = tap.cone_search(sc, radius, async_job=True)
          assert job is not None, "Expected a valid job"
          assert job.async_ is True, "Expected an asynchronous job"
          assert job.get_phase() == 'COMPLETED', f"Wrong job phase. Expected: {'COMPLETED'}, found {job.get_phase()}"
          assert job.failed is False, "Wrong job status (set Failed = True)"
          # results
          results = job.get_results()
          assert len(results) == 3, "Wrong job results (num rows). Expected: {3}, found {len(results)}"
          self.__check_results_column(results,
                                      'alpha',
                                      'alpha',
                                      None,
                                      np.float64)
          self.__check_results_column(results,
                                      'delta',
                                      'delta',
                                      None,
                                      np.float64)
          self.__check_results_column(results,
                                      'source_id',
                                      'source_id',
                                      None,
                                      object)
          self.__check_results_column(results,
                                      'table1_oid',
                                      'table1_oid',
                                      None,
                                      np.int32)
  
      def test_get_product_by_artifactid(self):
          dummyTapHandler = DummyTapHandler()
          jwst = JwstClass(tap_plus_handler=dummyTapHandler, data_handler=dummyTapHandler, show_messages=False)
          # default parameters
          with pytest.raises(ValueError) as err:
              jwst.get_product()
          assert "Missing required argument: 'artifact_id' or 'file_name'" in err.value.args[0]
  
          # test with parameters
          dummyTapHandler.reset()
  
          parameters = {}
          parameters['output_file'] = 'jw00617023001_02102_00001_nrcb4_uncal.fits'
          parameters['verbose'] = False
  
          param_dict = {}
          param_dict['RETRIEVAL_TYPE'] = 'PRODUCT'
          param_dict['TAPCLIENT'] = 'ASTROQUERY'
          param_dict['ARTIFACTID'] = '00000000-0000-0000-8740-65e2827c9895'
          parameters['params_dict'] = param_dict
  
          jwst.get_product(artifact_id='00000000-0000-0000-8740-65e2827c9895')
          dummyTapHandler.check_call('load_data', parameters)
  
      def test_get_product_by_filename(self):
          dummyTapHandler = DummyTapHandler()
          jwst = JwstClass(tap_plus_handler=dummyTapHandler, data_handler=dummyTapHandler, show_messages=False)
          # default parameters
          with pytest.raises(ValueError) as err:
              jwst.get_product()
          assert "Missing required argument: 'artifact_id' or 'file_name'" in err.value.args[0]
  
          # test with parameters
          dummyTapHandler.reset()
  
          parameters = {}
          parameters['output_file'] = 'file_name_id'
          parameters['verbose'] = False
  
          param_dict = {}
          param_dict['RETRIEVAL_TYPE'] = 'PRODUCT'
          param_dict['TAPCLIENT'] = 'ASTROQUERY'
          param_dict['ARTIFACTID'] = '00000000-0000-0000-8740-65e2827c9895'
          parameters['params_dict'] = param_dict
  
          jwst.get_product(file_name='file_name_id')
          dummyTapHandler.check_call('load_data', parameters)
  
      def test_get_products_list(self):
          dummyTapHandler = DummyTapHandler()
          jwst = JwstClass(tap_plus_handler=dummyTapHandler, data_handler=dummyTapHandler, show_messages=False)
          # default parameters
          with pytest.raises(ValueError) as err:
              jwst.get_product_list()
          assert "Missing required argument: 'observation_id'" in err.value.args[0]
  
          # test with parameters
          dummyTapHandler.reset()
  
          observation_id = "jw00777011001_02104_00001_nrcblong"
  
          query = (f"select distinct a.uri, a.artifactid, a.filename, "
                   f"a.contenttype, a.producttype, p.calibrationlevel, p.public "
                   f"FROM {conf.JWST_PLANE_TABLE} p JOIN {conf.JWST_ARTIFACT_TABLE} "
                   f"a ON (p.planeid=a.planeid) WHERE a.planeid "
                   f"IN {planeids};")
  
          parameters = {}
          parameters['query'] = query
          parameters['name'] = None
          parameters['output_file'] = None
          parameters['output_format'] = 'votable'
          parameters['verbose'] = False
          parameters['dump_to_file'] = False
          parameters['upload_resource'] = None
          parameters['upload_table_name'] = None
  
          jwst.get_product_list(observation_id=observation_id)
          dummyTapHandler.check_call('launch_job', parameters)
  
      def test_get_obs_products(self):
          dummyTapHandler = DummyTapHandler()
          jwst = JwstClass(tap_plus_handler=dummyTapHandler, data_handler=dummyTapHandler, show_messages=False)
          # default parameters
          with pytest.raises(ValueError) as err:
              jwst.get_obs_products()
          assert "Missing required argument: 'observation_id'" in err.value.args[0]
  
          # test with parameters
          dummyTapHandler.reset()
  
          output_file_full_path_dir = os.getcwd() + os.sep + "temp_test_jwsttap_get_obs_products_1"
          try:
              os.makedirs(output_file_full_path_dir, exist_ok=True)
          except OSError as err:
              print(f"Creation of the directory {output_file_full_path_dir} failed: {err.strerror}")
              raise err
  
          observation_id = 'jw00777011001_02104_00001_nrcblong'
  
          parameters = {}
          parameters['verbose'] = False
  
          param_dict = {}
          param_dict['RETRIEVAL_TYPE'] = 'OBSERVATION'
          param_dict['TAPCLIENT'] = 'ASTROQUERY'
          param_dict['planeid'] = planeids
          param_dict['calibrationlevel'] = 'ALL'
          parameters['params_dict'] = param_dict
  
          # Test single product tar
          file = data_path('single_product_retrieval.tar')
          output_file_full_path = output_file_full_path_dir + os.sep + os.path.basename(file)
          shutil.copy(file, output_file_full_path)
          parameters['output_file'] = output_file_full_path
  
          expected_files = []
          extracted_file_1 = output_file_full_path_dir + os.sep + 'single_product_retrieval_1.fits'
          expected_files.append(extracted_file_1)
          try:
              files_returned = (jwst.get_obs_products(
                                observation_id=observation_id, cal_level='ALL',
                                output_file=output_file_full_path))
              dummyTapHandler.check_call('load_data', parameters)
              self.__check_extracted_files(files_expected=expected_files,
                                           files_returned=files_returned)
          finally:
              shutil.rmtree(output_file_full_path_dir)
  
          # Test single file
          output_file_full_path_dir = os.getcwd() + os.sep +\
              "temp_test_jwsttap_get_obs_products_2"
          try:
              os.makedirs(output_file_full_path_dir, exist_ok=True)
          except OSError as err:
              print(f"Creation of the directory {output_file_full_path_dir} failed: {err.strerror}")
              raise err
  
          file = data_path('single_product_retrieval_1.fits')
          output_file_full_path = output_file_full_path_dir + os.sep +\
              os.path.basename(file)
          shutil.copy(file, output_file_full_path)
  
          parameters['output_file'] = output_file_full_path
  
          expected_files = []
          expected_files.append(output_file_full_path)
  
          try:
              files_returned = (jwst.get_obs_products(
                                observation_id=observation_id,
                                output_file=output_file_full_path))
              dummyTapHandler.check_call('load_data', parameters)
              self.__check_extracted_files(files_expected=expected_files,
                                           files_returned=files_returned)
          finally:
              # self.__remove_folder_contents(folder=output_file_full_path_dir)
              shutil.rmtree(output_file_full_path_dir)
  
          # Test single file zip
          output_file_full_path_dir = os.getcwd() + os.sep + "temp_test_jwsttap_get_obs_products_3"
          try:
              os.makedirs(output_file_full_path_dir, exist_ok=True)
          except OSError as err:
              print(f"Creation of the directory {output_file_full_path_dir} failed: {err.strerror}")
              raise err
  
          file = data_path('single_product_retrieval_3.fits.zip')
          output_file_full_path = output_file_full_path_dir + os.sep +\
              os.path.basename(file)
          shutil.copy(file, output_file_full_path)
  
          parameters['output_file'] = output_file_full_path
  
          expected_files = []
          extracted_file_1 = output_file_full_path_dir + os.sep + 'single_product_retrieval.fits'
          expected_files.append(extracted_file_1)
  
          try:
              files_returned = (jwst.get_obs_products(
                                observation_id=observation_id,
                                output_file=output_file_full_path))
              dummyTapHandler.check_call('load_data', parameters)
              self.__check_extracted_files(files_expected=expected_files,
                                           files_returned=files_returned)
          finally:
              # self.__remove_folder_contents(folder=output_file_full_path_dir)
              shutil.rmtree(output_file_full_path_dir)
  
          # Test single file gzip
          output_file_full_path_dir = (os.getcwd() + os.sep + "temp_test_jwsttap_get_obs_products_4")
          try:
              os.makedirs(output_file_full_path_dir, exist_ok=True)
          except OSError as err:
              print(f"Creation of the directory {output_file_full_path_dir} failed: {err.strerror}")
              raise err
  
          file = data_path('single_product_retrieval_2.fits.gz')
          output_file_full_path = output_file_full_path_dir + os.sep + os.path.basename(file)
          shutil.copy(file, output_file_full_path)
  
          parameters['output_file'] = output_file_full_path
  
          expected_files = []
          extracted_file_1 = output_file_full_path_dir + os.sep + 'single_product_retrieval_2.fits.gz'
          expected_files.append(extracted_file_1)
  
          try:
              files_returned = (jwst.get_obs_products(
                                observation_id=observation_id,
                                output_file=output_file_full_path))
              dummyTapHandler.check_call('load_data', parameters)
              self.__check_extracted_files(files_expected=expected_files,
                                           files_returned=files_returned)
          finally:
              # self.__remove_folder_contents(folder=output_file_full_path_dir)
              shutil.rmtree(output_file_full_path_dir)
  
          # Test tar with 3 files, a normal one, a gzip one and a zip one
          output_file_full_path_dir = (os.getcwd() + os.sep + "temp_test_jwsttap_get_obs_products_5")
          try:
              os.makedirs(output_file_full_path_dir, exist_ok=True)
          except OSError as err:
              print(f"Creation of the directory {output_file_full_path_dir} failed: {err.strerror}")
              raise err
  
          file = data_path('three_products_retrieval.tar')
          output_file_full_path = output_file_full_path_dir + os.sep + os.path.basename(file)
          shutil.copy(file, output_file_full_path)
  
          parameters['output_file'] = output_file_full_path
  
          expected_files = []
          extracted_file_1 = output_file_full_path_dir + os.sep + 'single_product_retrieval_1.fits'
          expected_files.append(extracted_file_1)
          extracted_file_2 = output_file_full_path_dir + os.sep + 'single_product_retrieval_2.fits.gz'
          expected_files.append(extracted_file_2)
          extracted_file_3 = output_file_full_path_dir + os.sep + 'single_product_retrieval_3.fits.zip'
          expected_files.append(extracted_file_3)
  
          try:
              files_returned = (jwst.get_obs_products(
                                observation_id=observation_id,
                                output_file=output_file_full_path))
              dummyTapHandler.check_call('load_data', parameters)
              self.__check_extracted_files(files_expected=expected_files,
                                           files_returned=files_returned)
          finally:
              # self.__remove_folder_contents(folder=output_file_full_path_dir)
              shutil.rmtree(output_file_full_path_dir)
  
      def test_gunzip_file(self):
          output_file_full_path_dir = (os.getcwd() + os.sep + "temp_test_jwsttap_gunzip")
          try:
              os.makedirs(output_file_full_path_dir, exist_ok=True)
          except OSError as err:
              print(f"Creation of the directory {output_file_full_path_dir} failed: {err.strerror}")
              raise err
  
          file = data_path('single_product_retrieval_2.fits.gz')
          output_file_full_path = output_file_full_path_dir + os.sep + os.path.basename(file)
          shutil.copy(file, output_file_full_path)
  
          expected_files = []
          extracted_file_1 = output_file_full_path_dir + os.sep + "single_product_retrieval_2.fits"
          expected_files.append(extracted_file_1)
  
          try:
              extracted_file = (JwstClass.gzip_uncompress_and_rename_single_file(
                                output_file_full_path))
              if extracted_file != extracted_file_1:
                  raise ValueError(f"Extracted file not found: {extracted_file_1}")
          finally:
              # self.__remove_folder_contents(folder=output_file_full_path_dir)
              shutil.rmtree(output_file_full_path_dir)
  
      def __check_results_column(self, results, columnName, description, unit,
                                 dataType):
          c = results[columnName]
          assert c.description == description, \
              f"Wrong description for results column '{columnName}'. Expected: '{description}', "\
              f"found '{c.description}'"
          assert c.unit == unit, \
              f"Wrong unit for results column '{columnName}'. Expected: '{unit}', found '{c.unit}'"
          assert c.dtype == dataType, \
              f"Wrong dataType for results column '{columnName}'. Expected: '{dataType}', found '{c.dtype}'"
  
      def __remove_folder_contents(self, folder):
          for root, dirs, files in os.walk(folder):
              for f in files:
                  os.unlink(os.path.join(root, f))
              for d in dirs:
                  shutil.rmtree(os.path.join(root, d))
  
      def __check_extracted_files(self, files_expected, files_returned):
          if len(files_expected) != len(files_returned):
              raise ValueError(f"Expected files size error. "
                               f"Found {len(files_returned)}, "
                               f"expected {len(files_expected)}")
          for f in files_expected:
              if not os.path.exists(f):
                  raise ValueError(f"Not found extracted file: "
                                   f"{f}")
              if f not in files_returned:
                  raise ValueError(f"Not found expected file: {f}")
  
      def test_query_target_error(self):
          jwst = JwstClass(show_messages=False)
          simbad = Simbad()
          ned = Ned()
          vizier = Vizier()
          # Testing default parameters
          with pytest.raises((ValueError, TableParseError)) as err:
              jwst.query_target(target_name="M1", target_resolver="")
          assert "This target resolver is not allowed" in err.value.args[0]
          with pytest.raises((ValueError, TableParseError)) as err:
              jwst.query_target("TEST")
          assert ("This target name cannot be determined with this resolver: ALL" in err.value.args[0]
                  or 'Failed to parse' in err.value.args[0])
          with pytest.raises((ValueError, TableParseError)) as err:
              jwst.query_target(target_name="M1", target_resolver="ALL")
          assert err.value.args[0] in ["This target name cannot be determined "
                                       "with this resolver: ALL", "Missing "
                                       "required argument: 'width'"]
  
          # Testing no valid coordinates from resolvers
          simbad_file = data_path('test_query_by_target_name_simbad_ned_error.vot')
          simbad_table = Table.read(simbad_file)
          simbad.query_object = MagicMock(return_value=simbad_table)
          ned_file = data_path('test_query_by_target_name_simbad_ned_error.vot')
          ned_table = Table.read(ned_file)
          ned.query_object = MagicMock(return_value=ned_table)
          vizier_file = data_path('test_query_by_target_name_vizier_error.vot')
          vizier_table = Table.read(vizier_file)
          vizier.query_object = MagicMock(return_value=vizier_table)
  
          # coordinate_error = 'coordinate must be either a string or astropy.coordinates'
          with pytest.raises((ValueError, TableParseError)) as err:
              jwst.query_target(target_name="test", target_resolver="SIMBAD",
                                radius=units.Quantity(5, units.deg))
          assert ('This target name cannot be determined with this resolver: SIMBAD' in err.value.args[0]
                  or 'Failed to parse' in err.value.args[0])
  
          with pytest.raises((ValueError, TableParseError)) as err:
              jwst.query_target(target_name="test", target_resolver="NED",
                                radius=units.Quantity(5, units.deg))
          assert ('This target name cannot be determined with this resolver: NED' in err.value.args[0]
                  or 'Failed to parse' in err.value.args[0])
  
          with pytest.raises((ValueError, TableParseError)) as err:
              jwst.query_target(target_name="test", target_resolver="VIZIER",
                                radius=units.Quantity(5, units.deg))
          assert ('This target name cannot be determined with this resolver: VIZIER' in err.value.args[0]
                  or 'Failed to parse' in err.value.args[0])
  
      def test_remove_jobs(self):
          dummyTapHandler = DummyTapHandler()
          tap = JwstClass(tap_plus_handler=dummyTapHandler, show_messages=False)
          job_list = ['dummyJob']
          parameters = {}
          parameters['jobs_list'] = job_list
          parameters['verbose'] = False
          tap.remove_jobs(job_list)
          dummyTapHandler.check_call('remove_jobs', parameters)
  
      def test_save_results(self):
          dummyTapHandler = DummyTapHandler()
          tap = JwstClass(tap_plus_handler=dummyTapHandler, show_messages=False)
          job = 'dummyJob'
          parameters = {}
          parameters['job'] = job
          parameters['verbose'] = False
          tap.save_results(job)
          dummyTapHandler.check_call('save_results', parameters)
  
      def test_login(self):
          dummyTapHandler = DummyTapHandler()
          tap = JwstClass(tap_plus_handler=dummyTapHandler, show_messages=False)
          parameters = {}
          parameters['user'] = 'test_user'
          parameters['password'] = 'test_password'
          parameters['credentials_file'] = None
          parameters['verbose'] = False
          tap.login(user='test_user', password='test_password')
          dummyTapHandler.check_call('login', parameters)
  
      def test_logout(self):
          dummyTapHandler = DummyTapHandler()
          tap = JwstClass(tap_plus_handler=dummyTapHandler, show_messages=False)
          parameters = {}
          parameters['verbose'] = False
          tap.logout()
          dummyTapHandler.check_call('logout', parameters)
  
      @pytest.mark.noautofixt
      def test_query_get_product(self):
          dummyTapHandler = DummyTapHandler()
          tap = JwstClass(tap_plus_handler=dummyTapHandler, show_messages=False)
          file = 'test_file'
          parameters = {}
          parameters['query'] = f"select * from jwst.artifact a where a.filename = '{file}'"
          parameters['name'] = None
          parameters['output_file'] = None
          parameters['output_format'] = 'votable'
          parameters['verbose'] = False
          parameters['dump_to_file'] = False
          parameters['upload_resource'] = None
          parameters['upload_table_name'] = None
          tap._query_get_product(file_name=file)
          dummyTapHandler.check_call('launch_job', parameters)
  
          artifact = 'test_artifact'
          parameters['query'] = f"select * from jwst.artifact a where a.artifactid = '{artifact}'"
          tap._query_get_product(artifact_id=artifact)
          dummyTapHandler.check_call('launch_job', parameters)
  
      def test_get_related_observations(self):
          dummyTapHandler = DummyTapHandler()
          tap = JwstClass(tap_plus_handler=dummyTapHandler, show_messages=False)
          obs = 'dummyObs'
          tap.get_related_observations(observation_id=obs)
          parameters = {}
          parameters['query'] = f"select * from jwst.main m where m.members like '%{obs}%'"
          parameters['name'] = None
          parameters['output_file'] = None
          parameters['output_format'] = 'votable'
          parameters['verbose'] = False
          parameters['dump_to_file'] = False
          parameters['upload_resource'] = None
          parameters['upload_table_name'] = None
          dummyTapHandler.check_call('launch_job', parameters)
  
files: astroquery/esa/jwst/tests/test_jwsttap.py
files: # Licensed under a 3-clause BSD style license - see LICENSE.rst
  """
  ===============
  eJWST TAP tests
  ===============
  
  European Space Astronomy Centre (ESAC)
  European Space Agency (ESA)
  
  """
  import os
  import shutil
  from unittest.mock import MagicMock
  
  import astropy.units as u
  import numpy as np
  import pytest
  from astropy import units
  from astropy.coordinates.sky_coordinate import SkyCoord
  from astropy.table import Table
  from astropy.units import Quantity
  from astroquery.exceptions import TableParseError
  
  from astroquery.esa.jwst import JwstClass
  from astroquery.esa.jwst.tests.DummyTapHandler import DummyTapHandler
  from astroquery.ipac.ned import Ned
  from astroquery.simbad import Simbad
  from astroquery.utils.tap.conn.tests.DummyConnHandler import DummyConnHandler
  from astroquery.utils.tap.conn.tests.DummyResponse import DummyResponse
  from astroquery.utils.tap.core import TapPlus
  from astroquery.utils.tap.xmlparser import utils
  from astroquery.vizier import Vizier
  
  from astroquery.esa.jwst import conf
  
  
  def data_path(filename):
      data_dir = os.path.join(os.path.dirname(__file__), 'data')
      return os.path.join(data_dir, filename)
  
  
  def get_plane_id_mock(url, *args, **kwargs):
      return ['00000000-0000-0000-879d-ae91fa2f43e2'], 3
  
  
  @pytest.fixture(autouse=True)
  def plane_id_request(request):
      mp = request.getfixturevalue("monkeypatch")
      mp.setattr(JwstClass, '_get_plane_id', get_plane_id_mock)
      return mp
  
  
  def get_associated_planes_mock(url, *args, **kwargs):
      if kwargs.get("max_cal_level") == 2:
          return "('00000000-0000-0000-879d-ae91fa2f43e2')"
      else:
          return planeids
  
  
  @pytest.fixture(autouse=True)
  def associated_planes_request(request):
      mp = request.getfixturevalue("monkeypatch")
      mp.setattr(JwstClass, '_get_associated_planes', get_associated_planes_mock)
      return mp
  
  
  def get_product_mock(params, *args, **kwargs):
      if ('file_name' in kwargs and kwargs.get('file_name') == 'file_name_id'):
          return "00000000-0000-0000-8740-65e2827c9895"
      else:
          return "jw00617023001_02102_00001_nrcb4_uncal.fits"
  
  
  @pytest.fixture(autouse=True)
  def get_product_request(request):
      if 'noautofixt' in request.keywords:
          return
      mp = request.getfixturevalue("monkeypatch")
      mp.setattr(JwstClass, '_query_get_product', get_product_mock)
      return mp
  
  
  planeids = "('00000000-0000-0000-879d-ae91fa2f43e2', '00000000-0000-0000-9852-a9fa8c63f7ef')"
  
  
  class TestTap:
  
      def test_load_tables(self):
          dummyTapHandler = DummyTapHandler()
          tap = JwstClass(tap_plus_handler=dummyTapHandler, show_messages=False)
          # default parameters
          parameters = {}
          parameters['only_names'] = False
          parameters['include_shared_tables'] = False
          parameters['verbose'] = False
          tap.load_tables()
          dummyTapHandler.check_call('load_tables', parameters)
          # test with parameters
          dummyTapHandler.reset()
          parameters = {}
          parameters['only_names'] = True
          parameters['include_shared_tables'] = True
          parameters['verbose'] = True
          tap.load_tables(only_names=True, include_shared_tables=True, verbose=True)
          dummyTapHandler.check_call('load_tables', parameters)
  
      def test_load_table(self):
          dummyTapHandler = DummyTapHandler()
          tap = JwstClass(tap_plus_handler=dummyTapHandler, show_messages=False)
          # default parameters
          parameters = {}
          parameters['table'] = 'table'
          parameters['verbose'] = False
          tap.load_table('table')
          dummyTapHandler.check_call('load_table', parameters)
          # test with parameters
          dummyTapHandler.reset()
          parameters = {}
          parameters['table'] = 'table'
          parameters['verbose'] = True
          tap.load_table('table', verbose=True)
          dummyTapHandler.check_call('load_table', parameters)
  
      def test_launch_sync_job(self):
          dummyTapHandler = DummyTapHandler()
          tap = JwstClass(tap_plus_handler=dummyTapHandler, show_messages=False)
          query = "query"
          # default parameters
          parameters = {}
          parameters['query'] = query
          parameters['name'] = None
          parameters['output_file'] = None
          parameters['output_format'] = 'votable'
          parameters['verbose'] = False
          parameters['dump_to_file'] = False
          parameters['upload_resource'] = None
          parameters['upload_table_name'] = None
          tap.launch_job(query)
          dummyTapHandler.check_call('launch_job', parameters)
          # test with parameters
          dummyTapHandler.reset()
          name = 'name'
          output_file = 'output'
          output_format = 'format'
          verbose = True
          dump_to_file = True
          upload_resource = 'upload_res'
          upload_table_name = 'upload_table'
          parameters['query'] = query
          parameters['name'] = name
          parameters['output_file'] = output_file
          parameters['output_format'] = output_format
          parameters['verbose'] = verbose
          parameters['dump_to_file'] = dump_to_file
          parameters['upload_resource'] = upload_resource
          parameters['upload_table_name'] = upload_table_name
          tap.launch_job(query,
                         name=name,
                         output_file=output_file,
                         output_format=output_format,
                         verbose=verbose,
                         dump_to_file=dump_to_file,
                         upload_resource=upload_resource,
                         upload_table_name=upload_table_name)
          dummyTapHandler.check_call('launch_job', parameters)
  
      def test_launch_async_job(self):
          dummyTapHandler = DummyTapHandler()
          tap = JwstClass(tap_plus_handler=dummyTapHandler, show_messages=False)
          query = "query"
          # default parameters
          parameters = {}
          parameters['query'] = query
          parameters['name'] = None
          parameters['output_file'] = None
          parameters['output_format'] = 'votable'
          parameters['verbose'] = False
          parameters['dump_to_file'] = False
          parameters['background'] = False
          parameters['upload_resource'] = None
          parameters['upload_table_name'] = None
          tap.launch_job(query, async_job=True)
          dummyTapHandler.check_call('launch_job_async', parameters)
          # test with parameters
          dummyTapHandler.reset()
          name = 'name'
          output_file = 'output'
          output_format = 'format'
          verbose = True
          dump_to_file = True
          background = True
          upload_resource = 'upload_res'
          upload_table_name = 'upload_table'
          parameters['query'] = query
          parameters['name'] = name
          parameters['output_file'] = output_file
          parameters['output_format'] = output_format
          parameters['verbose'] = verbose
          parameters['dump_to_file'] = dump_to_file
          parameters['background'] = background
          parameters['upload_resource'] = upload_resource
          parameters['upload_table_name'] = upload_table_name
          tap.launch_job(query,
                         name=name,
                         output_file=output_file,
                         output_format=output_format,
                         verbose=verbose,
                         dump_to_file=dump_to_file,
                         background=background,
                         upload_resource=upload_resource,
                         upload_table_name=upload_table_name,
                         async_job=True)
          dummyTapHandler.check_call('launch_job_async', parameters)
  
      def test_list_async_jobs(self):
          dummyTapHandler = DummyTapHandler()
          tap = JwstClass(tap_plus_handler=dummyTapHandler, show_messages=False)
          # default parameters
          parameters = {}
          parameters['verbose'] = False
          tap.list_async_jobs()
          dummyTapHandler.check_call('list_async_jobs', parameters)
          # test with parameters
          dummyTapHandler.reset()
          parameters['verbose'] = True
          tap.list_async_jobs(verbose=True)
          dummyTapHandler.check_call('list_async_jobs', parameters)
  
      def test_query_region(self):
          connHandler = DummyConnHandler()
          tapplus = TapPlus("http://test:1111/tap", connhandler=connHandler)
          tap = JwstClass(tap_plus_handler=tapplus, show_messages=False)
  
          # Launch response: we use default response because the
          # query contains decimals
          responseLaunchJob = DummyResponse(200)
          jobDataFile = data_path('job_1.vot')
          jobData = utils.read_file_content(jobDataFile)
          responseLaunchJob.set_data(method='POST', body=jobData)
          # The query contains decimals: force default response
          connHandler.set_default_response(responseLaunchJob)
          sc = SkyCoord(ra=29.0, dec=15.0, unit=(u.degree, u.degree),
                        frame='icrs')
          with pytest.raises(ValueError) as err:
              tap.query_region(sc)
          assert "Missing required argument: 'width'" in err.value.args[0]
  
          width = Quantity(12, u.deg)
          height = Quantity(10, u.deg)
  
          with pytest.raises(ValueError) as err:
              tap.query_region(sc, width=width)
          assert "Missing required argument: 'height'" in err.value.args[0]
  
          assert (isinstance(tap.query_region(sc, width=width, height=height), Table))
  
          # Test observation_id argument
          with pytest.raises(ValueError) as err:
              tap.query_region(sc, width=width, height=height, observation_id=1)
          assert "observation_id must be string" in err.value.args[0]
  
          assert (isinstance(tap.query_region(sc, width=width, height=height, observation_id="observation"), Table))
          # raise ValueError
  
          # Test cal_level argument
          with pytest.raises(ValueError) as err:
              tap.query_region(sc, width=width, height=height, cal_level='a')
          assert "cal_level must be either 'Top' or an integer" in err.value.args[0]
  
          assert (isinstance(tap.query_region(sc, width=width, height=height, cal_level='Top'), Table))
          assert (isinstance(tap.query_region(sc, width=width, height=height, cal_level=1), Table))
  
          # Test only_public
          with pytest.raises(ValueError) as err:
              tap.query_region(sc, width=width, height=height, only_public='a')
          assert "only_public must be boolean" in err.value.args[0]
  
          assert (isinstance(tap.query_region(sc, width=width, height=height, only_public=True), Table))
  
          # Test dataproduct_type argument
          with pytest.raises(ValueError) as err:
              tap.query_region(sc, width=width, height=height, prod_type=1)
          assert "prod_type must be string" in err.value.args[0]
  
          with pytest.raises(ValueError) as err:
              tap.query_region(sc, width=width, height=height, prod_type='a')
          assert "prod_type must be one of: " in err.value.args[0]
  
          assert (isinstance(tap.query_region(sc, width=width, height=height, prod_type='image'), Table))
  
          # Test instrument_name argument
          with pytest.raises(ValueError) as err:
              tap.query_region(sc, width=width, height=height, instrument_name=1)
          assert "instrument_name must be string" in err.value.args[0]
  
          assert (isinstance(tap.query_region(sc, width=width, height=height, instrument_name='NIRCAM'), Table))
  
          with pytest.raises(ValueError) as err:
              tap.query_region(sc, width=width, height=height,
                               instrument_name='a')
          assert "instrument_name must be one of: " in err.value.args[0]
  
          # Test filter_name argument
          with pytest.raises(ValueError) as err:
              tap.query_region(sc, width=width, height=height, filter_name=1)
          assert "filter_name must be string" in err.value.args[0]
  
          assert (isinstance(tap.query_region(sc, width=width, height=height, filter_name='filter'), Table))
  
          # Test proposal_id argument
          with pytest.raises(ValueError) as err:
              tap.query_region(sc, width=width, height=height, proposal_id=123)
          assert "proposal_id must be string" in err.value.args[0]
  
          assert (isinstance(tap.query_region(sc, width=width, height=height, proposal_id='123'), Table))
  
          table = tap.query_region(sc, width=width, height=height)
          assert len(table) == 3, f"Wrong job results (num rows). Expected: {3}, found {len(table)}"
          self.__check_results_column(table,
                                      'alpha',
                                      'alpha',
                                      None,
                                      np.float64)
          self.__check_results_column(table,
                                      'delta',
                                      'delta',
                                      None,
                                      np.float64)
          self.__check_results_column(table,
                                      'source_id',
                                      'source_id',
                                      None,
                                      object)
          self.__check_results_column(table,
                                      'table1_oid',
                                      'table1_oid',
                                      None,
                                      np.int32)
          # by radius
          radius = Quantity(1, u.deg)
          table = tap.query_region(sc, radius=radius)
          assert len(table) == 3, f"Wrong job results (num rows). Expected: {3}, found {len(table)}"
          self.__check_results_column(table,
                                      'alpha',
                                      'alpha',
                                      None,
                                      np.float64)
          self.__check_results_column(table,
                                      'delta',
                                      'delta',
                                      None,
                                      np.float64)
          self.__check_results_column(table,
                                      'source_id',
                                      'source_id',
                                      None,
                                      object)
          self.__check_results_column(table,
                                      'table1_oid',
                                      'table1_oid',
                                      None,
                                      np.int32)
  
      def test_query_region_async(self):
          connHandler = DummyConnHandler()
          tapplus = TapPlus("http://test:1111/tap", connhandler=connHandler)
          tap = JwstClass(tap_plus_handler=tapplus, show_messages=False)
          jobid = '12345'
          # Launch response
          responseLaunchJob = DummyResponse(303)
          # list of list (httplib implementation for headers in response)
          launchResponseHeaders = [['location', 'http://test:1111/tap/async/' + jobid]]
          responseLaunchJob.set_data(method='POST', headers=launchResponseHeaders)
          connHandler.set_default_response(responseLaunchJob)
          # Phase response
          responsePhase = DummyResponse(200)
          responsePhase.set_data(method='GET', body="COMPLETED")
          req = "async/" + jobid + "/phase"
          connHandler.set_response(req, responsePhase)
          # Results response
          responseResultsJob = DummyResponse(200)
          jobDataFile = data_path('job_1.vot')
          jobData = utils.read_file_content(jobDataFile)
          responseResultsJob.set_data(method='GET', body=jobData)
          req = "async/" + jobid + "/results/result"
          connHandler.set_response(req, responseResultsJob)
          sc = SkyCoord(ra=29.0, dec=15.0, unit=(u.degree, u.degree),
                        frame='icrs')
          width = Quantity(12, u.deg)
          height = Quantity(10, u.deg)
          table = tap.query_region(sc, width=width, height=height, async_job=True)
          assert len(table) == 3, f"Wrong job results (num rows). Expected: {3}, found {len(table)}"
          self.__check_results_column(table,
                                      'alpha',
                                      'alpha',
                                      None,
                                      np.float64)
          self.__check_results_column(table,
                                      'delta',
                                      'delta',
                                      None,
                                      np.float64)
          self.__check_results_column(table,
                                      'source_id',
                                      'source_id',
                                      None,
                                      object)
          self.__check_results_column(table,
                                      'table1_oid',
                                      'table1_oid',
                                      None,
                                      np.int32)
          # by radius
          radius = Quantity(1, u.deg)
          table = tap.query_region(sc, radius=radius, async_job=True)
          assert len(table) == 3, f"Wrong job results (num rows). Expected: {3}, found {len(table)}"
          self.__check_results_column(table,
                                      'alpha',
                                      'alpha',
                                      None,
                                      np.float64)
          self.__check_results_column(table,
                                      'delta',
                                      'delta',
                                      None,
                                      np.float64)
          self.__check_results_column(table,
                                      'source_id',
                                      'source_id',
                                      None,
                                      object)
          self.__check_results_column(table,
                                      'table1_oid',
                                      'table1_oid',
                                      None,
                                      np.int32)
  
      def test_cone_search_sync(self):
          connHandler = DummyConnHandler()
          tapplus = TapPlus("http://test:1111/tap", connhandler=connHandler)
          tap = JwstClass(tap_plus_handler=tapplus, show_messages=False)
          # Launch response: we use default response because the
          # query contains decimals
          responseLaunchJob = DummyResponse(200)
          jobDataFile = data_path('job_1.vot')
          jobData = utils.read_file_content(jobDataFile)
          responseLaunchJob.set_data(method='POST', body=jobData)
          ra = 19.0
          dec = 20.0
          sc = SkyCoord(ra=ra, dec=dec, unit=(u.degree, u.degree), frame='icrs')
          radius = Quantity(1.0, u.deg)
          connHandler.set_default_response(responseLaunchJob)
          job = tap.cone_search(sc, radius)
          assert job is not None, "Expected a valid job"
          assert job.async_ is False, "Expected a synchronous job"
          assert job.get_phase() == 'COMPLETED', f"Wrong job phase. Expected: {'COMPLETED'}, found {job.get_phase()}"
          assert job.failed is False, "Wrong job status (set Failed = True)"
          # results
          results = job.get_results()
          assert len(results) == 3, f"Wrong job results (num rows). Expected: {3}, found {len(results)}"
          self.__check_results_column(results,
                                      'alpha',
                                      'alpha',
                                      None,
                                      np.float64)
          self.__check_results_column(results,
                                      'delta',
                                      'delta',
                                      None,
                                      np.float64)
          self.__check_results_column(results,
                                      'source_id',
                                      'source_id',
                                      None,
                                      object)
          self.__check_results_column(results,
                                      'table1_oid',
                                      'table1_oid',
                                      None,
                                      np.int32)
  
          # Test observation_id argument
          with pytest.raises(ValueError) as err:
              tap.cone_search(sc, radius, observation_id=1)
          assert "observation_id must be string" in err.value.args[0]
  
          # Test cal_level argument
          with pytest.raises(ValueError) as err:
              tap.cone_search(sc, radius, cal_level='a')
          assert "cal_level must be either 'Top' or an integer" in err.value.args[0]
  
          # Test only_public
          with pytest.raises(ValueError) as err:
              tap.cone_search(sc, radius, only_public='a')
          assert "only_public must be boolean" in err.value.args[0]
  
          # Test dataproduct_type argument
          with pytest.raises(ValueError) as err:
              tap.cone_search(sc, radius, prod_type=1)
          assert "prod_type must be string" in err.value.args[0]
  
          with pytest.raises(ValueError) as err:
              tap.cone_search(sc, radius, prod_type='a')
          assert "prod_type must be one of: " in err.value.args[0]
  
          # Test instrument_name argument
          with pytest.raises(ValueError) as err:
              tap.cone_search(sc, radius, instrument_name=1)
          assert "instrument_name must be string" in err.value.args[0]
  
          with pytest.raises(ValueError) as err:
              tap.cone_search(sc, radius, instrument_name='a')
          assert "instrument_name must be one of: " in err.value.args[0]
  
          # Test filter_name argument
          with pytest.raises(ValueError) as err:
              tap.cone_search(sc, radius, filter_name=1)
          assert "filter_name must be string" in err.value.args[0]
  
          # Test proposal_id argument
          with pytest.raises(ValueError) as err:
              tap.cone_search(sc, radius, proposal_id=123)
          assert "proposal_id must be string" in err.value.args[0]
  
      def test_cone_search_async(self):
          connHandler = DummyConnHandler()
          tapplus = TapPlus("http://test:1111/tap", connhandler=connHandler)
          tap = JwstClass(tap_plus_handler=tapplus, show_messages=False)
          jobid = '12345'
          # Launch response
          responseLaunchJob = DummyResponse(303)
          # list of list (httplib implementation for headers in response)
          launchResponseHeaders = [['location', 'http://test:1111/tap/async/' + jobid]]
          responseLaunchJob.set_data(method='POST', headers=launchResponseHeaders)
          ra = 19
          dec = 20
          sc = SkyCoord(ra=ra, dec=dec, unit=(u.degree, u.degree), frame='icrs')
          radius = Quantity(1.0, u.deg)
          connHandler.set_default_response(responseLaunchJob)
          # Phase response
          responsePhase = DummyResponse(200)
          responsePhase.set_data(method='GET', body="COMPLETED")
          req = "async/" + jobid + "/phase"
          connHandler.set_response(req, responsePhase)
          # Results response
          responseResultsJob = DummyResponse(200)
          jobDataFile = data_path('job_1.vot')
          jobData = utils.read_file_content(jobDataFile)
          responseResultsJob.set_data(method='GET', body=jobData)
          req = "async/" + jobid + "/results/result"
          connHandler.set_response(req, responseResultsJob)
          job = tap.cone_search(sc, radius, async_job=True)
          assert job is not None, "Expected a valid job"
          assert job.async_ is True, "Expected an asynchronous job"
          assert job.get_phase() == 'COMPLETED', f"Wrong job phase. Expected: {'COMPLETED'}, found {job.get_phase()}"
          assert job.failed is False, "Wrong job status (set Failed = True)"
          # results
          results = job.get_results()
          assert len(results) == 3, "Wrong job results (num rows). Expected: {3}, found {len(results)}"
          self.__check_results_column(results,
                                      'alpha',
                                      'alpha',
                                      None,
                                      np.float64)
          self.__check_results_column(results,
                                      'delta',
                                      'delta',
                                      None,
                                      np.float64)
          self.__check_results_column(results,
                                      'source_id',
                                      'source_id',
                                      None,
                                      object)
          self.__check_results_column(results,
                                      'table1_oid',
                                      'table1_oid',
                                      None,
                                      np.int32)
  
      def test_get_product_by_artifactid(self):
          dummyTapHandler = DummyTapHandler()
          jwst = JwstClass(tap_plus_handler=dummyTapHandler, data_handler=dummyTapHandler, show_messages=False)
          # default parameters
          with pytest.raises(ValueError) as err:
              jwst.get_product()
          assert "Missing required argument: 'artifact_id' or 'file_name'" in err.value.args[0]
  
          # test with parameters
          dummyTapHandler.reset()
  
          parameters = {}
          parameters['output_file'] = 'jw00617023001_02102_00001_nrcb4_uncal.fits'
          parameters['verbose'] = False
  
          param_dict = {}
          param_dict['RETRIEVAL_TYPE'] = 'PRODUCT'
          param_dict['TAPCLIENT'] = 'ASTROQUERY'
          param_dict['ARTIFACTID'] = '00000000-0000-0000-8740-65e2827c9895'
          parameters['params_dict'] = param_dict
  
          jwst.get_product(artifact_id='00000000-0000-0000-8740-65e2827c9895')
          dummyTapHandler.check_call('load_data', parameters)
  
      def test_get_product_by_filename(self):
          dummyTapHandler = DummyTapHandler()
          jwst = JwstClass(tap_plus_handler=dummyTapHandler, data_handler=dummyTapHandler, show_messages=False)
          # default parameters
          with pytest.raises(ValueError) as err:
              jwst.get_product()
          assert "Missing required argument: 'artifact_id' or 'file_name'" in err.value.args[0]
  
          # test with parameters
          dummyTapHandler.reset()
  
          parameters = {}
          parameters['output_file'] = 'file_name_id'
          parameters['verbose'] = False
  
          param_dict = {}
          param_dict['RETRIEVAL_TYPE'] = 'PRODUCT'
          param_dict['TAPCLIENT'] = 'ASTROQUERY'
          param_dict['ARTIFACTID'] = '00000000-0000-0000-8740-65e2827c9895'
          parameters['params_dict'] = param_dict
  
          jwst.get_product(file_name='file_name_id')
          dummyTapHandler.check_call('load_data', parameters)
  
      def test_get_products_list(self):
          dummyTapHandler = DummyTapHandler()
          jwst = JwstClass(tap_plus_handler=dummyTapHandler, data_handler=dummyTapHandler, show_messages=False)
          # default parameters
          with pytest.raises(ValueError) as err:
              jwst.get_product_list()
          assert "Missing required argument: 'observation_id'" in err.value.args[0]
  
          # test with parameters
          dummyTapHandler.reset()
  
          observation_id = "jw00777011001_02104_00001_nrcblong"
  
          query = (f"select distinct a.uri, a.artifactid, a.filename, "
                   f"a.contenttype, a.producttype, p.calibrationlevel, p.public "
                   f"FROM {conf.JWST_PLANE_TABLE} p JOIN {conf.JWST_ARTIFACT_TABLE} "
                   f"a ON (p.planeid=a.planeid) WHERE a.planeid "
                   f"IN {planeids};")
  
          parameters = {}
          parameters['query'] = query
          parameters['name'] = None
          parameters['output_file'] = None
          parameters['output_format'] = 'votable'
          parameters['verbose'] = False
          parameters['dump_to_file'] = False
          parameters['upload_resource'] = None
          parameters['upload_table_name'] = None
  
          jwst.get_product_list(observation_id=observation_id)
          dummyTapHandler.check_call('launch_job', parameters)
  
      def test_get_obs_products(self):
          dummyTapHandler = DummyTapHandler()
          jwst = JwstClass(tap_plus_handler=dummyTapHandler, data_handler=dummyTapHandler, show_messages=False)
          # default parameters
          with pytest.raises(ValueError) as err:
              jwst.get_obs_products()
          assert "Missing required argument: 'observation_id'" in err.value.args[0]
  
          # test with parameters
          dummyTapHandler.reset()
  
          output_file_full_path_dir = os.getcwd() + os.sep + "temp_test_jwsttap_get_obs_products_1"
          try:
              os.makedirs(output_file_full_path_dir, exist_ok=True)
          except OSError as err:
              print(f"Creation of the directory {output_file_full_path_dir} failed: {err.strerror}")
              raise err
  
          observation_id = 'jw00777011001_02104_00001_nrcblong'
  
          parameters = {}
          parameters['verbose'] = False
  
          param_dict = {}
          param_dict['RETRIEVAL_TYPE'] = 'OBSERVATION'
          param_dict['TAPCLIENT'] = 'ASTROQUERY'
          param_dict['planeid'] = planeids
          param_dict['calibrationlevel'] = 'ALL'
          parameters['params_dict'] = param_dict
  
          # Test single product tar
          file = data_path('single_product_retrieval.tar')
          output_file_full_path = output_file_full_path_dir + os.sep + os.path.basename(file)
          shutil.copy(file, output_file_full_path)
          parameters['output_file'] = output_file_full_path
  
          expected_files = []
          extracted_file_1 = output_file_full_path_dir + os.sep + 'single_product_retrieval_1.fits'
          expected_files.append(extracted_file_1)
          try:
              files_returned = (jwst.get_obs_products(
                                observation_id=observation_id, cal_level='ALL',
                                output_file=output_file_full_path))
              dummyTapHandler.check_call('load_data', parameters)
              self.__check_extracted_files(files_expected=expected_files,
                                           files_returned=files_returned)
          finally:
              shutil.rmtree(output_file_full_path_dir)
  
          # Test single file
          output_file_full_path_dir = os.getcwd() + os.sep +\
              "temp_test_jwsttap_get_obs_products_2"
          try:
              os.makedirs(output_file_full_path_dir, exist_ok=True)
          except OSError as err:
              print(f"Creation of the directory {output_file_full_path_dir} failed: {err.strerror}")
              raise err
  
          file = data_path('single_product_retrieval_1.fits')
          output_file_full_path = output_file_full_path_dir + os.sep +\
              os.path.basename(file)
          shutil.copy(file, output_file_full_path)
  
          parameters['output_file'] = output_file_full_path
  
          expected_files = []
          expected_files.append(output_file_full_path)
  
          try:
              files_returned = (jwst.get_obs_products(
                                observation_id=observation_id,
                                output_file=output_file_full_path))
              dummyTapHandler.check_call('load_data', parameters)
              self.__check_extracted_files(files_expected=expected_files,
                                           files_returned=files_returned)
          finally:
              # self.__remove_folder_contents(folder=output_file_full_path_dir)
              shutil.rmtree(output_file_full_path_dir)
  
          # Test single file zip
          output_file_full_path_dir = os.getcwd() + os.sep + "temp_test_jwsttap_get_obs_products_3"
          try:
              os.makedirs(output_file_full_path_dir, exist_ok=True)
          except OSError as err:
              print(f"Creation of the directory {output_file_full_path_dir} failed: {err.strerror}")
              raise err
  
          file = data_path('single_product_retrieval_3.fits.zip')
          output_file_full_path = output_file_full_path_dir + os.sep +\
              os.path.basename(file)
          shutil.copy(file, output_file_full_path)
  
          parameters['output_file'] = output_file_full_path
  
          expected_files = []
          extracted_file_1 = output_file_full_path_dir + os.sep + 'single_product_retrieval.fits'
          expected_files.append(extracted_file_1)
  
          try:
              files_returned = (jwst.get_obs_products(
                                observation_id=observation_id,
                                output_file=output_file_full_path))
              dummyTapHandler.check_call('load_data', parameters)
              self.__check_extracted_files(files_expected=expected_files,
                                           files_returned=files_returned)
          finally:
              # self.__remove_folder_contents(folder=output_file_full_path_dir)
              shutil.rmtree(output_file_full_path_dir)
  
          # Test single file gzip
          output_file_full_path_dir = (os.getcwd() + os.sep + "temp_test_jwsttap_get_obs_products_4")
          try:
              os.makedirs(output_file_full_path_dir, exist_ok=True)
          except OSError as err:
              print(f"Creation of the directory {output_file_full_path_dir} failed: {err.strerror}")
              raise err
  
          file = data_path('single_product_retrieval_2.fits.gz')
          output_file_full_path = output_file_full_path_dir + os.sep + os.path.basename(file)
          shutil.copy(file, output_file_full_path)
  
          parameters['output_file'] = output_file_full_path
  
          expected_files = []
          extracted_file_1 = output_file_full_path_dir + os.sep + 'single_product_retrieval_2.fits.gz'
          expected_files.append(extracted_file_1)
  
          try:
              files_returned = (jwst.get_obs_products(
                                observation_id=observation_id,
                                output_file=output_file_full_path))
              dummyTapHandler.check_call('load_data', parameters)
              self.__check_extracted_files(files_expected=expected_files,
                                           files_returned=files_returned)
          finally:
              # self.__remove_folder_contents(folder=output_file_full_path_dir)
              shutil.rmtree(output_file_full_path_dir)
  
          # Test tar with 3 files, a normal one, a gzip one and a zip one
          output_file_full_path_dir = (os.getcwd() + os.sep + "temp_test_jwsttap_get_obs_products_5")
          try:
              os.makedirs(output_file_full_path_dir, exist_ok=True)
          except OSError as err:
              print(f"Creation of the directory {output_file_full_path_dir} failed: {err.strerror}")
              raise err
  
          file = data_path('three_products_retrieval.tar')
          output_file_full_path = output_file_full_path_dir + os.sep + os.path.basename(file)
          shutil.copy(file, output_file_full_path)
  
          parameters['output_file'] = output_file_full_path
  
          expected_files = []
          extracted_file_1 = output_file_full_path_dir + os.sep + 'single_product_retrieval_1.fits'
          expected_files.append(extracted_file_1)
          extracted_file_2 = output_file_full_path_dir + os.sep + 'single_product_retrieval_2.fits.gz'
          expected_files.append(extracted_file_2)
          extracted_file_3 = output_file_full_path_dir + os.sep + 'single_product_retrieval_3.fits.zip'
          expected_files.append(extracted_file_3)
  
          try:
              files_returned = (jwst.get_obs_products(
                                observation_id=observation_id,
                                output_file=output_file_full_path))
              dummyTapHandler.check_call('load_data', parameters)
              self.__check_extracted_files(files_expected=expected_files,
                                           files_returned=files_returned)
          finally:
              # self.__remove_folder_contents(folder=output_file_full_path_dir)
              shutil.rmtree(output_file_full_path_dir)
  
      def test_gunzip_file(self):
          output_file_full_path_dir = (os.getcwd() + os.sep + "temp_test_jwsttap_gunzip")
          try:
              os.makedirs(output_file_full_path_dir, exist_ok=True)
          except OSError as err:
              print(f"Creation of the directory {output_file_full_path_dir} failed: {err.strerror}")
              raise err
  
          file = data_path('single_product_retrieval_2.fits.gz')
          output_file_full_path = output_file_full_path_dir + os.sep + os.path.basename(file)
          shutil.copy(file, output_file_full_path)
  
          expected_files = []
          extracted_file_1 = output_file_full_path_dir + os.sep + "single_product_retrieval_2.fits"
          expected_files.append(extracted_file_1)
  
          try:
              extracted_file = (JwstClass.gzip_uncompress_and_rename_single_file(
                                output_file_full_path))
              if extracted_file != extracted_file_1:
                  raise ValueError(f"Extracted file not found: {extracted_file_1}")
          finally:
              # self.__remove_folder_contents(folder=output_file_full_path_dir)
              shutil.rmtree(output_file_full_path_dir)
  
      def __check_results_column(self, results, columnName, description, unit,
                                 dataType):
          c = results[columnName]
          assert c.description == description, \
              f"Wrong description for results column '{columnName}'. Expected: '{description}', "\
              f"found '{c.description}'"
          assert c.unit == unit, \
              f"Wrong unit for results column '{columnName}'. Expected: '{unit}', found '{c.unit}'"
          assert c.dtype == dataType, \
              f"Wrong dataType for results column '{columnName}'. Expected: '{dataType}', found '{c.dtype}'"
  
      def __remove_folder_contents(self, folder):
          for root, dirs, files in os.walk(folder):
              for f in files:
                  os.unlink(os.path.join(root, f))
              for d in dirs:
                  shutil.rmtree(os.path.join(root, d))
  
      def __check_extracted_files(self, files_expected, files_returned):
          if len(files_expected) != len(files_returned):
              raise ValueError(f"Expected files size error. "
                               f"Found {len(files_returned)}, "
                               f"expected {len(files_expected)}")
          for f in files_expected:
              if not os.path.exists(f):
                  raise ValueError(f"Not found extracted file: "
                                   f"{f}")
              if f not in files_returned:
                  raise ValueError(f"Not found expected file: {f}")
  
      def test_query_target_error(self):
          jwst = JwstClass(show_messages=False)
          simbad = Simbad()
          ned = Ned()
          vizier = Vizier()
          # Testing default parameters
          with pytest.raises((ValueError, TableParseError)) as err:
              jwst.query_target(target_name="M1", target_resolver="")
          assert "This target resolver is not allowed" in err.value.args[0]
          with pytest.raises((ValueError, TableParseError)) as err:
              jwst.query_target("TEST")
          assert ("This target name cannot be determined with this resolver: ALL" in err.value.args[0]
                  or 'Failed to parse' in err.value.args[0])
          with pytest.raises((ValueError, TableParseError)) as err:
              jwst.query_target(target_name="M1", target_resolver="ALL")
          assert err.value.args[0] in ["This target name cannot be determined "
                                       "with this resolver: ALL", "Missing "
                                       "required argument: 'width'"]
  
          # Testing no valid coordinates from resolvers
          simbad_file = data_path('test_query_by_target_name_simbad_ned_error.vot')
          simbad_table = Table.read(simbad_file)
          simbad.query_object = MagicMock(return_value=simbad_table)
          ned_file = data_path('test_query_by_target_name_simbad_ned_error.vot')
          ned_table = Table.read(ned_file)
          ned.query_object = MagicMock(return_value=ned_table)
          vizier_file = data_path('test_query_by_target_name_vizier_error.vot')
          vizier_table = Table.read(vizier_file)
          vizier.query_object = MagicMock(return_value=vizier_table)
  
          # coordinate_error = 'coordinate must be either a string or astropy.coordinates'
          with pytest.raises((ValueError, TableParseError)) as err:
              jwst.query_target(target_name="test", target_resolver="SIMBAD",
                                radius=units.Quantity(5, units.deg))
          assert ('This target name cannot be determined with this resolver: SIMBAD' in err.value.args[0]
                  or 'Failed to parse' in err.value.args[0])
  
          with pytest.raises((ValueError, TableParseError)) as err:
              jwst.query_target(target_name="test", target_resolver="NED",
                                radius=units.Quantity(5, units.deg))
          assert ('This target name cannot be determined with this resolver: NED' in err.value.args[0]
                  or 'Failed to parse' in err.value.args[0])
  
          with pytest.raises((ValueError, TableParseError)) as err:
              jwst.query_target(target_name="test", target_resolver="VIZIER",
                                radius=units.Quantity(5, units.deg))
          assert ('This target name cannot be determined with this resolver: VIZIER' in err.value.args[0]
                  or 'Failed to parse' in err.value.args[0])
  
      def test_remove_jobs(self):
          dummyTapHandler = DummyTapHandler()
          tap = JwstClass(tap_plus_handler=dummyTapHandler, show_messages=False)
          job_list = ['dummyJob']
          parameters = {}
          parameters['jobs_list'] = job_list
          parameters['verbose'] = False
          tap.remove_jobs(job_list)
          dummyTapHandler.check_call('remove_jobs', parameters)
  
      def test_save_results(self):
          dummyTapHandler = DummyTapHandler()
          tap = JwstClass(tap_plus_handler=dummyTapHandler, show_messages=False)
          job = 'dummyJob'
          parameters = {}
          parameters['job'] = job
          parameters['verbose'] = False
          tap.save_results(job)
          dummyTapHandler.check_call('save_results', parameters)
  
      def test_login(self):
          dummyTapHandler = DummyTapHandler()
          tap = JwstClass(tap_plus_handler=dummyTapHandler, show_messages=False)
          parameters = {}
          parameters['user'] = 'test_user'
          parameters['password'] = 'test_password'
          parameters['credentials_file'] = None
          parameters['verbose'] = False
          tap.login(user='test_user', password='test_password')
          dummyTapHandler.check_call('login', parameters)
  
      def test_logout(self):
          dummyTapHandler = DummyTapHandler()
          tap = JwstClass(tap_plus_handler=dummyTapHandler, show_messages=False)
          parameters = {}
          parameters['verbose'] = False
          tap.logout()
          dummyTapHandler.check_call('logout', parameters)
  
      @pytest.mark.noautofixt
      def test_query_get_product(self):
          dummyTapHandler = DummyTapHandler()
          tap = JwstClass(tap_plus_handler=dummyTapHandler, show_messages=False)
          file = 'test_file'
          parameters = {}
          parameters['query'] = f"select * from jwst.artifact a where a.filename = '{file}'"
          parameters['name'] = None
          parameters['output_file'] = None
          parameters['output_format'] = 'votable'
          parameters['verbose'] = False
          parameters['dump_to_file'] = False
          parameters['upload_resource'] = None
          parameters['upload_table_name'] = None
          tap._query_get_product(file_name=file)
          dummyTapHandler.check_call('launch_job', parameters)
  
          artifact = 'test_artifact'
          parameters['query'] = f"select * from jwst.artifact a where a.artifactid = '{artifact}'"
          tap._query_get_product(artifact_id=artifact)
          dummyTapHandler.check_call('launch_job', parameters)
  
      def test_get_related_observations(self):
          dummyTapHandler = DummyTapHandler()
          tap = JwstClass(tap_plus_handler=dummyTapHandler, show_messages=False)
          obs = 'dummyObs'
          tap.get_related_observations(observation_id=obs)
          parameters = {}
          parameters['query'] = f"select * from jwst.main m where m.members like '%{obs}%'"
          parameters['name'] = None
          parameters['output_file'] = None
          parameters['output_format'] = 'votable'
          parameters['verbose'] = False
          parameters['dump_to_file'] = False
          parameters['upload_resource'] = None
          parameters['upload_table_name'] = None
          dummyTapHandler.check_call('launch_job', parameters)
  
files: astroquery/esa/jwst/tests/test_jwsttap.py
files: # Licensed under a 3-clause BSD style license - see LICENSE.rst
  """
  =============
  Gaia TAP plus
  =============
  
  @author: Juan Carlos Segovia
  @contact: juan.carlos.segovia@sciops.esa.int
  
  European Space Astronomy Centre (ESAC)
  European Space Agency (ESA)
  
  Created on 30 jun. 2016
  
  
  """
  import os
  from pathlib import Path
  from unittest.mock import patch
  
  import pytest
  from astropy.table import Column
  from requests import HTTPError
  
  from astroquery.gaia import conf
  from astroquery.gaia.core import GaiaClass
  from astroquery.gaia.tests.DummyTapHandler import DummyTapHandler
  from astroquery.utils.tap.conn.tests.DummyConnHandler import DummyConnHandler
  from astroquery.utils.tap.conn.tests.DummyResponse import DummyResponse
  import astropy.units as u
  from astropy.coordinates.sky_coordinate import SkyCoord
  import numpy as np
  from astroquery.utils.tap.core import TapPlus
  
  
  job_data = (Path(__file__).with_name("data") / "job_1.vot").read_text()
  
  skycoord = SkyCoord(ra=19 * u.deg, dec=20 * u.deg, frame="icrs")
  
  
  @pytest.fixture(scope="module")
  def column_attrs():
      dtypes = {
          "alpha": np.float64,
          "delta": np.float64,
          "source_id": object,
          "table1_oid": np.int32
      }
      columns = {k: Column(name=k, description=k, dtype=v) for k, v in dtypes.items()}
  
      columns["source_id"].meta = {"_votable_string_dtype": "char"}
      return columns
  
  
  @pytest.fixture(scope="module")
  def mock_querier():
      conn_handler = DummyConnHandler()
      tapplus = TapPlus("http://test:1111/tap", connhandler=conn_handler)
      launch_response = DummyResponse(200)
      launch_response.set_data(method="POST", body=job_data)
      # The query contains decimals: default response is more robust.
      conn_handler.set_default_response(launch_response)
      return GaiaClass(conn_handler, tapplus, show_server_messages=False)
  
  
  @pytest.fixture(scope="module")
  def mock_querier_async():
      conn_handler = DummyConnHandler()
      tapplus = TapPlus("http://test:1111/tap", connhandler=conn_handler)
      jobid = "12345"
  
      launch_response = DummyResponse(303)
      launch_response_headers = [["location", "http://test:1111/tap/async/" + jobid]]
      launch_response.set_data(method="POST", headers=launch_response_headers)
      conn_handler.set_default_response(launch_response)
  
      phase_response = DummyResponse(200)
      phase_response.set_data(method="GET", body="COMPLETED")
      conn_handler.set_response("async/" + jobid + "/phase", phase_response)
  
      results_response = DummyResponse(200)
      results_response.set_data(method="GET", body=job_data)
      conn_handler.set_response("async/" + jobid + "/results/result", results_response)
  
      return GaiaClass(conn_handler, tapplus, show_server_messages=False)
  
  
  def test_show_message():
      connHandler = DummyConnHandler()
  
      dummy_response = DummyResponse(200)
  
      message_text = "1653401204784D[type: -100,-1]=Gaia dev is under maintenance"
  
      dummy_response.set_data(method='GET', body=message_text)
      connHandler.set_default_response(dummy_response)
  
      # show_server_messages
      tableRequest = 'notification?action=GetNotifications'
      connHandler.set_response(tableRequest, dummy_response)
  
      tapplus = TapPlus("http://test:1111/tap", connhandler=connHandler)
      GaiaClass(connHandler, tapplus, show_server_messages=True)
  
  
  def test_query_object(column_attrs, mock_querier):
      with pytest.raises(ValueError) as err:
          mock_querier.query_object(skycoord)
      assert "Missing required argument: width" in err.value.args[0]
  
      width = 12 * u.deg
  
      with pytest.raises(ValueError) as err:
          mock_querier.query_object(skycoord, width=width)
      assert "Missing required argument: height" in err.value.args[0]
  
      table = mock_querier.query_object(skycoord, width=width, height=10 * u.deg)
      assert len(table) == 3
      for colname, attrs in column_attrs.items():
          assert table[colname].attrs_equal(attrs)
      # by radius
      table = mock_querier.query_object(skycoord, radius=1 * u.deg)
      assert len(table) == 3
      for colname, attrs in column_attrs.items():
          assert table[colname].attrs_equal(attrs)
  
  
  def test_query_object_async(column_attrs, mock_querier_async):
      table = mock_querier_async.query_object_async(
          skycoord, width=12 * u.deg, height=10 * u.deg
      )
      assert len(table) == 3
      for colname, attrs in column_attrs.items():
          assert table[colname].attrs_equal(attrs)
      # by radius
      table = mock_querier_async.query_object_async(skycoord, radius=1 * u.deg)
      assert len(table) == 3
      for colname, attrs in column_attrs.items():
          assert table[colname].attrs_equal(attrs)
  
  
  def test_cone_search_sync(column_attrs, mock_querier):
      job = mock_querier.cone_search(skycoord, 1 * u.deg)
      assert job.async_ is False
      assert job.get_phase() == "COMPLETED"
      assert job.failed is False
      # results
      results = job.get_results()
      assert len(results) == 3
      for colname, attrs in column_attrs.items():
          assert results[colname].attrs_equal(attrs)
  
  
  def test_cone_search_async(column_attrs, mock_querier_async):
      radius = 1.0 * u.deg
      job = mock_querier_async.cone_search_async(skycoord, radius)
      assert job.async_ is True
      assert job.get_phase() == "COMPLETED"
      assert job.failed is False
      # results
      results = job.get_results()
      assert len(results) == 3
      for colname, attrs in column_attrs.items():
          assert results[colname].attrs_equal(attrs)
  
      # Regression test for #2093 and #2099 - changing the MAIN_GAIA_TABLE
      # had no effect.
      # The preceding tests should have used the default value.
      assert 'gaiadr3.gaia_source' in job.parameters['query']
      with conf.set_temp("MAIN_GAIA_TABLE", "name_from_conf"):
          job = mock_querier_async.cone_search_async(skycoord, radius)
          assert "name_from_conf" in job.parameters["query"]
          # Changing the value through the class should overrule conf.
          mock_querier_async.MAIN_GAIA_TABLE = "name_from_class"
          job = mock_querier_async.cone_search_async(skycoord, radius)
          assert "name_from_class" in job.parameters["query"]
  
  
  def test_load_data():
      dummy_handler = DummyTapHandler()
      tap = GaiaClass(dummy_handler, dummy_handler, show_server_messages=False)
  
      ids = "1,2,3,4"
      retrieval_type = "epoch_photometry"
      verbose = True
      output_file = os.path.abspath("output_file")
      path_to_end_with = os.path.join("gaia", "test", "output_file")
      if not output_file.endswith(path_to_end_with):
          output_file = os.path.abspath(path_to_end_with)
  
      tap.load_data(ids=ids,
                    retrieval_type=retrieval_type,
                    valid_data=True,
                    verbose=verbose,
                    output_file=output_file)
  
      parameters = {
          "params_dict": {
              "VALID_DATA": "true",
              "ID": ids,
              "FORMAT": "votable",
              "RETRIEVAL_TYPE": retrieval_type,
              "DATA_STRUCTURE": "INDIVIDUAL",
              "USE_ZIP_ALWAYS": "true",
          },
          "output_file": dummy_handler._DummyTapHandler__parameters["output_file"],
          "verbose": verbose,
      }
      dummy_handler.check_call('load_data', parameters)
  
  
  def test_get_datalinks():
      dummy_handler = DummyTapHandler()
      tap = GaiaClass(dummy_handler, dummy_handler, show_server_messages=False)
      ids = ["1", "2", "3", "4"]
      verbose = True
      tap.get_datalinks(ids, verbose)
      dummy_handler.check_call("get_datalinks", {"ids": ids, "verbose": verbose})
  
  
  def test_xmatch(mock_querier_async):
      # missing table A
      with pytest.raises(ValueError) as err:
          mock_querier_async.cross_match(
              full_qualified_table_name_b='schemaB.tableB',
              results_table_name='results',
          )
      assert "Table name A argument is mandatory" in err.value.args[0]
      # missing schema A
      with pytest.raises(ValueError) as err:
          mock_querier_async.cross_match(
              full_qualified_table_name_a='tableA',
              full_qualified_table_name_b='schemaB.tableB',
              results_table_name='results',
          )
      assert "Not found schema name in full qualified table A: 'tableA'" \
             in err.value.args[0]
      # missing table B
      with pytest.raises(ValueError) as err:
          mock_querier_async.cross_match(
              full_qualified_table_name_a='schemaA.tableA',
              results_table_name='results',
          )
      assert "Table name B argument is mandatory" in err.value.args[0]
      # missing schema B
      with pytest.raises(ValueError) as err:
          mock_querier_async.cross_match(
              full_qualified_table_name_a='schemaA.tableA',
              full_qualified_table_name_b='tableB',
              results_table_name='results',
          )
      assert "Not found schema name in full qualified table B: 'tableB'" \
             in err.value.args[0]
      # missing results table
      with pytest.raises(ValueError) as err:
          mock_querier_async.cross_match(
              full_qualified_table_name_a='schemaA.tableA',
              full_qualified_table_name_b='schemaB.tableB',
          )
      assert "Results table name argument is mandatory" in err.value.args[0]
      # wrong results table (with schema)
      with pytest.raises(ValueError) as err:
          mock_querier_async.cross_match(
              full_qualified_table_name_a='schemaA.tableA',
              full_qualified_table_name_b='schemaB.tableB',
              results_table_name='schema.results',
          )
      assert "Please, do not specify schema for 'results_table_name'" \
             in err.value.args[0]
      # radius < 0.1
      with pytest.raises(ValueError) as err:
          mock_querier_async.cross_match(
              full_qualified_table_name_a='schemaA.tableA',
              full_qualified_table_name_b='schemaB.tableB',
              results_table_name='results',
              radius=0.01,
          )
      assert "Invalid radius value. Found 0.01, valid range is: 0.1 to 10.0" \
             in err.value.args[0]
      # radius > 10.0
      with pytest.raises(ValueError) as err:
          mock_querier_async.cross_match(
              full_qualified_table_name_a='schemaA.tableA',
              full_qualified_table_name_b='schemaB.tableB',
              results_table_name='results',
              radius=10.1
          )
      assert "Invalid radius value. Found 10.1, valid range is: 0.1 to 10.0" \
             in err.value.args[0]
      job = mock_querier_async.cross_match(
          full_qualified_table_name_a='schemaA.tableA',
          full_qualified_table_name_b='schemaB.tableB',
          results_table_name='results',
      )
      assert job.async_ is True
      assert job.get_phase() == "COMPLETED"
      assert job.failed is False
      job = mock_querier_async.cross_match(
          full_qualified_table_name_a='schemaA.tableA',
          full_qualified_table_name_b='schemaB.tableB',
          results_table_name='results',
          background=True,
      )
      assert job.async_ is True
      assert job.get_phase() == "EXECUTING"
      assert job.failed is False
  
  
  @patch.object(TapPlus, 'login')
  def test_login(mock_login):
      conn_handler = DummyConnHandler()
      tapplus = TapPlus("http://test:1111/tap", connhandler=conn_handler)
      tap = GaiaClass(conn_handler, tapplus, show_server_messages=False)
      tap.login("user", "password")
      assert (mock_login.call_count == 2)
      mock_login.side_effect = HTTPError("Login error")
      tap.login("user", "password")
      assert (mock_login.call_count == 3)
  
  
  @patch.object(TapPlus, 'login_gui')
  @patch.object(TapPlus, 'login')
  def test_login_gui(mock_login_gui, mock_login):
      conn_handler = DummyConnHandler()
      tapplus = TapPlus("http://test:1111/tap", connhandler=conn_handler)
      tap = GaiaClass(conn_handler, tapplus, show_server_messages=False)
      tap.login_gui()
      assert (mock_login_gui.call_count == 1)
      mock_login_gui.side_effect = HTTPError("Login error")
      tap.login("user", "password")
      assert (mock_login.call_count == 1)
  
  
  @patch.object(TapPlus, 'logout')
  def test_logout(mock_logout):
      conn_handler = DummyConnHandler()
      tapplus = TapPlus("http://test:1111/tap", connhandler=conn_handler)
      tap = GaiaClass(conn_handler, tapplus, show_server_messages=False)
      tap.logout()
      assert (mock_logout.call_count == 2)
      mock_logout.side_effect = HTTPError("Login error")
      tap.logout()
      assert (mock_logout.call_count == 3)
  
files: astroquery/gaia/tests/test_gaiatap.py
files: # Licensed under a 3-clause BSD style license - see LICENSE.rst
  """
  =============
  Gaia TAP plus
  =============
  
  @author: Juan Carlos Segovia
  @contact: juan.carlos.segovia@sciops.esa.int
  
  European Space Astronomy Centre (ESAC)
  European Space Agency (ESA)
  
  Created on 30 jun. 2016
  
  
  """
  import os
  from pathlib import Path
  from unittest.mock import patch
  
  import pytest
  from astropy.table import Column
  from requests import HTTPError
  
  from astroquery.gaia import conf
  from astroquery.gaia.core import GaiaClass
  from astroquery.gaia.tests.DummyTapHandler import DummyTapHandler
  from astroquery.utils.tap.conn.tests.DummyConnHandler import DummyConnHandler
  from astroquery.utils.tap.conn.tests.DummyResponse import DummyResponse
  import astropy.units as u
  from astropy.coordinates.sky_coordinate import SkyCoord
  import numpy as np
  from astroquery.utils.tap.xmlparser import utils
  from astroquery.utils.tap.core import TapPlus
  
  
  job_data = utils.read_file_content(Path(__file__).parent.joinpath("data", "job_1.vot"))
  
  skycoord = SkyCoord(ra=19 * u.deg, dec=20 * u.deg, frame="icrs")
  
  
  @pytest.fixture(scope="module")
  def column_attrs():
      dtypes = {
          "alpha": np.float64,
          "delta": np.float64,
          "source_id": object,
          "table1_oid": np.int32
      }
      columns = {k: Column(name=k, description=k, dtype=v) for k, v in dtypes.items()}
  
      columns["source_id"].meta = {"_votable_string_dtype": "char"}
      return columns
  
  
  @pytest.fixture(scope="module")
  def mock_querier():
      conn_handler = DummyConnHandler()
      tapplus = TapPlus("http://test:1111/tap", connhandler=conn_handler)
      launch_response = DummyResponse(200)
      launch_response.set_data(method="POST", body=job_data)
      # The query contains decimals: default response is more robust.
      conn_handler.set_default_response(launch_response)
      return GaiaClass(conn_handler, tapplus, show_server_messages=False)
  
  
  @pytest.fixture(scope="module")
  def mock_querier_async():
      conn_handler = DummyConnHandler()
      tapplus = TapPlus("http://test:1111/tap", connhandler=conn_handler)
      jobid = "12345"
  
      launch_response = DummyResponse(303)
      launch_response_headers = [["location", "http://test:1111/tap/async/" + jobid]]
      launch_response.set_data(method="POST", headers=launch_response_headers)
      conn_handler.set_default_response(launch_response)
  
      phase_response = DummyResponse(200)
      phase_response.set_data(method="GET", body="COMPLETED")
      conn_handler.set_response("async/" + jobid + "/phase", phase_response)
  
      results_response = DummyResponse(200)
      results_response.set_data(method="GET", body=job_data)
      conn_handler.set_response("async/" + jobid + "/results/result", results_response)
  
      return GaiaClass(conn_handler, tapplus, show_server_messages=False)
  
  
  def test_show_message():
      connHandler = DummyConnHandler()
  
      dummy_response = DummyResponse(200)
  
      message_text = "1653401204784D[type: -100,-1]=Gaia dev is under maintenance"
  
      dummy_response.set_data(method='GET', body=message_text)
      connHandler.set_default_response(dummy_response)
  
      # show_server_messages
      tableRequest = 'notification?action=GetNotifications'
      connHandler.set_response(tableRequest, dummy_response)
  
      tapplus = TapPlus("http://test:1111/tap", connhandler=connHandler)
      GaiaClass(connHandler, tapplus, show_server_messages=True)
  
  
  def test_query_object(column_attrs, mock_querier):
      with pytest.raises(ValueError) as err:
          mock_querier.query_object(skycoord)
      assert "Missing required argument: width" in err.value.args[0]
  
      width = 12 * u.deg
  
      with pytest.raises(ValueError) as err:
          mock_querier.query_object(skycoord, width=width)
      assert "Missing required argument: height" in err.value.args[0]
  
      table = mock_querier.query_object(skycoord, width=width, height=10 * u.deg)
      assert len(table) == 3
      for colname, attrs in column_attrs.items():
          assert table[colname].attrs_equal(attrs)
      # by radius
      table = mock_querier.query_object(skycoord, radius=1 * u.deg)
      assert len(table) == 3
      for colname, attrs in column_attrs.items():
          assert table[colname].attrs_equal(attrs)
  
  
  def test_query_object_async(column_attrs, mock_querier_async):
      table = mock_querier_async.query_object_async(
          skycoord, width=12 * u.deg, height=10 * u.deg
      )
      assert len(table) == 3
      for colname, attrs in column_attrs.items():
          assert table[colname].attrs_equal(attrs)
      # by radius
      table = mock_querier_async.query_object_async(skycoord, radius=1 * u.deg)
      assert len(table) == 3
      for colname, attrs in column_attrs.items():
          assert table[colname].attrs_equal(attrs)
  
  
  def test_cone_search_sync(column_attrs, mock_querier):
      job = mock_querier.cone_search(skycoord, 1 * u.deg)
      assert job.async_ is False
      assert job.get_phase() == "COMPLETED"
      assert job.failed is False
      # results
      results = job.get_results()
      assert len(results) == 3
      for colname, attrs in column_attrs.items():
          assert results[colname].attrs_equal(attrs)
  
  
  def test_cone_search_async(column_attrs, mock_querier_async):
      radius = 1.0 * u.deg
      job = mock_querier_async.cone_search_async(skycoord, radius)
      assert job.async_ is True
      assert job.get_phase() == "COMPLETED"
      assert job.failed is False
      # results
      results = job.get_results()
      assert len(results) == 3
      for colname, attrs in column_attrs.items():
          assert results[colname].attrs_equal(attrs)
  
      # Regression test for #2093 and #2099 - changing the MAIN_GAIA_TABLE
      # had no effect.
      # The preceding tests should have used the default value.
      assert 'gaiadr3.gaia_source' in job.parameters['query']
      with conf.set_temp("MAIN_GAIA_TABLE", "name_from_conf"):
          job = mock_querier_async.cone_search_async(skycoord, radius)
          assert "name_from_conf" in job.parameters["query"]
          # Changing the value through the class should overrule conf.
          mock_querier_async.MAIN_GAIA_TABLE = "name_from_class"
          job = mock_querier_async.cone_search_async(skycoord, radius)
          assert "name_from_class" in job.parameters["query"]
  
  
  def test_load_data():
      dummy_handler = DummyTapHandler()
      tap = GaiaClass(dummy_handler, dummy_handler, show_server_messages=False)
  
      ids = "1,2,3,4"
      retrieval_type = "epoch_photometry"
      verbose = True
      output_file = os.path.abspath("output_file")
      path_to_end_with = os.path.join("gaia", "test", "output_file")
      if not output_file.endswith(path_to_end_with):
          output_file = os.path.abspath(path_to_end_with)
  
      tap.load_data(ids=ids,
                    retrieval_type=retrieval_type,
                    valid_data=True,
                    verbose=verbose,
                    output_file=output_file)
  
      parameters = {
          "params_dict": {
              "VALID_DATA": "true",
              "ID": ids,
              "FORMAT": "votable",
              "RETRIEVAL_TYPE": retrieval_type,
              "DATA_STRUCTURE": "INDIVIDUAL",
              "USE_ZIP_ALWAYS": "true",
          },
          "output_file": dummy_handler._DummyTapHandler__parameters["output_file"],
          "verbose": verbose,
      }
      dummy_handler.check_call('load_data', parameters)
  
  
  def test_get_datalinks():
      dummy_handler = DummyTapHandler()
      tap = GaiaClass(dummy_handler, dummy_handler, show_server_messages=False)
      ids = ["1", "2", "3", "4"]
      verbose = True
      tap.get_datalinks(ids, verbose)
      dummy_handler.check_call("get_datalinks", {"ids": ids, "verbose": verbose})
  
  
  def test_xmatch(mock_querier_async):
      # missing table A
      with pytest.raises(ValueError) as err:
          mock_querier_async.cross_match(
              full_qualified_table_name_b='schemaB.tableB',
              results_table_name='results',
          )
      assert "Table name A argument is mandatory" in err.value.args[0]
      # missing schema A
      with pytest.raises(ValueError) as err:
          mock_querier_async.cross_match(
              full_qualified_table_name_a='tableA',
              full_qualified_table_name_b='schemaB.tableB',
              results_table_name='results',
          )
      assert "Not found schema name in full qualified table A: 'tableA'" \
             in err.value.args[0]
      # missing table B
      with pytest.raises(ValueError) as err:
          mock_querier_async.cross_match(
              full_qualified_table_name_a='schemaA.tableA',
              results_table_name='results',
          )
      assert "Table name B argument is mandatory" in err.value.args[0]
      # missing schema B
      with pytest.raises(ValueError) as err:
          mock_querier_async.cross_match(
              full_qualified_table_name_a='schemaA.tableA',
              full_qualified_table_name_b='tableB',
              results_table_name='results',
          )
      assert "Not found schema name in full qualified table B: 'tableB'" \
             in err.value.args[0]
      # missing results table
      with pytest.raises(ValueError) as err:
          mock_querier_async.cross_match(
              full_qualified_table_name_a='schemaA.tableA',
              full_qualified_table_name_b='schemaB.tableB',
          )
      assert "Results table name argument is mandatory" in err.value.args[0]
      # wrong results table (with schema)
      with pytest.raises(ValueError) as err:
          mock_querier_async.cross_match(
              full_qualified_table_name_a='schemaA.tableA',
              full_qualified_table_name_b='schemaB.tableB',
              results_table_name='schema.results',
          )
      assert "Please, do not specify schema for 'results_table_name'" \
             in err.value.args[0]
      # radius < 0.1
      with pytest.raises(ValueError) as err:
          mock_querier_async.cross_match(
              full_qualified_table_name_a='schemaA.tableA',
              full_qualified_table_name_b='schemaB.tableB',
              results_table_name='results',
              radius=0.01,
          )
      assert "Invalid radius value. Found 0.01, valid range is: 0.1 to 10.0" \
             in err.value.args[0]
      # radius > 10.0
      with pytest.raises(ValueError) as err:
          mock_querier_async.cross_match(
              full_qualified_table_name_a='schemaA.tableA',
              full_qualified_table_name_b='schemaB.tableB',
              results_table_name='results',
              radius=10.1
          )
      assert "Invalid radius value. Found 10.1, valid range is: 0.1 to 10.0" \
             in err.value.args[0]
      job = mock_querier_async.cross_match(
          full_qualified_table_name_a='schemaA.tableA',
          full_qualified_table_name_b='schemaB.tableB',
          results_table_name='results',
      )
      assert job.async_ is True
      assert job.get_phase() == "COMPLETED"
      assert job.failed is False
      job = mock_querier_async.cross_match(
          full_qualified_table_name_a='schemaA.tableA',
          full_qualified_table_name_b='schemaB.tableB',
          results_table_name='results',
          background=True,
      )
      assert job.async_ is True
      assert job.get_phase() == "EXECUTING"
      assert job.failed is False
  
  
  @patch.object(TapPlus, 'login')
  def test_login(mock_login):
      conn_handler = DummyConnHandler()
      tapplus = TapPlus("http://test:1111/tap", connhandler=conn_handler)
      tap = GaiaClass(conn_handler, tapplus, show_server_messages=False)
      tap.login("user", "password")
      assert (mock_login.call_count == 2)
      mock_login.side_effect = HTTPError("Login error")
      tap.login("user", "password")
      assert (mock_login.call_count == 3)
  
  
  @patch.object(TapPlus, 'login_gui')
  @patch.object(TapPlus, 'login')
  def test_login_gui(mock_login_gui, mock_login):
      conn_handler = DummyConnHandler()
      tapplus = TapPlus("http://test:1111/tap", connhandler=conn_handler)
      tap = GaiaClass(conn_handler, tapplus, show_server_messages=False)
      tap.login_gui()
      assert (mock_login_gui.call_count == 1)
      mock_login_gui.side_effect = HTTPError("Login error")
      tap.login("user", "password")
      assert (mock_login.call_count == 1)
  
  
  @patch.object(TapPlus, 'logout')
  def test_logout(mock_logout):
      conn_handler = DummyConnHandler()
      tapplus = TapPlus("http://test:1111/tap", connhandler=conn_handler)
      tap = GaiaClass(conn_handler, tapplus, show_server_messages=False)
      tap.logout()
      assert (mock_logout.call_count == 2)
      mock_logout.side_effect = HTTPError("Login error")
      tap.logout()
      assert (mock_logout.call_count == 3)
  
files: astroquery/gaia/tests/test_gaiatap.py
files: # Licensed under a 3-clause BSD style license - see LICENSE.rst
  """
  =============
  TAP plus
  =============
  
  @author: Juan Carlos Segovia
  @contact: juan.carlos.segovia@sciops.esa.int
  
  European Space Astronomy Centre (ESAC)
  European Space Agency (ESA)
  
  Created on 30 jun. 2016
  
  
  """
  from pathlib import Path
  
  import pytest
  
  from astroquery.utils.tap.model.job import Job
  from astroquery.utils.tap.conn.tests.DummyConnHandler import DummyConnHandler
  from astroquery.utils.tap.conn.tests.DummyResponse import DummyResponse
  
  
  def test_job_basic():
  
      with pytest.raises(AttributeError):
          job = Job(async_job=True)
          job.get_results()
  
  
  def test_job_get_results(capsys, tmpdir):
      job = Job(async_job=True)
      jobid = "12345"
      outputFormat = "votable"
      job.jobid = jobid
      job.parameters['format'] = outputFormat
      responseCheckPhase = DummyResponse(500)
      responseCheckPhase.set_data(method='GET', body='FINISHED')
      waitRequest = f"async/{jobid}/phase"
      connHandler = DummyConnHandler()
      connHandler.set_response(waitRequest, responseCheckPhase)
      job.connHandler = connHandler
  
      with pytest.raises(Exception):
          job.get_results()
  
      responseCheckPhase.set_status_code(200)
      responseGetData = DummyResponse(500)
      responseGetData.set_data(
          method="GET",
          body=(Path(__file__).with_name("data") / "result_1.vot").read_text())
      dataRequest = f"async/{jobid}/results/result"
      connHandler.set_response(dataRequest, responseGetData)
  
      with pytest.raises(Exception):
          job.get_results()
  
      responseGetData.set_status_code(200)
      res = job.get_results()
      assert len(res) == 3
      assert len(res.columns) == 4
      for cn in ['alpha', 'delta', 'source_id', 'table1_oid']:
          if cn not in res.colnames:
              pytest.fail(f"{cn} column name not found: {res.colnames}")
  
      # Regression test for #2299; messages were printed even with `verbose=False`
      capsys.readouterr()
      job._Job__resultInMemory = False
      job.save_results(verbose=False)
      assert 'Saving results to:' not in capsys.readouterr().out
      job.save_results(verbose=True)
      assert 'Saving results to:' in capsys.readouterr().out
  
  
  def test_job_phase():
      job = Job(async_job=True)
      jobid = "12345"
      outputFormat = "votable"
      job.jobid = jobid
      job.parameters['format'] = outputFormat
      job.set_phase("COMPLETED")
      try:
          job.set_phase("RUN")
          pytest.fail("Exception expected. Phase cannot be changed for a finished job")
      except ValueError:
          # ok
          pass
      try:
          job.start()
          pytest.fail("Exception expected. A job in 'COMPLETE' phase cannot be started")
      except ValueError:
          # ok
          pass
      try:
          job.abort()
          pytest.fail("Exception expected. A job in 'COMPLETE' phase cannot be aborted")
      except ValueError:
          # ok
          pass
  
files: astroquery/utils/tap/model/tests/test_job.py
files: # Licensed under a 3-clause BSD style license - see LICENSE.rst
  """
  =============
  TAP plus
  =============
  
  @author: Juan Carlos Segovia
  @contact: juan.carlos.segovia@sciops.esa.int
  
  European Space Astronomy Centre (ESAC)
  European Space Agency (ESA)
  
  Created on 30 jun. 2016
  
  
  """
  import os
  import pytest
  
  from astroquery.utils.tap.model.job import Job
  from astroquery.utils.tap.conn.tests.DummyConnHandler import DummyConnHandler
  from astroquery.utils.tap.conn.tests.DummyResponse import DummyResponse
  from astroquery.utils.tap.xmlparser import utils
  
  
  def data_path(filename):
      data_dir = os.path.join(os.path.dirname(__file__), 'data')
      return os.path.join(data_dir, filename)
  
  
  def test_job_basic():
  
      with pytest.raises(AttributeError):
          job = Job(async_job=True)
          job.get_results()
  
  
  def test_job_get_results(capsys, tmpdir):
      job = Job(async_job=True)
      jobid = "12345"
      outputFormat = "votable"
      job.jobid = jobid
      job.parameters['format'] = outputFormat
      responseCheckPhase = DummyResponse(500)
      responseCheckPhase.set_data(method='GET', body='FINISHED')
      waitRequest = f"async/{jobid}/phase"
      connHandler = DummyConnHandler()
      connHandler.set_response(waitRequest, responseCheckPhase)
      job.connHandler = connHandler
  
      with pytest.raises(Exception):
          job.get_results()
  
      responseCheckPhase.set_status_code(200)
      responseGetData = DummyResponse(500)
      jobContentFileName = data_path('result_1.vot')
      jobContent = utils.read_file_content(jobContentFileName)
      responseGetData.set_data(method='GET', body=jobContent)
      dataRequest = f"async/{jobid}/results/result"
      connHandler.set_response(dataRequest, responseGetData)
  
      with pytest.raises(Exception):
          job.get_results()
  
      responseGetData.set_status_code(200)
      res = job.get_results()
      assert len(res) == 3
      assert len(res.columns) == 4
      for cn in ['alpha', 'delta', 'source_id', 'table1_oid']:
          if cn not in res.colnames:
              pytest.fail(f"{cn} column name not found: {res.colnames}")
  
      # Regression test for #2299; messages were printed even with `verbose=False`
      capsys.readouterr()
      job._Job__resultInMemory = False
      job.save_results(verbose=False)
      assert 'Saving results to:' not in capsys.readouterr().out
      job.save_results(verbose=True)
      assert 'Saving results to:' in capsys.readouterr().out
  
  
  def test_job_phase():
      job = Job(async_job=True)
      jobid = "12345"
      outputFormat = "votable"
      job.jobid = jobid
      job.parameters['format'] = outputFormat
      job.set_phase("COMPLETED")
      try:
          job.set_phase("RUN")
          pytest.fail("Exception expected. Phase cannot be changed for a finished job")
      except ValueError:
          # ok
          pass
      try:
          job.start()
          pytest.fail("Exception expected. A job in 'COMPLETE' phase cannot be started")
      except ValueError:
          # ok
          pass
      try:
          job.abort()
          pytest.fail("Exception expected. A job in 'COMPLETE' phase cannot be aborted")
      except ValueError:
          # ok
          pass
  
files: astroquery/utils/tap/model/tests/test_job.py
files: # Licensed under a 3-clause BSD style license - see LICENSE.rst
  """
  =============
  TAP plus
  =============
  
  @author: Juan Carlos Segovia
  @contact: juan.carlos.segovia@sciops.esa.int
  
  European Space Astronomy Centre (ESAC)
  European Space Agency (ESA)
  
  Created on 30 jun. 2016
  """
  from pathlib import Path
  from unittest.mock import patch
  from urllib.parse import quote_plus, urlencode
  
  import numpy as np
  import pytest
  from requests import HTTPError
  
  from astroquery.utils.tap.model.tapcolumn import TapColumn
  
  from astroquery.utils.tap.conn.tests.DummyConnHandler import DummyConnHandler
  from astroquery.utils.tap.conn.tests.DummyResponse import DummyResponse
  from astroquery.utils.tap.core import TapPlus
  from astroquery.utils.tap import taputils
  
  
  TEST_DATA = {f.name: f.read_text() for f in Path(__file__).with_name("data").iterdir()}
  
  
  def test_load_tables():
      connHandler = DummyConnHandler()
      tap = TapPlus("http://test:1111/tap", connhandler=connHandler)
      responseLoadTable = DummyResponse(500)
      responseLoadTable.set_data(method='GET', body=TEST_DATA["test_tables.xml"])
      tableRequest = "tables"
      connHandler.set_response(tableRequest, responseLoadTable)
      with pytest.raises(Exception):
          tap.load_tables()
  
      responseLoadTable.set_status_code(200)
      res = tap.load_tables()
      assert len(res) == 2
  
      # Table 1
      table = __find_table('public', 'table1', res)
      assert table.description == 'Table1 desc'
      columns = table.columns
      assert len(columns) == 2
      col = __find_column('table1_col1', columns)
      __check_column(col, 'Table1 Column1 desc', '', 'VARCHAR', 'indexed')
      col = __find_column('table1_col2', columns)
      __check_column(col, 'Table1 Column2 desc', '', 'INTEGER', None)
  
      # Table 2
      table = __find_table('public', 'table2', res)
      assert table.description == 'Table2 desc'
      columns = table.columns
      assert len(columns) == 3
      col = __find_column('table2_col1', columns)
      __check_column(col, 'Table2 Column1 desc', '', 'VARCHAR', 'indexed')
      col = __find_column('table2_col2', columns)
      __check_column(col, 'Table2 Column2 desc', '', 'INTEGER', None)
      col = __find_column('table2_col3', columns)
      __check_column(col, 'Table2 Column3 desc', '', 'INTEGER', None)
  
  
  def test_load_tables_parameters():
      connHandler = DummyConnHandler()
      tap = TapPlus("http://test:1111/tap", connhandler=connHandler)
      responseLoadTable = DummyResponse(200)
      responseLoadTable.set_data(method='GET', body=TEST_DATA["test_tables.xml"])
      tableRequest = "tables"
      connHandler.set_response(tableRequest, responseLoadTable)
  
      # empty request
      tap.load_tables()
      request = connHandler.get_last_request()
      assert request == tableRequest
  
      # flag only_names=false & share_accessible=false: equals to
      # empty request
      tap.load_tables(only_names=False, include_shared_tables=False)
      request = connHandler.get_last_request()
      assert request == tableRequest
  
      # flag only_names
      tableRequest = "tables?only_tables=true"
      connHandler.set_response(tableRequest, responseLoadTable)
      tap.load_tables(only_names=True)
      request = connHandler.get_last_request()
      assert request == tableRequest
  
      # flag share_accessible=true
      tableRequest = "tables?share_accessible=true"
      connHandler.set_response(tableRequest, responseLoadTable)
      tap.load_tables(include_shared_tables=True)
      request = connHandler.get_last_request()
      assert request == tableRequest
  
      # flag only_names=true & share_accessible=true
      tableRequest = "tables?only_tables=true&share_accessible=true"
      connHandler.set_response(tableRequest, responseLoadTable)
      tap.load_tables(only_names=True, include_shared_tables=True)
      request = connHandler.get_last_request()
      assert request == tableRequest
  
  
  def test_load_table():
      connHandler = DummyConnHandler()
      tap = TapPlus("http://test:1111/tap", connhandler=connHandler)
  
      # No arguments
      with pytest.raises(Exception):
          tap.load_table()
  
      responseLoadTable = DummyResponse(500)
      responseLoadTable.set_data(method='GET', body=TEST_DATA["test_table1.xml"])
      tableSchema = "public"
      tableName = "table1"
      fullQualifiedTableName = f"{tableSchema}.{tableName}"
      tableRequest = f"tables?tables={fullQualifiedTableName}"
      connHandler.set_response(tableRequest, responseLoadTable)
  
      with pytest.raises(Exception):
          tap.load_table(fullQualifiedTableName)
  
      responseLoadTable.set_status_code(200)
      table = tap.load_table(fullQualifiedTableName)
      assert table is not None
      assert table.description == 'Table1 desc'
      columns = table.columns
      assert len(columns) == 2
      col = __find_column('table1_col1', columns)
      __check_column(col, 'Table1 Column1 desc', '', 'VARCHAR', 'indexed')
      col = __find_column('table1_col2', columns)
      __check_column(col, 'Table1 Column2 desc', '', 'INTEGER', None)
  
  
  def test_launch_sync_job():
      connHandler = DummyConnHandler()
      tap = TapPlus("http://test:1111/tap", connhandler=connHandler)
      responseLaunchJob = DummyResponse(500)
      responseLaunchJob.set_data(method='POST', body=TEST_DATA["job_1.vot"])
      query = 'select top 5 * from table'
      dictTmp = {
          "REQUEST": "doQuery",
          "LANG": "ADQL",
          "FORMAT": "votable",
          "tapclient": str(tap.tap_client_id),
          "PHASE": "RUN",
          "QUERY": quote_plus(query)}
      sortedKey = taputils.taputil_create_sorted_dict_key(dictTmp)
      jobRequest = f"sync?{sortedKey}"
      connHandler.set_response(jobRequest, responseLaunchJob)
  
      with pytest.raises(Exception):
          tap.launch_job(query, maxrec=10)
  
      responseLaunchJob.set_status_code(200)
  
      job = tap.launch_job(query)
  
      assert job is not None
      assert job.async_ is False
      assert job.get_phase() == 'COMPLETED'
      assert job.failed is False
  
      # results
      results = job.get_results()
      assert len(results) == 3
      __check_results_column(results,
                             'ra',
                             'ra',
                             None,
                             np.float64)
      __check_results_column(results,
                             'dec',
                             'dec',
                             None,
                             np.float64)
      __check_results_column(results,
                             'source_id',
                             'source_id',
                             None,
                             object)
      __check_results_column(results,
                             'table1_oid',
                             'table1_oid',
                             None,
                             np.int32)
  
  
  def test_launch_sync_job_redirect():
      connHandler = DummyConnHandler()
      tap = TapPlus("http://test:1111/tap", connhandler=connHandler)
      responseLaunchJob = DummyResponse(500)
      jobid = '12345'
      resultsReq = f'sync/{jobid}'
      resultsLocation = f'http://test:1111/tap/{resultsReq}'
      launchResponseHeaders = [
          ['location', resultsLocation]
      ]
      responseLaunchJob.set_data(method='POST')
      query = 'select top 5 * from table'
      dictTmp = {
          "REQUEST": "doQuery",
          "LANG": "ADQL",
          "FORMAT": "votable",
          "tapclient": str(tap.tap_client_id),
          "PHASE": "RUN",
          "QUERY": quote_plus(query)}
      sortedKey = taputils.taputil_create_sorted_dict_key(dictTmp)
      jobRequest = f"sync?{sortedKey}"
      connHandler.set_response(jobRequest, responseLaunchJob)
      # Results response
      responseResultsJob = DummyResponse(500)
      responseResultsJob.set_data(method='GET', body=TEST_DATA["job_1.vot"])
      connHandler.set_response(resultsReq, responseResultsJob)
  
      with pytest.raises(Exception):
          tap.launch_job(query)
  
      # Response is redirect (303)
      # No location available
      responseLaunchJob.set_status_code(303)
      with pytest.raises(Exception):
          tap.launch_job(query)
  
      # Response is redirect (303)
      # Location available
      # Results raises error (500)
      responseResultsJob.set_status_code(200)
      responseLaunchJob.set_data(method='POST', headers=launchResponseHeaders)
      responseResultsJob.set_status_code(500)
      with pytest.raises(Exception):
          tap.launch_job(query)
  
      # Response is redirect (303)
      # Results is 200
      # Location available
      responseResultsJob.set_status_code(200)
      job = tap.launch_job(query)
      assert job is not None
      assert job.async_ is False
      assert job.get_phase() == 'COMPLETED'
      assert job.failed is False
  
      # Results
      results = job.get_results()
      assert len(results) == 3
      __check_results_column(results,
                             'ra',
                             'ra',
                             None,
                             np.float64)
      __check_results_column(results,
                             'dec',
                             'dec',
                             None,
                             np.float64)
      __check_results_column(results,
                             'source_id',
                             'source_id',
                             None,
                             object)
      __check_results_column(results,
                             'table1_oid',
                             'table1_oid',
                             None,
                             np.int32)
  
  
  def test_launch_async_job():
      connHandler = DummyConnHandler()
      tap = TapPlus("http://test:1111/tap", connhandler=connHandler)
      jobid = '12345'
      # Launch response
      responseLaunchJob = DummyResponse(500)
      # list of list (httplib implementation for headers in response)
      launchResponseHeaders = [
          ['location', f'http://test:1111/tap/async/{jobid}']
      ]
      responseLaunchJob.set_data(method='POST', headers=launchResponseHeaders)
      query = 'query'
      dictTmp = {
          "REQUEST": "doQuery",
          "LANG": "ADQL",
          "FORMAT": "votable",
          "tapclient": str(tap.tap_client_id),
          "PHASE": "RUN",
          "QUERY": str(query)}
      sortedKey = taputils.taputil_create_sorted_dict_key(dictTmp)
      req = f"async?{sortedKey}"
      connHandler.set_response(req, responseLaunchJob)
      # Phase response
      responsePhase = DummyResponse(500)
      responsePhase.set_data(method='GET', body="COMPLETED")
      req = f"async/{jobid}/phase"
      connHandler.set_response(req, responsePhase)
      # Results response
      responseResultsJob = DummyResponse(500)
      responseResultsJob.set_data(method='GET', body=TEST_DATA["job_1.vot"])
      req = f"async/{jobid}/results/result"
      connHandler.set_response(req, responseResultsJob)
  
      with pytest.raises(Exception):
          tap.launch_job_async(query)
  
      responseLaunchJob.set_status_code(303)
      with pytest.raises(Exception):
          tap.launch_job_async(query)
  
      responsePhase.set_status_code(200)
      with pytest.raises(Exception):
          tap.launch_job_async(query)
  
      responseResultsJob.set_status_code(200)
      job = tap.launch_job_async(query)
      assert job is not None
      assert job.async_ is True
      assert job.get_phase() == 'COMPLETED'
      assert job.failed is False
  
      # results
      results = job.get_results()
      assert len(results) == 3
      __check_results_column(results,
                             'ra',
                             'ra',
                             None,
                             np.float64)
      __check_results_column(results,
                             'dec',
                             'dec',
                             None,
                             np.float64)
      __check_results_column(results,
                             'source_id',
                             'source_id',
                             None,
                             object)
      __check_results_column(results,
                             'table1_oid',
                             'table1_oid',
                             None,
                             np.int32)
  
  
  def test_start_job():
      connHandler = DummyConnHandler()
      tap = TapPlus("http://test:1111/tap", connhandler=connHandler)
      jobid = '12345'
      # Phase POST response
      responsePhase = DummyResponse(200)
      responsePhase.set_data(method='POST')
      req = f"async/{jobid}/phase?PHASE=RUN"
      connHandler.set_response(req, responsePhase)
      # Launch response
      responseLaunchJob = DummyResponse(303)
      # list of list (httplib implementation for headers in response)
      launchResponseHeaders = [
          ['location', f'http://test:1111/tap/async/{jobid}']
      ]
      responseLaunchJob.set_data(method='POST', headers=launchResponseHeaders)
      query = 'query'
      dictTmp = {
          "REQUEST": "doQuery",
          "LANG": "ADQL",
          "FORMAT": "votable",
          "tapclient": str(tap.tap_client_id),
          "QUERY": str(query)}
      sortedKey = taputils.taputil_create_sorted_dict_key(dictTmp)
      req = f"async?{sortedKey}"
      connHandler.set_response(req, responseLaunchJob)
      # Phase response
      responsePhase = DummyResponse(200)
      responsePhase.set_data(method='GET', body="COMPLETED")
      req = f"async/{jobid}/phase"
      connHandler.set_response(req, responsePhase)
      # Results response
      responseResultsJob = DummyResponse(200)
      responseResultsJob.set_data(method='GET', body=TEST_DATA["job_1.vot"])
      req = f"async/{jobid}/results/result"
      connHandler.set_response(req, responseResultsJob)
  
      responseResultsJob.set_status_code(200)
      job = tap.launch_job_async(query, autorun=False)
      assert job is not None
      assert job.get_phase() == 'PENDING'
  
      # start job
      job.start()
      assert job.get_phase() == 'QUEUED'
  
      # results
      results = job.get_results()
      assert len(results) == 3
      assert job.get_phase() == 'COMPLETED'
      # try to start again
      with pytest.raises(Exception):
          job.start()
  
  
  def test_abort_job():
      connHandler = DummyConnHandler()
      tap = TapPlus("http://test:1111/tap", connhandler=connHandler)
      jobid = '12345'
      # Phase POST response
      responsePhase = DummyResponse(200)
      responsePhase.set_data(method='POST')
      req = f"async/{jobid}/phase?PHASE=ABORT"
      connHandler.set_response(req, responsePhase)
      # Launch response
      responseLaunchJob = DummyResponse(303)
      # list of list (httplib implementation for headers in response)
      launchResponseHeaders = [
          ['location', f'http://test:1111/tap/async/{jobid}']
      ]
      responseLaunchJob.set_data(method='POST', headers=launchResponseHeaders)
      query = 'query'
      dictTmp = {
          "REQUEST": "doQuery",
          "LANG": "ADQL",
          "FORMAT": "votable",
          "MAXREC": 10,
          "tapclient": str(tap.tap_client_id),
          "QUERY": str(query)}
      sortedKey = taputils.taputil_create_sorted_dict_key(dictTmp)
      req = f"async?{sortedKey}"
      connHandler.set_response(req, responseLaunchJob)
  
      job = tap.launch_job_async(query, autorun=False, maxrec=10)
      assert job is not None
      assert job.get_phase() == 'PENDING'
      # abort job
      job.abort()
      assert job.get_phase() == 'ABORT'
      # try to abort again
      with pytest.raises(Exception):
          job.abort()
  
  
  def test_job_parameters():
      connHandler = DummyConnHandler()
      tap = TapPlus("http://test:1111/tap", connhandler=connHandler)
      jobid = '12345'
      # Launch response
      responseLaunchJob = DummyResponse(303)
      # list of list (httplib implementation for headers in response)
      launchResponseHeaders = [
          ['location', f'http://test:1111/tap/async/{jobid}']
      ]
      responseLaunchJob.set_data(method='POST', headers=launchResponseHeaders)
      query = 'query'
      dictTmp = {
          "REQUEST": "doQuery",
          "LANG": "ADQL",
          "FORMAT": "votable",
          "MAXREC": 10,
          "tapclient": str(tap.tap_client_id),
          "QUERY": str(query)}
      sortedKey = taputils.taputil_create_sorted_dict_key(dictTmp)
      req = f"async?{sortedKey}"
      connHandler.set_response(req, responseLaunchJob)
      # Phase response
      responsePhase = DummyResponse(200)
      responsePhase.set_data(method='GET', body="COMPLETED")
      req = f"async/{jobid}/phase"
      connHandler.set_response(req, responsePhase)
      # Results response
      responseResultsJob = DummyResponse(200)
      responseResultsJob.set_data(method='GET', body=TEST_DATA["job_1.vot"])
      req = f"async/{jobid}/results/result"
      connHandler.set_response(req, responseResultsJob)
  
      responseResultsJob.set_status_code(200)
      job = tap.launch_job_async(query, maxrec=10, autorun=False)
      assert job is not None
      assert job.get_phase() == 'PENDING'
  
      # parameter response
      responseParameters = DummyResponse(200)
      responseParameters.set_data(method='GET')
      req = f"async/{jobid}?param1=value1"
      connHandler.set_response(req, responseParameters)
      # Phase POST response
      responsePhase = DummyResponse(200)
      responsePhase.set_data(method='POST')
      req = f"async/{jobid}/phase?PHASE=RUN"
      connHandler.set_response(req, responsePhase)
  
      # send parameter OK
      job.send_parameter("param1", "value1")
      # start job
      job.start()
      assert job.get_phase() == 'QUEUED'
      # try to send a parameter after execution
      with pytest.raises(Exception):
          job.send_parameter("param2", "value2")
  
  
  def test_list_async_jobs():
      connHandler = DummyConnHandler()
      tap = TapPlus("http://test:1111/tap", connhandler=connHandler)
      response = DummyResponse(500)
      response.set_data(method='GET', body=TEST_DATA["jobs_list.xml"])
      req = "async"
      connHandler.set_response(req, response)
      with pytest.raises(Exception):
          tap.list_async_jobs()
  
      response.set_status_code(200)
      jobs = tap.list_async_jobs()
      assert len(jobs) == 2
      assert jobs[0].jobid == '12345'
      assert jobs[0].get_phase() == 'COMPLETED'
      assert jobs[1].jobid == '77777'
      assert jobs[1].get_phase() == 'ERROR'
  
  
  def test_data():
      connHandler = DummyConnHandler()
      tap = TapPlus("http://test:1111/tap",
                    data_context="data",
                    connhandler=connHandler)
      responseResultsJob = DummyResponse(200)
      responseResultsJob.set_data(method='GET', body=TEST_DATA["job_1.vot"])
      req = "?ID=1%2C2&format=votable"
      connHandler.set_response(req, responseResultsJob)
      req = "?ID=1%2C2"
      connHandler.set_response(req, responseResultsJob)
  
      # error
      responseResultsJob.set_status_code(500)
      params_dict = {}
      params_dict['ID'] = "1,2"
      with pytest.raises(Exception):
          tap.load_data(params_dict)
  
      # OK
      responseResultsJob.set_status_code(200)
  
      # results
      results = tap.load_data(params_dict)
      assert len(results) == 3
      # error: no params dictionary
      with pytest.raises(Exception):
          # no dictionary: exception
          tap.load_data("1,2")
      params_dict['format'] = "votable"
      results = tap.load_data(params_dict)
      assert len(results) == 3
  
  
  def test_datalink():
      connHandler = DummyConnHandler()
      tap = TapPlus("http://test:1111/tap",
                    datalink_context="datalink",
                    connhandler=connHandler)
      responseResultsJob = DummyResponse(200)
      responseResultsJob.set_data(method='GET', body=TEST_DATA["job_1.vot"])
      req = "links?ID=1,2"
      connHandler.set_response(req, responseResultsJob)
  
      # error
      responseResultsJob.set_status_code(500)
      with pytest.raises(Exception):
          # missing IDS parameter
          tap.get_datalinks(ids=None)
  
      # OK
      responseResultsJob.set_status_code(200)
      # results
      results = tap.get_datalinks("1,2")
      assert len(results) == 3
      results = tap.get_datalinks([1, 2])
      assert len(results) == 3
      results = tap.get_datalinks(['1', '2'])
      assert len(results) == 3
  
  
  def test_get_new_column_values_for_update():
      # Different column (no modifications)
      list_of_changes = [['column2', 'flags', 'Dec'],
                         ['column2', 'indexed', 'true'],
                         ['column2', 'ucd', 'new_ucd_column2'],
                         ['column2', 'utype', 'new_utype_column2']]
      column_name = 'column'
      c_flags = 'Ra'
      c_indexed = 'false'
      c_ucd = 'ucd'
      c_utype = 'utype_'
      flags, indexed, ucd, utype = TapPlus.get_new_column_values_for_update(
          list_of_changes, column_name, c_flags, c_indexed, c_ucd, c_utype)
      assert flags == c_flags
      assert indexed == c_indexed
      assert ucd == c_ucd
      assert utype == c_utype
  
      # Modifciations
      list_of_changes = [['column2', 'flags', 'Dec'],
                         ['column2', 'indexed', 'true'],
                         ['column2', 'ucd', 'new_ucd_column2'],
                         ['column2', 'utype', 'new_utype_column2'],
                         ['column', 'flags', 'Dec'],
                         ['column', 'indexed', 'true'],
                         ['column', 'ucd', 'new_ucd'],
                         ['column', 'utype', 'new_utype']]
      column_name = 'column'
      c_flags = 'Ra'
      c_indexed = 'false'
      c_ucd = 'ucd'
      c_utype = 'utype'
      n_flags = 'Dec'
      n_indexed = 'True'
      n_ucd = 'new_ucd'
      n_utype = 'new_utype'
      flags, indexed, ucd, utype = TapPlus.get_new_column_values_for_update(
          list_of_changes, column_name, c_flags, c_indexed, c_ucd, c_utype)
      assert flags == n_flags
      assert indexed == n_indexed
      assert ucd == n_ucd
      assert utype == n_utype
  
      # Previous value Ra/indexed, new value remove Ra/indexed
      list_of_changes = [['column', 'flags', ''],
                         ['column', 'indexed', 'false']]
      column_name = 'column'
      c_flags = 'Ra'
      c_indexed = 'true'
      n_flags = ''
      n_indexed = 'false'
      flags, indexed, ucd, utype = TapPlus.get_new_column_values_for_update(
          list_of_changes, column_name, c_flags, c_indexed, c_ucd, c_utype)
      assert flags == n_flags
      assert indexed == n_indexed
  
      # Test if flag is Ra, indexed is True
      list_of_changes = [['column', 'flags', 'Ra'],
                         ['column', 'indexed', 'false']]
      column_name = 'column'
      c_flags = 'Ra'
      c_indexed = 'true'
      n_flags = 'Ra'
      n_indexed = 'True'
      flags, indexed, ucd, utype = TapPlus.get_new_column_values_for_update(
          list_of_changes, column_name, c_flags, c_indexed, c_ucd, c_utype)
      assert flags == n_flags
      assert indexed == n_indexed
  
  
  def test_get_current_column_values_for_update():
      column = TapColumn(None)
      column.name = 'colname'
      column.flags = 0
      column.flag = None
      column.ucd = 'ucd'
      column.utype = 'utype'
      column_name, flags, indexed, ucd, utype = \
          TapPlus.get_current_column_values_for_update(column)
      assert column_name == 'colname'
      assert flags is None
      assert indexed is False
      assert ucd == 'ucd'
      assert utype == 'utype'
  
      column.flags = '1'
      column_name, flags, indexed, ucd, utype = \
          TapPlus.get_current_column_values_for_update(column)
      assert flags == 'Ra'
      assert indexed is True
      column.flags = '2'
      column_name, flags, indexed, ucd, utype = \
          TapPlus.get_current_column_values_for_update(column)
      assert flags == 'Dec'
      assert indexed is True
      column.flags = '4'
      column_name, flags, indexed, ucd, utype = \
          TapPlus.get_current_column_values_for_update(column)
      assert flags == 'Flux'
      assert indexed is False
      column.flags = '8'
      column_name, flags, indexed, ucd, utype = \
          TapPlus.get_current_column_values_for_update(column)
      assert flags == 'Mag'
      assert indexed is False
      column.flags = '16'
      column_name, flags, indexed, ucd, utype = \
          TapPlus.get_current_column_values_for_update(column)
      assert flags == 'PK'
      assert indexed is True
      column.flags = '33'
      column_name, flags, indexed, ucd, utype = \
          TapPlus.get_current_column_values_for_update(column)
      assert flags == 'Ra'
      assert indexed is True
      column.flags = '34'
      column_name, flags, indexed, ucd, utype = \
          TapPlus.get_current_column_values_for_update(column)
      assert flags == 'Dec'
      assert indexed is True
      column.flags = '38'
      column_name, flags, indexed, ucd, utype = \
          TapPlus.get_current_column_values_for_update(column)
      assert flags == 'Flux'
      assert indexed is False
      column.flags = '40'
      column_name, flags, indexed, ucd, utype = \
          TapPlus.get_current_column_values_for_update(column)
      assert flags == 'Mag'
      assert indexed is False
      column.flags = '48'
      column_name, flags, indexed, ucd, utype = \
          TapPlus.get_current_column_values_for_update(column)
      assert flags == 'PK'
      assert indexed is True
  
  
  def test_update_user_table():
      tableName = 'table'
      connHandler = DummyConnHandler()
      tap = TapPlus("http://test:1111/tap", connhandler=connHandler)
      dummyResponse = DummyResponse(200)
      dummyResponse.set_data(method='GET', body=TEST_DATA["test_table_update.xml"])
      tableRequest = f"tables?tables={tableName}"
      connHandler.set_response(tableRequest, dummyResponse)
  
      with pytest.raises(Exception):
          tap.update_user_table()
      with pytest.raises(Exception):
          tap.update_user_table(table_name=tableName)
      with pytest.raises(Exception):
          tap.update_user_table(table_name=tableName, list_of_changes=[])
      with pytest.raises(Exception):
          tap.update_user_table(table_name=tableName, list_of_changes=[[]])
      with pytest.raises(Exception):
          tap.update_user_table(table_name=tableName, list_of_changes=[['', '', '']])
  
      # Test Ra & Dec are provided
      list_of_changes = [['alpha', 'flags', 'Ra']]
      with pytest.raises(Exception):
          tap.update_user_table(table_name=tableName, list_of_changes=list_of_changes)
      list_of_changes = [['delta', 'flags', 'Dec']]
      with pytest.raises(Exception):
          tap.update_user_table(table_name=tableName, list_of_changes=list_of_changes)
  
      # OK
      responseEditTable = DummyResponse(200)
      dictTmp = {
          "ACTION": "edit",
          "NUMTABLES": "1",
          "TABLE0": tableName,
          "TABLE0_COL0": "table6_oid",
          "TABLE0_COL0_FLAGS": "PK",
          "TABLE0_COL0_INDEXED": "True",
          "TABLE0_COL0_UCD": "",
          "TABLE0_COL0_UTYPE": "",
          "TABLE0_COL1": "source_id",
          "TABLE0_COL1_FLAGS": "None",
          "TABLE0_COL1_INDEXED": "False",
          "TABLE0_COL1_UCD": "None",
          "TABLE0_COL1_UTYPE": "None",
          "TABLE0_COL2": "alpha",
          "TABLE0_COL2_FLAGS": "Ra",
          "TABLE0_COL2_INDEXED": "False",
          "TABLE0_COL2_UCD": "",
          "TABLE0_COL2_UTYPE": "",
          "TABLE0_COL3": "delta",
          "TABLE0_COL3_FLAGS": "Dec",
          "TABLE0_COL3_INDEXED": "False",
          "TABLE0_COL3_UCD": "",
          "TABLE0_COL3_UTYPE": "",
          "TABLE0_NUMCOLS": "4"
      }
      sortedKey = taputils.taputil_create_sorted_dict_key(dictTmp)
      req = f"tableEdit?{sortedKey}"
      connHandler.set_response(req, responseEditTable)
  
      list_of_changes = [['alpha', 'flags', 'Ra'], ['delta', 'flags', 'Dec']]
      tap.update_user_table(table_name=tableName, list_of_changes=list_of_changes)
  
  
  def test_rename_table():
      tableName = 'user_test.table_test_rename'
      newTableName = 'user_test.table_test_rename_new'
      newColumnNames = {'ra': 'alpha', 'dec': 'delta'}
      connHandler = DummyConnHandler()
      tap = TapPlus("http://test:1111/tap", connhandler=connHandler)
      dummyResponse = DummyResponse(200)
      dummyResponse.set_data(method='GET', body=TEST_DATA["test_table_rename.xml"])
  
      with pytest.raises(Exception):
          tap.rename_table()
      with pytest.raises(Exception):
          tap.rename_table(table_name=tableName)
      with pytest.raises(Exception):
          tap.rename_table(table_name=tableName, new_table_name=None, new_column_names_dict=None)
  
      # Test OK.
      responseRenameTable = DummyResponse(200)
      dictArgs = {
          "action": "rename",
          "new_column_names": "ra:alpha,dec:delta",
          "new_table_name": newTableName,
          "table_name": tableName,
      }
      connHandler.set_response(f"TableTool?{urlencode(dictArgs)}", responseRenameTable)
      tap.rename_table(table_name=tableName, new_table_name=newTableName, new_column_names_dict=newColumnNames)
  
  
  def __find_table(schemaName, tableName, tables):
      qualifiedName = f"{schemaName}.{tableName}"
      for table in (tables):
          if table.get_qualified_name() == qualifiedName:
              return table
      # not found: raise exception
      pytest.fail(f"Table '{qualifiedName}' not found")
  
  
  def __find_column(columnName, columns):
      for c in (columns):
          if c.name == columnName:
              return c
      # not found: raise exception
      pytest.fail(f"Column '{columnName}' not found")
  
  
  def __check_column(column, description, unit, dataType, flag):
      assert column.description == description
      assert column.unit == unit
      assert column.data_type == dataType
      assert column.flag == flag
  
  
  def __check_results_column(results, columnName, description, unit,
                             dataType):
      c = results[columnName]
      assert c.description == description
      assert c.unit == unit
      assert c.dtype == dataType
  
  
  @patch.object(TapPlus, 'login')
  def test_login(mock_login):
      conn_handler = DummyConnHandler()
      tap = TapPlus("http://test:1111/tap", connhandler=conn_handler)
      tap.login("user", "password")
      assert (mock_login.call_count == 1)
      mock_login.side_effect = HTTPError("Login error")
      with pytest.raises(HTTPError):
          tap.login("user", "password")
      assert (mock_login.call_count == 2)
  
  
  @patch.object(TapPlus, 'login_gui')
  @patch.object(TapPlus, 'login')
  def test_login_gui(mock_login_gui, mock_login):
      conn_handler = DummyConnHandler()
      tap = TapPlus("http://test:1111/tap", connhandler=conn_handler)
      tap.login_gui()
      assert (mock_login_gui.call_count == 0)
      mock_login_gui.side_effect = HTTPError("Login error")
      with pytest.raises(HTTPError):
          tap.login("user", "password")
      assert (mock_login.call_count == 1)
  
  
  @patch.object(TapPlus, 'logout')
  def test_logout(mock_logout):
      conn_handler = DummyConnHandler()
      tap = TapPlus("http://test:1111/tap", connhandler=conn_handler)
      tap.logout()
      assert (mock_logout.call_count == 1)
      mock_logout.side_effect = HTTPError("Login error")
      with pytest.raises(HTTPError):
          tap.logout()
      assert (mock_logout.call_count == 2)
  
files: astroquery/utils/tap/tests/test_tap.py
files: # Licensed under a 3-clause BSD style license - see LICENSE.rst
  """
  =============
  TAP plus
  =============
  
  @author: Juan Carlos Segovia
  @contact: juan.carlos.segovia@sciops.esa.int
  
  European Space Astronomy Centre (ESAC)
  European Space Agency (ESA)
  
  Created on 30 jun. 2016
  """
  import os
  from unittest.mock import patch
  from urllib.parse import quote_plus, urlencode
  
  import numpy as np
  import pytest
  from requests import HTTPError
  
  from astroquery.utils.tap.model.tapcolumn import TapColumn
  
  from astroquery.utils.tap.conn.tests.DummyConnHandler import DummyConnHandler
  from astroquery.utils.tap.conn.tests.DummyResponse import DummyResponse
  from astroquery.utils.tap.core import TapPlus
  from astroquery.utils.tap.xmlparser import utils
  from astroquery.utils.tap import taputils
  
  
  def data_path(filename):
      data_dir = os.path.join(os.path.dirname(__file__), 'data')
      return os.path.join(data_dir, filename)
  
  
  def test_load_tables():
      connHandler = DummyConnHandler()
      tap = TapPlus("http://test:1111/tap", connhandler=connHandler)
      responseLoadTable = DummyResponse(500)
      tableDataFile = data_path('test_tables.xml')
      tableData = utils.read_file_content(tableDataFile)
      responseLoadTable.set_data(method='GET', body=tableData)
      tableRequest = "tables"
      connHandler.set_response(tableRequest, responseLoadTable)
      with pytest.raises(Exception):
          tap.load_tables()
  
      responseLoadTable.set_status_code(200)
      res = tap.load_tables()
      assert len(res) == 2
  
      # Table 1
      table = __find_table('public', 'table1', res)
      assert table.description == 'Table1 desc'
      columns = table.columns
      assert len(columns) == 2
      col = __find_column('table1_col1', columns)
      __check_column(col, 'Table1 Column1 desc', '', 'VARCHAR', 'indexed')
      col = __find_column('table1_col2', columns)
      __check_column(col, 'Table1 Column2 desc', '', 'INTEGER', None)
  
      # Table 2
      table = __find_table('public', 'table2', res)
      assert table.description == 'Table2 desc'
      columns = table.columns
      assert len(columns) == 3
      col = __find_column('table2_col1', columns)
      __check_column(col, 'Table2 Column1 desc', '', 'VARCHAR', 'indexed')
      col = __find_column('table2_col2', columns)
      __check_column(col, 'Table2 Column2 desc', '', 'INTEGER', None)
      col = __find_column('table2_col3', columns)
      __check_column(col, 'Table2 Column3 desc', '', 'INTEGER', None)
  
  
  def test_load_tables_parameters():
      connHandler = DummyConnHandler()
      tap = TapPlus("http://test:1111/tap", connhandler=connHandler)
      responseLoadTable = DummyResponse(200)
      tableDataFile = data_path('test_tables.xml')
      tableData = utils.read_file_content(tableDataFile)
      responseLoadTable.set_data(method='GET', body=tableData)
      tableRequest = "tables"
      connHandler.set_response(tableRequest, responseLoadTable)
  
      # empty request
      tap.load_tables()
      request = connHandler.get_last_request()
      assert request == tableRequest
  
      # flag only_names=false & share_accessible=false: equals to
      # empty request
      tap.load_tables(only_names=False, include_shared_tables=False)
      request = connHandler.get_last_request()
      assert request == tableRequest
  
      # flag only_names
      tableRequest = "tables?only_tables=true"
      connHandler.set_response(tableRequest, responseLoadTable)
      tap.load_tables(only_names=True)
      request = connHandler.get_last_request()
      assert request == tableRequest
  
      # flag share_accessible=true
      tableRequest = "tables?share_accessible=true"
      connHandler.set_response(tableRequest, responseLoadTable)
      tap.load_tables(include_shared_tables=True)
      request = connHandler.get_last_request()
      assert request == tableRequest
  
      # flag only_names=true & share_accessible=true
      tableRequest = "tables?only_tables=true&share_accessible=true"
      connHandler.set_response(tableRequest, responseLoadTable)
      tap.load_tables(only_names=True, include_shared_tables=True)
      request = connHandler.get_last_request()
      assert request == tableRequest
  
  
  def test_load_table():
      connHandler = DummyConnHandler()
      tap = TapPlus("http://test:1111/tap", connhandler=connHandler)
  
      # No arguments
      with pytest.raises(Exception):
          tap.load_table()
  
      responseLoadTable = DummyResponse(500)
      tableDataFile = data_path('test_table1.xml')
      tableData = utils.read_file_content(tableDataFile)
      responseLoadTable.set_data(method='GET', body=tableData)
      tableSchema = "public"
      tableName = "table1"
      fullQualifiedTableName = f"{tableSchema}.{tableName}"
      tableRequest = f"tables?tables={fullQualifiedTableName}"
      connHandler.set_response(tableRequest, responseLoadTable)
  
      with pytest.raises(Exception):
          tap.load_table(fullQualifiedTableName)
  
      responseLoadTable.set_status_code(200)
      table = tap.load_table(fullQualifiedTableName)
      assert table is not None
      assert table.description == 'Table1 desc'
      columns = table.columns
      assert len(columns) == 2
      col = __find_column('table1_col1', columns)
      __check_column(col, 'Table1 Column1 desc', '', 'VARCHAR', 'indexed')
      col = __find_column('table1_col2', columns)
      __check_column(col, 'Table1 Column2 desc', '', 'INTEGER', None)
  
  
  def test_launch_sync_job():
      connHandler = DummyConnHandler()
      tap = TapPlus("http://test:1111/tap", connhandler=connHandler)
      responseLaunchJob = DummyResponse(500)
      jobDataFile = data_path('job_1.vot')
      jobData = utils.read_file_content(jobDataFile)
      responseLaunchJob.set_data(method='POST', body=jobData)
      query = 'select top 5 * from table'
      dictTmp = {
          "REQUEST": "doQuery",
          "LANG": "ADQL",
          "FORMAT": "votable",
          "tapclient": str(tap.tap_client_id),
          "PHASE": "RUN",
          "QUERY": quote_plus(query)}
      sortedKey = taputils.taputil_create_sorted_dict_key(dictTmp)
      jobRequest = f"sync?{sortedKey}"
      connHandler.set_response(jobRequest, responseLaunchJob)
  
      with pytest.raises(Exception):
          tap.launch_job(query, maxrec=10)
  
      responseLaunchJob.set_status_code(200)
  
      job = tap.launch_job(query)
  
      assert job is not None
      assert job.async_ is False
      assert job.get_phase() == 'COMPLETED'
      assert job.failed is False
  
      # results
      results = job.get_results()
      assert len(results) == 3
      __check_results_column(results,
                             'ra',
                             'ra',
                             None,
                             np.float64)
      __check_results_column(results,
                             'dec',
                             'dec',
                             None,
                             np.float64)
      __check_results_column(results,
                             'source_id',
                             'source_id',
                             None,
                             object)
      __check_results_column(results,
                             'table1_oid',
                             'table1_oid',
                             None,
                             np.int32)
  
  
  def test_launch_sync_job_redirect():
      connHandler = DummyConnHandler()
      tap = TapPlus("http://test:1111/tap", connhandler=connHandler)
      responseLaunchJob = DummyResponse(500)
      jobid = '12345'
      resultsReq = f'sync/{jobid}'
      resultsLocation = f'http://test:1111/tap/{resultsReq}'
      launchResponseHeaders = [
          ['location', resultsLocation]
      ]
      responseLaunchJob.set_data(method='POST')
      query = 'select top 5 * from table'
      dictTmp = {
          "REQUEST": "doQuery",
          "LANG": "ADQL",
          "FORMAT": "votable",
          "tapclient": str(tap.tap_client_id),
          "PHASE": "RUN",
          "QUERY": quote_plus(query)}
      sortedKey = taputils.taputil_create_sorted_dict_key(dictTmp)
      jobRequest = f"sync?{sortedKey}"
      connHandler.set_response(jobRequest, responseLaunchJob)
      # Results response
      responseResultsJob = DummyResponse(500)
      jobDataFile = data_path('job_1.vot')
      jobData = utils.read_file_content(jobDataFile)
      responseResultsJob.set_data(method='GET', body=jobData)
      connHandler.set_response(resultsReq, responseResultsJob)
  
      with pytest.raises(Exception):
          tap.launch_job(query)
  
      # Response is redirect (303)
      # No location available
      responseLaunchJob.set_status_code(303)
      with pytest.raises(Exception):
          tap.launch_job(query)
  
      # Response is redirect (303)
      # Location available
      # Results raises error (500)
      responseResultsJob.set_status_code(200)
      responseLaunchJob.set_data(method='POST', headers=launchResponseHeaders)
      responseResultsJob.set_status_code(500)
      with pytest.raises(Exception):
          tap.launch_job(query)
  
      # Response is redirect (303)
      # Results is 200
      # Location available
      responseResultsJob.set_status_code(200)
      job = tap.launch_job(query)
      assert job is not None
      assert job.async_ is False
      assert job.get_phase() == 'COMPLETED'
      assert job.failed is False
  
      # Results
      results = job.get_results()
      assert len(results) == 3
      __check_results_column(results,
                             'ra',
                             'ra',
                             None,
                             np.float64)
      __check_results_column(results,
                             'dec',
                             'dec',
                             None,
                             np.float64)
      __check_results_column(results,
                             'source_id',
                             'source_id',
                             None,
                             object)
      __check_results_column(results,
                             'table1_oid',
                             'table1_oid',
                             None,
                             np.int32)
  
  
  def test_launch_async_job():
      connHandler = DummyConnHandler()
      tap = TapPlus("http://test:1111/tap", connhandler=connHandler)
      jobid = '12345'
      # Launch response
      responseLaunchJob = DummyResponse(500)
      # list of list (httplib implementation for headers in response)
      launchResponseHeaders = [
          ['location', f'http://test:1111/tap/async/{jobid}']
      ]
      responseLaunchJob.set_data(method='POST', headers=launchResponseHeaders)
      query = 'query'
      dictTmp = {
          "REQUEST": "doQuery",
          "LANG": "ADQL",
          "FORMAT": "votable",
          "tapclient": str(tap.tap_client_id),
          "PHASE": "RUN",
          "QUERY": str(query)}
      sortedKey = taputils.taputil_create_sorted_dict_key(dictTmp)
      req = f"async?{sortedKey}"
      connHandler.set_response(req, responseLaunchJob)
      # Phase response
      responsePhase = DummyResponse(500)
      responsePhase.set_data(method='GET', body="COMPLETED")
      req = f"async/{jobid}/phase"
      connHandler.set_response(req, responsePhase)
      # Results response
      responseResultsJob = DummyResponse(500)
      jobDataFile = data_path('job_1.vot')
      jobData = utils.read_file_content(jobDataFile)
      responseResultsJob.set_data(method='GET', body=jobData)
      req = f"async/{jobid}/results/result"
      connHandler.set_response(req, responseResultsJob)
  
      with pytest.raises(Exception):
          tap.launch_job_async(query)
  
      responseLaunchJob.set_status_code(303)
      with pytest.raises(Exception):
          tap.launch_job_async(query)
  
      responsePhase.set_status_code(200)
      with pytest.raises(Exception):
          tap.launch_job_async(query)
  
      responseResultsJob.set_status_code(200)
      job = tap.launch_job_async(query)
      assert job is not None
      assert job.async_ is True
      assert job.get_phase() == 'COMPLETED'
      assert job.failed is False
  
      # results
      results = job.get_results()
      assert len(results) == 3
      __check_results_column(results,
                             'ra',
                             'ra',
                             None,
                             np.float64)
      __check_results_column(results,
                             'dec',
                             'dec',
                             None,
                             np.float64)
      __check_results_column(results,
                             'source_id',
                             'source_id',
                             None,
                             object)
      __check_results_column(results,
                             'table1_oid',
                             'table1_oid',
                             None,
                             np.int32)
  
  
  def test_start_job():
      connHandler = DummyConnHandler()
      tap = TapPlus("http://test:1111/tap", connhandler=connHandler)
      jobid = '12345'
      # Phase POST response
      responsePhase = DummyResponse(200)
      responsePhase.set_data(method='POST')
      req = f"async/{jobid}/phase?PHASE=RUN"
      connHandler.set_response(req, responsePhase)
      # Launch response
      responseLaunchJob = DummyResponse(303)
      # list of list (httplib implementation for headers in response)
      launchResponseHeaders = [
          ['location', f'http://test:1111/tap/async/{jobid}']
      ]
      responseLaunchJob.set_data(method='POST', headers=launchResponseHeaders)
      query = 'query'
      dictTmp = {
          "REQUEST": "doQuery",
          "LANG": "ADQL",
          "FORMAT": "votable",
          "tapclient": str(tap.tap_client_id),
          "QUERY": str(query)}
      sortedKey = taputils.taputil_create_sorted_dict_key(dictTmp)
      req = f"async?{sortedKey}"
      connHandler.set_response(req, responseLaunchJob)
      # Phase response
      responsePhase = DummyResponse(200)
      responsePhase.set_data(method='GET', body="COMPLETED")
      req = f"async/{jobid}/phase"
      connHandler.set_response(req, responsePhase)
      # Results response
      responseResultsJob = DummyResponse(200)
      jobDataFile = data_path('job_1.vot')
      jobData = utils.read_file_content(jobDataFile)
      responseResultsJob.set_data(method='GET', body=jobData)
      req = f"async/{jobid}/results/result"
      connHandler.set_response(req, responseResultsJob)
  
      responseResultsJob.set_status_code(200)
      job = tap.launch_job_async(query, autorun=False)
      assert job is not None
      assert job.get_phase() == 'PENDING'
  
      # start job
      job.start()
      assert job.get_phase() == 'QUEUED'
  
      # results
      results = job.get_results()
      assert len(results) == 3
      assert job.get_phase() == 'COMPLETED'
      # try to start again
      with pytest.raises(Exception):
          job.start()
  
  
  def test_abort_job():
      connHandler = DummyConnHandler()
      tap = TapPlus("http://test:1111/tap", connhandler=connHandler)
      jobid = '12345'
      # Phase POST response
      responsePhase = DummyResponse(200)
      responsePhase.set_data(method='POST')
      req = f"async/{jobid}/phase?PHASE=ABORT"
      connHandler.set_response(req, responsePhase)
      # Launch response
      responseLaunchJob = DummyResponse(303)
      # list of list (httplib implementation for headers in response)
      launchResponseHeaders = [
          ['location', f'http://test:1111/tap/async/{jobid}']
      ]
      responseLaunchJob.set_data(method='POST', headers=launchResponseHeaders)
      query = 'query'
      dictTmp = {
          "REQUEST": "doQuery",
          "LANG": "ADQL",
          "FORMAT": "votable",
          "MAXREC": 10,
          "tapclient": str(tap.tap_client_id),
          "QUERY": str(query)}
      sortedKey = taputils.taputil_create_sorted_dict_key(dictTmp)
      req = f"async?{sortedKey}"
      connHandler.set_response(req, responseLaunchJob)
  
      job = tap.launch_job_async(query, autorun=False, maxrec=10)
      assert job is not None
      assert job.get_phase() == 'PENDING'
      # abort job
      job.abort()
      assert job.get_phase() == 'ABORT'
      # try to abort again
      with pytest.raises(Exception):
          job.abort()
  
  
  def test_job_parameters():
      connHandler = DummyConnHandler()
      tap = TapPlus("http://test:1111/tap", connhandler=connHandler)
      jobid = '12345'
      # Launch response
      responseLaunchJob = DummyResponse(303)
      # list of list (httplib implementation for headers in response)
      launchResponseHeaders = [
          ['location', f'http://test:1111/tap/async/{jobid}']
      ]
      responseLaunchJob.set_data(method='POST', headers=launchResponseHeaders)
      query = 'query'
      dictTmp = {
          "REQUEST": "doQuery",
          "LANG": "ADQL",
          "FORMAT": "votable",
          "MAXREC": 10,
          "tapclient": str(tap.tap_client_id),
          "QUERY": str(query)}
      sortedKey = taputils.taputil_create_sorted_dict_key(dictTmp)
      req = f"async?{sortedKey}"
      connHandler.set_response(req, responseLaunchJob)
      # Phase response
      responsePhase = DummyResponse(200)
      responsePhase.set_data(method='GET', body="COMPLETED")
      req = f"async/{jobid}/phase"
      connHandler.set_response(req, responsePhase)
      # Results response
      responseResultsJob = DummyResponse(200)
      jobDataFile = data_path('job_1.vot')
      jobData = utils.read_file_content(jobDataFile)
      responseResultsJob.set_data(method='GET', body=jobData)
      req = f"async/{jobid}/results/result"
      connHandler.set_response(req, responseResultsJob)
  
      responseResultsJob.set_status_code(200)
      job = tap.launch_job_async(query, maxrec=10, autorun=False)
      assert job is not None
      assert job.get_phase() == 'PENDING'
  
      # parameter response
      responseParameters = DummyResponse(200)
      responseParameters.set_data(method='GET')
      req = f"async/{jobid}?param1=value1"
      connHandler.set_response(req, responseParameters)
      # Phase POST response
      responsePhase = DummyResponse(200)
      responsePhase.set_data(method='POST')
      req = f"async/{jobid}/phase?PHASE=RUN"
      connHandler.set_response(req, responsePhase)
  
      # send parameter OK
      job.send_parameter("param1", "value1")
      # start job
      job.start()
      assert job.get_phase() == 'QUEUED'
      # try to send a parameter after execution
      with pytest.raises(Exception):
          job.send_parameter("param2", "value2")
  
  
  def test_list_async_jobs():
      connHandler = DummyConnHandler()
      tap = TapPlus("http://test:1111/tap", connhandler=connHandler)
      response = DummyResponse(500)
      jobDataFile = data_path('jobs_list.xml')
      jobData = utils.read_file_content(jobDataFile)
      response.set_data(method='GET', body=jobData)
      req = "async"
      connHandler.set_response(req, response)
      with pytest.raises(Exception):
          tap.list_async_jobs()
  
      response.set_status_code(200)
      jobs = tap.list_async_jobs()
      assert len(jobs) == 2
      assert jobs[0].jobid == '12345'
      assert jobs[0].get_phase() == 'COMPLETED'
      assert jobs[1].jobid == '77777'
      assert jobs[1].get_phase() == 'ERROR'
  
  
  def test_data():
      connHandler = DummyConnHandler()
      tap = TapPlus("http://test:1111/tap",
                    data_context="data",
                    connhandler=connHandler)
      responseResultsJob = DummyResponse(200)
      jobDataFile = data_path('job_1.vot')
      jobData = utils.read_file_content(jobDataFile)
      responseResultsJob.set_data(method='GET', body=jobData)
      req = "?ID=1%2C2&format=votable"
      connHandler.set_response(req, responseResultsJob)
      req = "?ID=1%2C2"
      connHandler.set_response(req, responseResultsJob)
  
      # error
      responseResultsJob.set_status_code(500)
      params_dict = {}
      params_dict['ID'] = "1,2"
      with pytest.raises(Exception):
          tap.load_data(params_dict)
  
      # OK
      responseResultsJob.set_status_code(200)
  
      # results
      results = tap.load_data(params_dict)
      assert len(results) == 3
      # error: no params dictionary
      with pytest.raises(Exception):
          # no dictionary: exception
          tap.load_data("1,2")
      params_dict['format'] = "votable"
      results = tap.load_data(params_dict)
      assert len(results) == 3
  
  
  def test_datalink():
      connHandler = DummyConnHandler()
      tap = TapPlus("http://test:1111/tap",
                    datalink_context="datalink",
                    connhandler=connHandler)
      responseResultsJob = DummyResponse(200)
      jobDataFile = data_path('job_1.vot')
      jobData = utils.read_file_content(jobDataFile)
      responseResultsJob.set_data(method='GET', body=jobData)
      req = "links?ID=1,2"
      connHandler.set_response(req, responseResultsJob)
  
      # error
      responseResultsJob.set_status_code(500)
      with pytest.raises(Exception):
          # missing IDS parameter
          tap.get_datalinks(ids=None)
  
      # OK
      responseResultsJob.set_status_code(200)
      # results
      results = tap.get_datalinks("1,2")
      assert len(results) == 3
      results = tap.get_datalinks([1, 2])
      assert len(results) == 3
      results = tap.get_datalinks(['1', '2'])
      assert len(results) == 3
  
  
  def test_get_new_column_values_for_update():
      # Different column (no modifications)
      list_of_changes = [['column2', 'flags', 'Dec'],
                         ['column2', 'indexed', 'true'],
                         ['column2', 'ucd', 'new_ucd_column2'],
                         ['column2', 'utype', 'new_utype_column2']]
      column_name = 'column'
      c_flags = 'Ra'
      c_indexed = 'false'
      c_ucd = 'ucd'
      c_utype = 'utype_'
      flags, indexed, ucd, utype = TapPlus.get_new_column_values_for_update(
          list_of_changes, column_name, c_flags, c_indexed, c_ucd, c_utype)
      assert flags == c_flags
      assert indexed == c_indexed
      assert ucd == c_ucd
      assert utype == c_utype
  
      # Modifciations
      list_of_changes = [['column2', 'flags', 'Dec'],
                         ['column2', 'indexed', 'true'],
                         ['column2', 'ucd', 'new_ucd_column2'],
                         ['column2', 'utype', 'new_utype_column2'],
                         ['column', 'flags', 'Dec'],
                         ['column', 'indexed', 'true'],
                         ['column', 'ucd', 'new_ucd'],
                         ['column', 'utype', 'new_utype']]
      column_name = 'column'
      c_flags = 'Ra'
      c_indexed = 'false'
      c_ucd = 'ucd'
      c_utype = 'utype'
      n_flags = 'Dec'
      n_indexed = 'True'
      n_ucd = 'new_ucd'
      n_utype = 'new_utype'
      flags, indexed, ucd, utype = TapPlus.get_new_column_values_for_update(
          list_of_changes, column_name, c_flags, c_indexed, c_ucd, c_utype)
      assert flags == n_flags
      assert indexed == n_indexed
      assert ucd == n_ucd
      assert utype == n_utype
  
      # Previous value Ra/indexed, new value remove Ra/indexed
      list_of_changes = [['column', 'flags', ''],
                         ['column', 'indexed', 'false']]
      column_name = 'column'
      c_flags = 'Ra'
      c_indexed = 'true'
      n_flags = ''
      n_indexed = 'false'
      flags, indexed, ucd, utype = TapPlus.get_new_column_values_for_update(
          list_of_changes, column_name, c_flags, c_indexed, c_ucd, c_utype)
      assert flags == n_flags
      assert indexed == n_indexed
  
      # Test if flag is Ra, indexed is True
      list_of_changes = [['column', 'flags', 'Ra'],
                         ['column', 'indexed', 'false']]
      column_name = 'column'
      c_flags = 'Ra'
      c_indexed = 'true'
      n_flags = 'Ra'
      n_indexed = 'True'
      flags, indexed, ucd, utype = TapPlus.get_new_column_values_for_update(
          list_of_changes, column_name, c_flags, c_indexed, c_ucd, c_utype)
      assert flags == n_flags
      assert indexed == n_indexed
  
  
  def test_get_current_column_values_for_update():
      column = TapColumn(None)
      column.name = 'colname'
      column.flags = 0
      column.flag = None
      column.ucd = 'ucd'
      column.utype = 'utype'
      column_name, flags, indexed, ucd, utype = \
          TapPlus.get_current_column_values_for_update(column)
      assert column_name == 'colname'
      assert flags is None
      assert indexed is False
      assert ucd == 'ucd'
      assert utype == 'utype'
  
      column.flags = '1'
      column_name, flags, indexed, ucd, utype = \
          TapPlus.get_current_column_values_for_update(column)
      assert flags == 'Ra'
      assert indexed is True
      column.flags = '2'
      column_name, flags, indexed, ucd, utype = \
          TapPlus.get_current_column_values_for_update(column)
      assert flags == 'Dec'
      assert indexed is True
      column.flags = '4'
      column_name, flags, indexed, ucd, utype = \
          TapPlus.get_current_column_values_for_update(column)
      assert flags == 'Flux'
      assert indexed is False
      column.flags = '8'
      column_name, flags, indexed, ucd, utype = \
          TapPlus.get_current_column_values_for_update(column)
      assert flags == 'Mag'
      assert indexed is False
      column.flags = '16'
      column_name, flags, indexed, ucd, utype = \
          TapPlus.get_current_column_values_for_update(column)
      assert flags == 'PK'
      assert indexed is True
      column.flags = '33'
      column_name, flags, indexed, ucd, utype = \
          TapPlus.get_current_column_values_for_update(column)
      assert flags == 'Ra'
      assert indexed is True
      column.flags = '34'
      column_name, flags, indexed, ucd, utype = \
          TapPlus.get_current_column_values_for_update(column)
      assert flags == 'Dec'
      assert indexed is True
      column.flags = '38'
      column_name, flags, indexed, ucd, utype = \
          TapPlus.get_current_column_values_for_update(column)
      assert flags == 'Flux'
      assert indexed is False
      column.flags = '40'
      column_name, flags, indexed, ucd, utype = \
          TapPlus.get_current_column_values_for_update(column)
      assert flags == 'Mag'
      assert indexed is False
      column.flags = '48'
      column_name, flags, indexed, ucd, utype = \
          TapPlus.get_current_column_values_for_update(column)
      assert flags == 'PK'
      assert indexed is True
  
  
  def test_update_user_table():
      tableName = 'table'
      connHandler = DummyConnHandler()
      tap = TapPlus("http://test:1111/tap", connhandler=connHandler)
      dummyResponse = DummyResponse(200)
      tableDataFile = data_path('test_table_update.xml')
      tableData = utils.read_file_content(tableDataFile)
      dummyResponse.set_data(method='GET', body=tableData)
      tableRequest = f"tables?tables={tableName}"
      connHandler.set_response(tableRequest, dummyResponse)
  
      with pytest.raises(Exception):
          tap.update_user_table()
      with pytest.raises(Exception):
          tap.update_user_table(table_name=tableName)
      with pytest.raises(Exception):
          tap.update_user_table(table_name=tableName, list_of_changes=[])
      with pytest.raises(Exception):
          tap.update_user_table(table_name=tableName, list_of_changes=[[]])
      with pytest.raises(Exception):
          tap.update_user_table(table_name=tableName, list_of_changes=[['', '', '']])
  
      # Test Ra & Dec are provided
      list_of_changes = [['alpha', 'flags', 'Ra']]
      with pytest.raises(Exception):
          tap.update_user_table(table_name=tableName, list_of_changes=list_of_changes)
      list_of_changes = [['delta', 'flags', 'Dec']]
      with pytest.raises(Exception):
          tap.update_user_table(table_name=tableName, list_of_changes=list_of_changes)
  
      # OK
      responseEditTable = DummyResponse(200)
      dictTmp = {
          "ACTION": "edit",
          "NUMTABLES": "1",
          "TABLE0": tableName,
          "TABLE0_COL0": "table6_oid",
          "TABLE0_COL0_FLAGS": "PK",
          "TABLE0_COL0_INDEXED": "True",
          "TABLE0_COL0_UCD": "",
          "TABLE0_COL0_UTYPE": "",
          "TABLE0_COL1": "source_id",
          "TABLE0_COL1_FLAGS": "None",
          "TABLE0_COL1_INDEXED": "False",
          "TABLE0_COL1_UCD": "None",
          "TABLE0_COL1_UTYPE": "None",
          "TABLE0_COL2": "alpha",
          "TABLE0_COL2_FLAGS": "Ra",
          "TABLE0_COL2_INDEXED": "False",
          "TABLE0_COL2_UCD": "",
          "TABLE0_COL2_UTYPE": "",
          "TABLE0_COL3": "delta",
          "TABLE0_COL3_FLAGS": "Dec",
          "TABLE0_COL3_INDEXED": "False",
          "TABLE0_COL3_UCD": "",
          "TABLE0_COL3_UTYPE": "",
          "TABLE0_NUMCOLS": "4"
      }
      sortedKey = taputils.taputil_create_sorted_dict_key(dictTmp)
      req = f"tableEdit?{sortedKey}"
      connHandler.set_response(req, responseEditTable)
  
      list_of_changes = [['alpha', 'flags', 'Ra'], ['delta', 'flags', 'Dec']]
      tap.update_user_table(table_name=tableName, list_of_changes=list_of_changes)
  
  
  def test_rename_table():
      tableName = 'user_test.table_test_rename'
      newTableName = 'user_test.table_test_rename_new'
      newColumnNames = {'ra': 'alpha', 'dec': 'delta'}
      connHandler = DummyConnHandler()
      tap = TapPlus("http://test:1111/tap", connhandler=connHandler)
      dummyResponse = DummyResponse(200)
      tableDataFile = data_path('test_table_rename.xml')
      tableData = utils.read_file_content(tableDataFile)
      dummyResponse.set_data(method='GET', body=tableData)
  
      with pytest.raises(Exception):
          tap.rename_table()
      with pytest.raises(Exception):
          tap.rename_table(table_name=tableName)
      with pytest.raises(Exception):
          tap.rename_table(table_name=tableName, new_table_name=None, new_column_names_dict=None)
  
      # Test OK.
      responseRenameTable = DummyResponse(200)
      dictArgs = {
          "action": "rename",
          "new_column_names": "ra:alpha,dec:delta",
          "new_table_name": newTableName,
          "table_name": tableName,
      }
      connHandler.set_response(f"TableTool?{urlencode(dictArgs)}", responseRenameTable)
      tap.rename_table(table_name=tableName, new_table_name=newTableName, new_column_names_dict=newColumnNames)
  
  
  def __find_table(schemaName, tableName, tables):
      qualifiedName = f"{schemaName}.{tableName}"
      for table in (tables):
          if table.get_qualified_name() == qualifiedName:
              return table
      # not found: raise exception
      pytest.fail(f"Table '{qualifiedName}' not found")
  
  
  def __find_column(columnName, columns):
      for c in (columns):
          if c.name == columnName:
              return c
      # not found: raise exception
      pytest.fail(f"Column '{columnName}' not found")
  
  
  def __check_column(column, description, unit, dataType, flag):
      assert column.description == description
      assert column.unit == unit
      assert column.data_type == dataType
      assert column.flag == flag
  
  
  def __check_results_column(results, columnName, description, unit,
                             dataType):
      c = results[columnName]
      assert c.description == description
      assert c.unit == unit
      assert c.dtype == dataType
  
  
  @patch.object(TapPlus, 'login')
  def test_login(mock_login):
      conn_handler = DummyConnHandler()
      tap = TapPlus("http://test:1111/tap", connhandler=conn_handler)
      tap.login("user", "password")
      assert (mock_login.call_count == 1)
      mock_login.side_effect = HTTPError("Login error")
      with pytest.raises(HTTPError):
          tap.login("user", "password")
      assert (mock_login.call_count == 2)
  
  
  @patch.object(TapPlus, 'login_gui')
  @patch.object(TapPlus, 'login')
  def test_login_gui(mock_login_gui, mock_login):
      conn_handler = DummyConnHandler()
      tap = TapPlus("http://test:1111/tap", connhandler=conn_handler)
      tap.login_gui()
      assert (mock_login_gui.call_count == 0)
      mock_login_gui.side_effect = HTTPError("Login error")
      with pytest.raises(HTTPError):
          tap.login("user", "password")
      assert (mock_login.call_count == 1)
  
  
  @patch.object(TapPlus, 'logout')
  def test_logout(mock_logout):
      conn_handler = DummyConnHandler()
      tap = TapPlus("http://test:1111/tap", connhandler=conn_handler)
      tap.logout()
      assert (mock_logout.call_count == 1)
      mock_logout.side_effect = HTTPError("Login error")
      with pytest.raises(HTTPError):
          tap.logout()
      assert (mock_logout.call_count == 2)
  
files: astroquery/utils/tap/tests/test_tap.py
files: # Licensed under a 3-clause BSD style license - see LICENSE.rst
  """
  =============
  TAP plus
  =============
  
  @author: Juan Carlos Segovia
  @contact: juan.carlos.segovia@sciops.esa.int
  
  European Space Astronomy Centre (ESAC)
  European Space Agency (ESA)
  
  Created on 30 jun. 2016
  
  
  """
  import gzip
  import io
  from astropy import units as u
  from astropy.table import Table as APTable
  
  
  def util_create_string_from_buffer(buffer):
      return ''.join(map(str, buffer))
  
  
  def read_http_response(response, output_format, correct_units=True):
      astropy_format = get_suitable_astropy_format(output_format)
  
      # If we want to use astropy.table, we have to read the data
      data = io.BytesIO(response.read())
  
      try:
          result = APTable.read(io.BytesIO(gzip.decompress(data.read())), format=astropy_format)
      except OSError:
          # data is not a valid gzip file by BadGzipFile.
          result = APTable.read(data, format=astropy_format)
          pass
  
      if correct_units:
          modify_unrecognized_table_units(result)
  
      return result
  
  
  def get_suitable_astropy_format(output_format):
      if 'ecsv' == output_format:
          return 'ascii.ecsv'
      elif 'csv' == output_format:
          return 'ascii.csv'
      elif 'votable_plain' == output_format:
          return 'votable'
      return output_format
  
  
  def modify_unrecognized_table_units(table):
      """Modifies the units of an input table column in place
      """
      for cn in table.colnames:
          col = table[cn]
          if isinstance(col.unit, u.UnrecognizedUnit):
              try:
                  col.unit = u.Unit(col.unit.name.replace(".", " ").replace("'", ""))
              except Exception:
                  pass
          elif isinstance(col.unit, str):
              col.unit = col.unit.replace(".", " ").replace("'", "")
  
files: astroquery/utils/tap/xmlparser/utils.py
files: # Licensed under a 3-clause BSD style license - see LICENSE.rst
  """
  =============
  TAP plus
  =============
  
  @author: Juan Carlos Segovia
  @contact: juan.carlos.segovia@sciops.esa.int
  
  European Space Astronomy Centre (ESAC)
  European Space Agency (ESA)
  
  Created on 30 jun. 2016
  
  
  """
  import gzip
  import io
  from astropy import units as u
  from astropy.table import Table as APTable
  
  
  def util_create_string_from_buffer(buffer):
      return ''.join(map(str, buffer))
  
  
  def read_http_response(response, output_format, correct_units=True):
      astropy_format = get_suitable_astropy_format(output_format)
  
      # If we want to use astropy.table, we have to read the data
      data = io.BytesIO(response.read())
  
      try:
          result = APTable.read(io.BytesIO(gzip.decompress(data.read())), format=astropy_format)
      except OSError:
          # data is not a valid gzip file by BadGzipFile.
          result = APTable.read(data, format=astropy_format)
          pass
  
      if correct_units:
          modify_unrecognized_table_units(result)
  
      return result
  
  
  def get_suitable_astropy_format(output_format):
      if 'ecsv' == output_format:
          return 'ascii.ecsv'
      elif 'csv' == output_format:
          return 'ascii.csv'
      elif 'votable_plain' == output_format:
          return 'votable'
      return output_format
  
  
  def read_file_content(file_path):
      file_handler = open(file_path, 'r')
      file_content = file_handler.read()
      file_handler.close()
      return file_content
  
  
  def modify_unrecognized_table_units(table):
      """Modifies the units of an input table column in place
      """
      for cn in table.colnames:
          col = table[cn]
          if isinstance(col.unit, u.UnrecognizedUnit):
              try:
                  col.unit = u.Unit(col.unit.name.replace(".", " ").replace("'", ""))
              except Exception:
                  pass
          elif isinstance(col.unit, str):
              col.unit = col.unit.replace(".", " ").replace("'", "")
  
files: astroquery/utils/tap/xmlparser/utils.py