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.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 import TableList
  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
         
  
  [... 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.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 import TableList
  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
         
  
  [... 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 unittest.mock import patch
  
  import pytest
  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
  from astropy.units import Quantity
  import numpy as np
  from astroquery.utils.tap.xmlparser import utils
  from astroquery.utils.tap.core import TapPlus, TAP_CLIENT_ID
  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)
  
  
  class TestTap:
  
      def test_show_message(self):
          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)
          tap = GaiaClass(connHandler, tapplus, show_server_messages=True)
  
      def test_query_object(self):
          conn_handler = DummyConnHandler()
          # Launch response: we use default response because the query contains
          # decimals
          dummy_response = DummyResponse(200)
  
          message_text = "1653401204784D[type: -100,-1]=Gaia dev is under maintenance"
  
          dummy_response.set_data(method='GET', body=message_text)
          conn_handler.set_default_response(dummy_response)
  
          # show_server_messages
          tableRequest = 'notification?action=GetNotifications'
          conn_handler.set_response(tableRequest, dummy_response)
  
          tapplus = TapPlus("http://test:1111/tap", connhandler=conn_handler)
          tap = GaiaClass(conn_handler, tapplus, show_server_messages=True)
          # Launch response: we use default response because the query contains
          # decimals
          response_launch_job = DummyResponse(200)
          job_data_file = data_path('job_1.vot')
          job_data = utils.read_file_content(job_data_file)
          response_launch_job.set_data(method='POST', body=job_data)
          # The query contains decimals: force default response
          conn_handler.set_default_response(response_launch_job)
          sc = SkyCoord(ra=29.0, dec=15.0, unit=(u.degree, u.degree),
                        frame='icrs')
          with pytest.raises(ValueError) as err:
              tap.query_object(sc)
          assert "Missing required argument: width" in err.value.args[0]
  
          width = Quantity(12, u.deg)
  
          with pytest.raises(ValueError) as err:
              tap.query_object(sc, width=width)
          assert "Missing required argument: height" in err.value.args[0]
  
          height = Quantity(10, u.deg)
          table = tap.query_object(sc, width=width, height=height)
          assert len(table) == 3, \
              "Wrong job results (num rows). Expected: %d, found %d" % \
              (3, 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_object(sc, radius=radius)
          assert len(table) == 3, \
              "Wrong job results (num rows). Expected: %d, found %d" % \
              (3, 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_object_async(self):
          conn_handler = DummyConnHandler()
          tapplus = TapPlus("http://test:1111/tap", connhandler=conn_handler)
          tap = GaiaClass(conn_handler, tapplus, show_server_messages=False)
          jobid = '12345'
          # Launch response
          response_launch_job = DummyResponse(303)
          # list of list (httplib implementation for headers in response)
          launch_response_headers = [
              ['location', 'http://test:1111/tap/async/' + jobid]
          ]
          response_launch_job.set_data(method='POST', headers=launch_response_headers)
          conn_handler.set_default_response(response_launch_job)
          # Phase response
         
  
  [... 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 unittest.mock import patch
  
  import pytest
  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
  from astropy.units import Quantity
  import numpy as np
  from astroquery.utils.tap.xmlparser import utils
  from astroquery.utils.tap.core import TapPlus, TAP_CLIENT_ID
  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)
  
  
  class TestTap:
  
      def test_show_message(self):
          connHandler = DummyConnHandler()
  
          dummy_response = DummyResponse(200)
  
          message_text = "1653401204784D[type: -100,-1]=Gaia dev is under maintenance"
  
          dummy_response.set_data(method='GET',
                                  context=None,
                                  body=message_text,
                                  headers=None)
          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)
          tap = GaiaClass(connHandler, tapplus, show_server_messages=True)
  
      def test_query_object(self):
          conn_handler = DummyConnHandler()
          # Launch response: we use default response because the query contains
          # decimals
          dummy_response = DummyResponse(200)
  
          message_text = "1653401204784D[type: -100,-1]=Gaia dev is under maintenance"
  
          dummy_response.set_data(method='GET',
                                  context=None,
                                  body=message_text,
                                  headers=None)
          conn_handler.set_default_response(dummy_response)
  
          # show_server_messages
          tableRequest = 'notification?action=GetNotifications'
          conn_handler.set_response(tableRequest, dummy_response)
  
          tapplus = TapPlus("http://test:1111/tap", connhandler=conn_handler)
          tap = GaiaClass(conn_handler, tapplus, show_server_messages=True)
          # Launch response: we use default response because the query contains
          # decimals
          response_launch_job = DummyResponse(200)
          job_data_file = data_path('job_1.vot')
          job_data = utils.read_file_content(job_data_file)
          response_launch_job.set_data(method='POST',
                                       context=None,
                                       body=job_data,
                                       headers=None)
          # The query contains decimals: force default response
          conn_handler.set_default_response(response_launch_job)
          sc = SkyCoord(ra=29.0, dec=15.0, unit=(u.degree, u.degree),
                        frame='icrs')
          with pytest.raises(ValueError) as err:
              tap.query_object(sc)
          assert "Missing required argument: width" in err.value.args[0]
  
          width = Quantity(12, u.deg)
  
          with pytest.raises(ValueError) as err:
              tap.query_object(sc, width=width)
          assert "Missing required argument: height" in err.value.args[0]
  
          height = Quantity(10, u.deg)
          table = tap.query_object(sc, width=width, height=height)
          assert len(table) == 3, \
              "Wrong job results (num rows). Expected: %d, found %d" % \
              (3, 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_object(sc, radius=radius)
          assert len(table) == 3, \
              "Wrong job results (num rows). Expected: %d, found %d" % \
              (3, 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_object_async(self):
          conn_handler = DummyConnHandler()
          tapplus = TapPlus("http://test:1111/tap", connhandler=conn_handler)
          tap = GaiaClass(conn_handler, tapplus, show_server_messages=False)
          jobid = '12345'
          # Launch response
          response_launch_job = DummyResponse(303)
          # list of list (httplib implementation for headers in response)
          launch_response_headers = [
              ['location', 'http://test:1111/tap/async/' + jobid]
          ]
          response_launch_job.set_data(method='POST',
  
  
  [... 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 astroquery.utils.tap.conn.tests.DummyResponse import DummyResponse
  
  
  class DummyConn:
      '''
      classdocs
      '''
  
      def __init__(self, name=None):
          self.response = DummyResponse()
          self.name = name
          self.httpConn = DummyHttpConn(self.response)
          self.cookie = None
          self.ishttps = False
  
      def request(self, method, context=None, body=None, headers=None):
          self.response.set_data(method, context, body, headers)
  
      def getresponse(self):
          return self.response
  
      def get_connection(self, ishttps=False, cookie=None, verbose=False):
          self.ishttps = ishttps
          self.cookie = cookie
          return self.httpConn
  
      def get_connection_secure(self, verbose):
          return self.httpConn
  
      def __str__(self):
          return self.name
  
  
  class DummyHttpConn:
  
      def __init__(self, response):
          self.response = response
          self.reqmethod = None
          self.requrl = None
          self.reqbody = None
          self.reqheaders = None
  
      def request(self, method, url, body=None, headers=None):
          self.reqmethod = method
          self.requrl = url
          self.reqbody = body
          self.reqheaders = headers
          self.response.set_data(method, url, body, headers)
          return self.response
  
      def getresponse(self):
          return self.response
  
  
  [... the rest is omitted ...]
files: astroquery/utils/tap/conn/tests/DummyConn.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 astroquery.utils.tap.conn.tests.DummyResponse import DummyResponse
  
  
  class DummyConn:
      '''
      classdocs
      '''
  
      def __init__(self, name=None):
          self.response = DummyResponse()
          self.name = name
          self.httpConn = DummyHttpConn(self.response)
          self.cookie = None
          self.ishttps = False
  
      def request(self, method, context, body, headers):
          self.response.set_data(method, context, body, headers)
  
      def getresponse(self):
          return self.response
  
      def get_connection(self, ishttps=False, cookie=None, verbose=False):
          self.ishttps = ishttps
          self.cookie = cookie
          return self.httpConn
  
      def get_connection_secure(self, verbose):
          return self.httpConn
  
      def __str__(self):
          return self.name
  
  
  class DummyHttpConn:
  
      def __init__(self, response):
          self.response = response
          self.reqmethod = None
          self.requrl = None
          self.reqbody = None
          self.reqheaders = None
  
      def request(self, method, url, body=None, headers=None):
          self.reqmethod = method
          self.requrl = url
          self.reqbody = body
          self.reqheaders = headers
          self.response.set_data(method, url, body, headers)
          return self.response
  
      def getresponse(self):
          return self.response
  
  
  [... the rest is omitted ...]
files: astroquery/utils/tap/conn/tests/DummyConn.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
  
  
  """
  
  
  class DummyResponse:
      '''
      classdocs
      '''
  
      STATUS_MESSAGES = {200: "OK", 303: "OK", 500: "ERROR"}
  
      def __init__(self, status_code=None):
          self.reason = ""
          self.set_status_code(status_code)
          self.index = 0
          self.set_data(None, None, None, None)
  
      def set_status_code(self, status_code):
          self.status = status_code
          self.reason = self.STATUS_MESSAGES.get(status_code)
  
      def set_data(self, method, context=None, body=None, headers=None):
          self.method = method
          self.context = context
          self.body = body
          self.headers = headers
          self.index = 0
  
      def get_method(self):
          return self.method
  
      def get_context(self):
          return self.context
  
      def get_body(self):
          return self.body
  
      def read(self, size=None):
          v = self.get_body()
          if v is None:
              return None
          else:
              if size is None or size < 0:
                  # read all
                  return v.encode(encoding='utf_8', errors='strict')
              else:
                  bodyLength = len(v)
                  if self.index < 0:
                      return ""
                  if size >= bodyLength:
                      size = bodyLength - 1
                  if size <= 0:
                      return ""
                  endPos = self.index + size
                  tmp = v[self.index:endPos]
                  self.index = endPos
                  if endPos >= (bodyLength - 1):
                      self.index = -1
                  return tmp.encode(encoding='utf_8', errors='strict')
  
      def close(self):
          self.index = 0
  
      def getheaders(self):
          return self.headers
  
  
  [... the rest is omitted ...]
files: astroquery/utils/tap/conn/tests/DummyResponse.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
  
  
  """
  
  
  class DummyResponse:
      '''
      classdocs
      '''
  
      STATUS_MESSAGES = {200: "OK", 303: "OK", 500: "ERROR"}
  
      def __init__(self, status_code=None):
          self.reason = ""
          self.set_status_code(status_code)
          self.index = 0
          self.set_data(None, None, None, None)
  
      def set_status_code(self, status_code):
          self.status = status_code
          self.reason = self.STATUS_MESSAGES.get(status_code)
  
      def set_data(self, method, context, body, headers):
          self.method = method
          self.context = context
          self.body = body
          self.headers = headers
          self.index = 0
  
      def get_method(self):
          return self.method
  
      def get_context(self):
          return self.context
  
      def get_body(self):
          return self.body
  
      def read(self, size=None):
          v = self.get_body()
          if v is None:
              return None
          else:
              if size is None or size < 0:
                  # read all
                  return v.encode(encoding='utf_8', errors='strict')
              else:
                  bodyLength = len(v)
                  if self.index < 0:
                      return ""
                  if size >= bodyLength:
                      size = bodyLength - 1
                  if size <= 0:
                      return ""
                  endPos = self.index + size
                  tmp = v[self.index:endPos]
                  self.index = endPos
                  if endPos >= (bodyLength - 1):
                      self.index = -1
                  return tmp.encode(encoding='utf_8', errors='strict')
  
      def close(self):
          self.index = 0
  
      def getheaders(self):
          return self.headers
  
  
  [... the rest is omitted ...]
files: astroquery/utils/tap/conn/tests/DummyResponse.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
  
  
  """
  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',
                                  context=None,
                                  body='FINISHED',
                                  headers=None)
      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',
                               context=None,
                               body=jobContent,
                               headers=None)
      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
  from unittest.mock import patch
  
  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, TAP_CLIENT_ID
  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 Column
  
  [... 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
  
  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, TAP_CLIENT_ID
  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',
                                 context=None,
                                 body=tableData,
                                 headers=None)
      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',
                                 context=None,
                                 body=tableData,
                                 headers=None)
      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',
                                 context=None,
                                 body=tableData,
                                 headers=None)
      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', '
  
  [... the rest is omitted ...]
files: astroquery/utils/tap/tests/test_tap.py