files.new_contents: # 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,
  
  [... the rest is omitted ...]
files.new_path: astroquery/esa/jwst/tests/test_jwsttap.py
files.old_contents: # 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_
  
  [... the rest is omitted ...]
files.old_path: astroquery/esa/jwst/tests/test_jwsttap.py
files.new_contents: # 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_
  
  [... the rest is omitted ...]
files.new_path: astroquery/gaia/tests/test_gaiatap.py
files.old_contents: # 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
  
  [... the rest is omitted ...]
files.old_path: astroquery/gaia/tests/test_gaiatap.py
files.new_contents: # 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
  
  [... the rest is omitted ...]
files.new_path: astroquery/utils/tap/model/tests/test_job.py
files.old_contents: # 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
  
  [... the rest is omitted ...]
files.old_path: astroquery/utils/tap/model/tests/test_job.py
files.new_contents: # 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
     
  
  [... the rest is omitted ...]
files.new_path: astroquery/utils/tap/tests/test_tap.py
files.old_contents: # 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
  
  [... the rest is omitted ...]
files.old_path: astroquery/utils/tap/tests/test_tap.py
files.new_contents: # 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
  
  [... the rest is omitted ...]
files.new_path: astroquery/utils/tap/xmlparser/utils.py
files.old_contents: # 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
      """
     
  
  [... the rest is omitted ...]
files.old_path: astroquery/utils/tap/xmlparser/utils.py