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
  
  [... the rest is omitted ...]
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 = '
  
  [... the rest is omitted ...]
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
  
  [... the rest is omitted ...]
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_
  
  [... the rest is omitted ...]
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
  
  
  [... the rest is omitted ...]
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
  
  
  [... the rest is omitted ...]
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,
                             '
  
  [... the rest is omitted ...]
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.l
  
  [... the rest is omitted ...]
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("'", "")
  
  
  [... the rest is omitted ...]
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("'", "")
  
  
  [... the rest is omitted ...]
files: astroquery/utils/tap/xmlparser/utils.py