# Licensed under a 3-clause BSD style license - see LICENSE.rst

import logging
from pathlib import Path
import numpy as np
import os
import pytest
import json

from requests.models import Response

from astropy.table import Table, unique
from astropy.coordinates import SkyCoord
from astropy.io import fits
import astropy.units as u

from astroquery.mast import Observations, utils, Mast, Catalogs, Hapcut, Tesscut, Zcut, MastMissions

from ..utils import ResolverError
from ...exceptions import (InputWarning, InvalidQueryError, MaxResultsWarning,
                           NoResultsWarning)


@pytest.fixture(scope="module")
def msa_product_table():
    # Pull products for a JWST NIRSpec MSA observation with 6 known
    # duplicates of the MSA configuration file, propID=2736
    obs = Observations.query_criteria(proposal_id=2736,
                                      obs_collection='JWST',
                                      instrument_name='NIRSPEC/MSA',
                                      filters='F170LP;G235M',
                                      target_name='MPTCAT-0628')
    products = Observations.get_product_list(obs['obsid'][0])

    # Filter out everything but the MSA config file
    mask = np.char.find(np.char.asarray(products["dataURI"]), "_msa.fits") != -1
    products = products[mask]

    return products


@pytest.mark.remote_data
class TestMast:

    ###############
    # utils tests #
    ###############

    def test_resolve_object(self):
        m101_loc = utils.resolve_object("M101")
        assert round(m101_loc.separation(SkyCoord("210.80243 54.34875", unit='deg')).value, 4) == 0

        ticobj_loc = utils.resolve_object("TIC 141914082")
        assert round(ticobj_loc.separation(SkyCoord("94.6175354 -72.04484622", unit='deg')).value, 4) == 0

        # Try the same object with different resolvers
        # The position of objects can change with different resolvers
        ned_loc = utils.resolve_object("jw100", resolver="NED")
        assert round(ned_loc.separation(SkyCoord("354.10436 21.15083", unit='deg')).value, 4) == 0

        simbad_loc = utils.resolve_object("jw100", resolver="simbad")
        assert round(simbad_loc.separation(SkyCoord("83.70341477 -5.55918309", unit="deg")).value, 4) == 0

        # Try an object from a MAST catalog with a resolver
        catalog_loc = utils.resolve_object("TIC 307210830", resolver="SIMBAD")
        assert round(catalog_loc.separation(SkyCoord("124.5317560 -68.31300149", unit="deg")).value, 4) == 0

        # Use resolve_all to get all resolvers
        loc_dict = utils.resolve_object("jw100", resolve_all=True)
        assert isinstance(loc_dict, dict)
        assert loc_dict['NED'] == ned_loc
        assert loc_dict['SIMBAD'] == simbad_loc

        # Error if coordinates cannot be resolved
        with pytest.raises(ResolverError, match='Could not resolve "invalid" to a sky position.'):
            utils.resolve_object("invalid")

        # Error if coordinates cannot be resolved with a specific resolver
        with pytest.raises(ResolverError, match='Could not resolve "invalid" to a sky position using resolver "NED"'):
            utils.resolve_object("invalid", resolver="NED")

    ###########################
    # MissionSearchClass Test #
    ###########################

    def test_missions_get_column_list(self):
        columns = MastMissions().get_column_list()
        assert len(columns) > 1
        assert isinstance(columns, Table)
        assert list(columns.columns.keys()) == ['name', 'data_type', 'description']

    def test_missions_query_region_async(self):
        coords = SkyCoord(83.6287, 22.0147, unit="deg")
        response = MastMissions.query_region_async(coords, radius=1)
        assert isinstance(response, Response)
        assert response.status_code == 200

    def test_missions_query_region(self):
        select_cols = ['sci_targname', 'sci_instrume']
        result = MastMissions.query_region("245.89675 -26.52575",
                                           radius=0.1,
                                           sci_instrume=["WFC3", "ACS"],
                                           select_cols=select_cols,
                                           sort_by="sci_data_set_name",
                                           sort_desc=True)
        assert isinstance(result, Table)
        assert len(result) > 0
        assert (result['ang_sep'].data.data.astype('float') < 0.1).all()
        ins_strip = np.char.strip(result['sci_instrume'].data)
        assert ((ins_strip == 'WFC3') | (ins_strip == 'ACS')).all()
        assert all(c in list(result.columns.keys()) for c in select_cols)
        assert list(result['sci_data_set_name']) == sorted(result['sci_data_set_name'], reverse=True)

    def test_missions_query_object_async(self):
        response = MastMissions.query_object_async("M4", radius=0.1)
        assert isinstance(response, Response)
        assert response.status_code == 200

    def test_missions_query_object(self):
        result = MastMissions.query_object("NGC6121",
                                           radius=6*u.arcsec,
                                           sci_pi_last_name='*LE*',
                                           sci_spec_1234='!F395N'
                                           )
        assert isinstance(result, Table)
        assert len(result) > 0
        assert "NGC6121" in result["sci_targname"]
        assert (result['ang_sep'].data.data.astype('float') < 0.1).all()
        assert (result['sci_pi_last_name'] == 'LEE').all()
        assert 'F395N' not in result['sci_spec_1234']

    def test_missions_query_criteria_async(self):
        response = MastMissions.query_criteria_async(sci_pep_id=12557,
                                                     sci_obs_type='SPECTRUM',
                                                     sci_aec='S')
        assert isinstance(response, Response)
        assert response.status_code == 200

    def test_missions_query_criteria(self):
        # Non-positional search
        with pytest.warns(MaxResultsWarning):
            result = MastMissions.query_criteria(sci_pep_id=12557,
                                                 sci_obs_type='SPECTRUM',
                                                 sci_aec='S',
                                                 limit=3,
                                                 select_cols=['sci_pep_id', 'sci_obs_type', 'sci_aec'])
        assert isinstance(result, Table)
        assert len(result) == 3
        assert (result['sci_pep_id'] == 12557).all()
        assert (result['sci_obs_type'] == 'SPECTRUM').all()
        assert (result['sci_aec'] == 'S').all()

        # Positional criteria search
        result = MastMissions.query_criteria(object_names='NGC6121',
                                             radius=0.1,
                                             sci_start_time='<2012',
                                             sci_actual_duration='0..200',
                                             select_cols='*')
        assert len(result) == 3
        assert result.colnames == MastMissions.get_column_list()['name'].value.tolist()
        assert (result['ang_sep'].data.data.astype('float') < 0.1).all()
        assert (result['sci_start_time'] < '2012').all()
        assert ((result['sci_actual_duration'] >= 0) & (result['sci_actual_duration'] <= 200)).all()

        # Search with multiple positional inputs
        coord = SkyCoord(245.89675, -26.52575, unit='deg')
        result = MastMissions.query_criteria(coordinates=[coord, "205.54842 28.37728"],
                                             object_names=["M2", "M9"],
                                             radius=0.1)
        assert len(set(result['search_pos'])) == 4  # Should have four different search positions

        # Raise error if invalid input is given
        with pytest.raises(InvalidQueryError):
            MastMissions.query_criteria(coordinates="245.89675 -26.52575",
                                        radius=1,
                                        sci_pep_id="invalid")

    def test_missions_query_criteria_invalid_keyword(self):
        # Attempt to make a criteria query with invalid keyword
        with pytest.raises(InvalidQueryError) as err_no_alt:
            MastMissions.query_criteria(select_cols=['sci_targname'],
                                        not_a_keyword='test')
        assert "Filter 'not_a_keyword' does not exist." in str(err_no_alt.value)

        # Attempt to make a region query with invalid keyword
        with pytest.raises(InvalidQueryError) as err_no_alt:
            MastMissions.query_region(coordinates="245.89675 -26.52575",
                                      invalid_keyword='test')
        assert "Filter 'invalid_keyword' does not exist." in str(err_no_alt.value)

        # Keyword is close enough for difflib to offer alternative
        with pytest.raises(InvalidQueryError) as err_with_alt:
            MastMissions.query_criteria(search_position='30 30')
        assert 'search_pos' in str(err_with_alt.value)

        # Should be case sensitive
        with pytest.raises(InvalidQueryError) as err_case:
            MastMissions.query_criteria(Search_Pos='30 30')
        assert 'search_pos' in str(err_case.value)

    def test_missions_get_product_list_async(self):
        datasets = MastMissions.query_object("M4", radius=0.1)

        # Table as input
        responses = MastMissions.get_product_list_async(datasets[:3])
        assert isinstance(responses, list)

        # Row as input
        responses = MastMissions.get_product_list_async(datasets[0])
        assert isinstance(responses, list)

        # String as input
        responses = MastMissions.get_product_list_async(datasets[0]['sci_data_set_name'])
        assert isinstance(responses, list)

        # Column as input
        responses = MastMissions.get_product_list_async(datasets[:3]['sci_data_set_name'])
        assert isinstance(responses, list)

        # Batching
        responses = MastMissions.get_product_list_async(datasets[:4], batch_size=2)
        assert isinstance(responses, list)
        assert len(responses) == 2
        assert isinstance(responses[0], Response)

        # Unsupported data type for datasets
        with pytest.raises(TypeError) as err_type:
            MastMissions.get_product_list_async(1)
        assert 'Unsupported data type' in str(err_type.value)

        # Empty dataset list
        with pytest.raises(InvalidQueryError) as err_empty:
            MastMissions.get_product_list_async([' '])
        assert 'Dataset list is empty' in str(err_empty.value)

    def test_missions_get_product_list(self, capsys):
        datasets = MastMissions.query_object("M4", radius=0.1)
        test_dataset = datasets[0]['sci_data_set_name']
        multi_dataset = list(datasets[:2]['sci_data_set_name'])

        # Compare Row input and string input
        result_str = MastMissions.get_product_list(test_dataset)
        result_row = MastMissions.get_product_list(datasets[0])
        assert isinstance(result_str, Table)
        assert len(result_str) == len(result_row)
        assert set(result_str['filename']) == set(result_row['filename'])

        # Compare Table input and list input
        result_list = MastMissions.get_product_list(multi_dataset)
        result_table = MastMissions.get_product_list(datasets[:2])
        assert isinstance(result_list, Table)
        assert len(result_list) == len(result_table)
        assert set(result_list['filename']) == set(result_table['filename'])

        # Filter datasets based on sci_data_set_name and verify products
        filtered = datasets[datasets['sci_data_set_name'] == 'IBKH03020']
        result = MastMissions.get_product_list(filtered)
        assert isinstance(result, Table)
        assert (result['dataset'] == 'IBKH03020').all()

        # Test batching
        result_batch = MastMissions.get_product_list(datasets[:2], batch_size=1)
        out, _ = capsys.readouterr()
        assert isinstance(result_batch, Table)
        assert len(result_batch) == len(result_table)
        assert set(result_batch['filename']) == set(result_table['filename'])
        assert 'Fetching products for 2 unique datasets in 2 batches' in out

    def test_missions_get_unique_product_list(self, caplog):
        # Check that no rows are filtered out when all products are unique
        dataset_ids = ['JBTAA8010']
        products = MastMissions.get_product_list(dataset_ids)
        unique_products = MastMissions.get_unique_product_list(dataset_ids)

        # Should have the same length
        assert len(products) == len(unique_products)
        # No INFO messages should be logged
        with caplog.at_level('INFO', logger='astroquery'):
            assert caplog.text == ''

        # Check that rows are filtered out when products are not unique
        dataset_ids.append('JBTAA8020')
        products = MastMissions.get_product_list(dataset_ids)
        unique_products = MastMissions.get_unique_product_list(dataset_ids)

        # Unique product list should have fewer rows
        assert len(products) > len(unique_products)
        # Rows should be unique based on filename
        assert (len(unique_products) == len(unique(unique_products, keys='filename')))
        # Check that INFO messages were logged
        with caplog.at_level('INFO', logger='astroquery'):
            assert 'products were duplicates' in caplog.text
            assert 'To return all products' in caplog.text

    def test_missions_filter_products(self):
        # Filter by extension
        products = MastMissions.get_product_list('W0FX0301T')
        filtered = MastMissions.filter_products(products,
                                                extension='jpg')
        assert isinstance(filtered, Table)
        assert all(filename.endswith('.jpg') for filename in filtered['filename'])

        # Filter by existing column
        filtered = MastMissions.filter_products(products,
                                                category='CALIBRATED')
        assert isinstance(filtered, Table)
        assert all(filtered['category'] == 'CALIBRATED')

    def test_missions_download_products(self, tmp_path):
        def check_filepath(path):
            assert path.is_file()

        # Check string input
        test_dataset_id = 'Z14Z0104T'
        result = MastMissions.download_products(test_dataset_id,
                                                download_dir=tmp_path)
        for row in result:
            if row['Status'] == 'COMPLETE':
                check_filepath(row['Local Path'])

        # Check Row input
        datasets = MastMissions.query_object("M4", radius=0.1)
        prods = MastMissions.get_product_list(datasets[0])[0]
        result = MastMissions.download_products(prods,
                                                download_dir=tmp_path)
        check_filepath(result['Local Path'][0])

        # JSON data input
        json_data = [{'fileset': 'Z14Z0104T',
                      'filename': 'z14z0104t_pdq.fits'}]
        result = MastMissions.download_products(json_data,
                                                download_dir=tmp_path)
        check_filepath(result['Local Path'][0])

        # JSON file input
        json_file = tmp_path / 'products.json'
        json_file.write_text(json.dumps(json_data))
        result = MastMissions.download_products(json_file,
                                                download_dir=tmp_path)
        check_filepath(result['Local Path'][0])

        # Warn about no products
        with pytest.warns(NoResultsWarning):
            result = MastMissions.download_products(test_dataset_id,
                                                    extension='jpg',
                                                    download_dir=tmp_path)

    def test_missions_download_products_flat(self, tmp_path):
        # Download products without creating subdirectories
        result = MastMissions.download_products('Z14Z0104T',
                                                flat=True,
                                                download_dir=tmp_path)
        for row in result:
            if row['Status'] == 'COMPLETE':
                assert row['Local Path'].parent == tmp_path

    def test_missions_download_file(self, tmp_path):
        def check_result(result, path):
            assert result == ('COMPLETE', None, None)
            assert path.is_file()

        # Get URI from data product
        product = MastMissions.get_product_list('Z14Z0104T')[0]
        uri = product['uri']
        filename = Path(uri).name

        # Download with unspecified local_path
        # Should download to current working directory
        result = MastMissions.download_file(uri)
        check_result(result, Path(os.getcwd(), filename))
        Path.unlink(filename)  # clean up file

        # Download with directory as local_path parameter
        local_path = Path(tmp_path, filename)
        result = MastMissions.download_file(uri, local_path=tmp_path)
        check_result(result, local_path)

        # Download with filename as local_path parameter
        local_path_file = Path(tmp_path, 'test.fits')
        result = MastMissions.download_file(uri, local_path=local_path_file)
        check_result(result, local_path_file)

    @pytest.mark.parametrize("mission, query_params", [
        ('jwst', {'fileSetName': 'jw01189001001_02101_00001'}),
        ('classy', {'Target': 'J0021+0052'}),
        ('ullyses', {'host_galaxy_name': 'WLM', 'select_cols': ['observation_id']}),
        ('roman', {'program': 3}),
        ('iue', {'iue_data_id': 'LWR08496'}),
    ])
    def test_missions_workflow(self, tmp_path, mission, query_params):
        # Test workflow with other missions
        m = MastMissions(mission=mission)

        # Roman requires extra setup to point towards the test server
        if mission == 'roman':
            m._service_api_connection.SERVICE_URL = 'https://masttest.stsci.edu'
            m._service_api_connection.REQUEST_URL = 'https://masttest.stsci.edu/search/roman/api/v0.1/'

        # Criteria query
        datasets = m.query_criteria(**query_params)
        assert isinstance(datasets, Table)
        assert len(datasets)

        # Get products
        prods = m.get_product_list(datasets[0])
        assert isinstance(prods, Table)
        assert len(prods)

        # Download products
        result = m.download_products(prods[:3],
                                     download_dir=tmp_path)
        for row in result:
            if row['Status'] == 'COMPLETE':
                assert (row['Local Path']).is_file()

    ###################
    # MastClass tests #
    ###################

    def test_mast_service_request_async(self):
        service = 'Mast.Caom.Cone'
        params = {'ra': 184.3,
                  'dec': 54.5,
                  'radius': 0.001}
        responses = Mast.service_request_async(service, params)

        assert isinstance(responses, list)

    def test_mast_service_request(self):
        service = 'Mast.Caom.Cone'
        params = {'ra': 184.3,
                  'dec': 54.5,
                  'radius': 0.001}

        result = Mast.service_request(service, params, pagesize=10, page=2)

        # Is result in the right format
        assert isinstance(result, Table)

        # Is result limited to ten rows
        assert len(result) == 10

    def test_mast_query(self):
        # Cone search (unfiltered)
        result = Mast.mast_query('Mast.Caom.Cone', ra=184.3, dec=54.5, radius=0.005)
        assert isinstance(result, Table)
        assert "GALEX" in result['obs_collection']
        assert len(result[np.where(result["obs_id"] == "6374399093149532160")]) == 2

        # Filtered query
        columns = ['target_name', 'obs_collection', 'calib_level', 'sequence_number', 't_min']
        result = Mast.mast_query('Mast.Caom.Filtered',
                                 target_name=375422201,
                                 obs_collection={'TESS'},
                                 calib_level=np.asarray(3),
                                 sequence_number=[15, 16],
                                 t_min={'min': 58710, 'max': 58720},
                                 columns=columns)
        assert isinstance(result, Table)
        assert all(result['target_name'] == '375422201')
        assert all(result['obs_collection'] == 'TESS')
        assert all(result['calib_level'] == 3)
        assert all((result['sequence_number'] == 15) | (result['sequence_number'] == 16))
        assert (result['t_min'] >= 58710).all() and (result['t_min'] <= 58720).all()
        assert all(c in list(result.columns.keys()) for c in columns)
        assert len(result.columns) == 5

    def test_mast_session_info(self):
        sessionInfo = Mast.session_info(verbose=False)
        assert sessionInfo['ezid'] == 'anonymous'
        assert sessionInfo['token'] is None

    ###########################
    # ObservationsClass tests #
    ###########################

    def test_observations_list_missions(self):
        missions = Observations.list_missions()
        assert isinstance(missions, list)
        for m in ['HST', 'HLA', 'GALEX', 'Kepler']:
            assert m in missions

    def test_get_metadata(self):
        # observations
        meta_table = Observations.get_metadata("observations")
        assert isinstance(meta_table, Table)
        assert "Column Name" in meta_table.colnames
        assert "Mission" in meta_table["Column Label"]
        assert "obsid" in meta_table["Column Name"]

        # products
        meta_table = Observations.get_metadata("products")
        assert isinstance(meta_table, Table)
        assert "Column Name" in meta_table.colnames
        assert "Observation ID" in meta_table["Column Label"]
        assert "parent_obsid" in meta_table["Column Name"]

    # query functions

    def test_observations_query_region_async(self):
        responses = Observations.query_region_async("322.49324 12.16683", radius="0.005 deg")
        assert isinstance(responses, list)

    def test_observations_query_region(self):
        result = Observations.query_region("322.49324 12.16683", radius="0.005 deg")
        assert isinstance(result, Table)
        assert len(result) > 500
        assert result[np.where(result['obs_id'] == '00031992001')]

        result = Observations.query_region("322.49324 12.16683", radius="0.005 deg",
                                           pagesize=1, page=1)
        assert isinstance(result, Table)
        assert len(result) == 1

    def test_observations_query_object_async(self):
        responses = Observations.query_object_async("M8", radius=".02 deg")
        assert isinstance(responses, list)

    def test_observations_query_object(self):
        result = Observations.query_object("M8", radius=".04 deg")
        assert isinstance(result, Table)
        assert len(result) > 150
        assert result[np.where(result['obs_id'] == 'ktwo200071160-c92_lc')]

    def test_observations_query_criteria_async(self):
        # without position
        responses = Observations.query_criteria_async(dataproduct_type=["image"],
                                                      proposal_pi="*Ost*",
                                                      s_dec=[43.5, 43.6])
        assert isinstance(responses, list)

        # with position
        responses = Observations.query_criteria_async(filters=["NUV", "FUV"],
                                                      object_name="M10")
        assert isinstance(responses, list)

    def test_observations_query_criteria(self):
        # without position
        result = Observations.query_criteria(instrument_name="*WFPC2*",
                                             proposal_id=8169,
                                             t_min=[51361, 51362])

        assert isinstance(result, Table)
        assert len(result) == 13
        assert ((result['obs_collection'] == 'HST') | (result['obs_collection'] == 'HLA')).all()

        # with position
        result = Observations.query_criteria(object_name="M10",
                                             filters=["NUV", "FUV"],
                                             obs_collection="GALEX")
        assert isinstance(result, Table)
        assert len(result) == 7
        assert (result['obs_collection'] == 'GALEX').all()
        assert sum(result['filters'] == 'NUV') == 4

        result = Observations.query_criteria(object_name="M10",
                                             dataproduct_type="IMAGE",
                                             intentType="calibration")
        assert (result["intentType"] == "calibration").all()

        # with case-insensitive keyword arguments
        result = Observations.query_criteria(Instrument_Name="*WFPC2*",
                                             proposal_ID=8169,
                                             T_min=[51361, 51362])
        assert isinstance(result, Table)
        assert len(result) == 13

    def test_observations_query_criteria_invalid_keyword(self):
        # attempt to make a criteria query with invalid keyword
        with pytest.raises(InvalidQueryError) as err_no_alt:
            Observations.query_criteria_count(not_a_keyword='TESS')
        assert "Filter 'not_a_keyword' does not exist." in str(err_no_alt.value)

        # keyword is close enough for difflib to offer alternative
        with pytest.raises(InvalidQueryError) as err_with_alt:
            Observations.query_criteria_count(oops_collection='TESS')
        assert 'obs_collection' in str(err_with_alt.value)

    # count functions
    def test_observations_query_region_count(self):
        maxRes = Observations.query_criteria_count()
        result = Observations.query_region_count("322.49324 12.16683",
                                                 radius=1e-10 * u.deg)
        assert isinstance(result, (np.int64, int))
        assert result > 840
        assert result < maxRes

    def test_observations_query_object_count(self):
        maxRes = Observations.query_criteria_count()
        result = Observations.query_object_count("M8", radius=".02 deg")
        assert isinstance(result, (np.int64, int))
        assert result > 150
        assert result < maxRes

    def test_observations_query_criteria_count(self):
        maxRes = Observations.query_criteria_count()
        result = Observations.query_criteria_count(proposal_id=8880)
        assert isinstance(result, (np.int64, int))
        assert result == 8
        assert result < maxRes

    # product functions
    def test_observations_get_product_list_async(self):

        test_obs = Observations.query_criteria(filters=["NUV", "FUV"], object_name="M10")

        responses = Observations.get_product_list_async(test_obs[0]["obsid"])
        assert isinstance(responses, list)

        responses = Observations.get_product_list_async(test_obs[2:3])
        assert isinstance(responses, list)

        observations = Observations.query_criteria(object_name="M8", obs_collection=["K2", "IUE"])
        responses = Observations.get_product_list_async(observations[0])
        assert isinstance(responses, list)

        responses = Observations.get_product_list_async(observations[0:4])
        assert isinstance(responses, list)

        # Batching
        responses = Observations.get_product_list_async(observations[0:4], batch_size=2)
        assert isinstance(responses, list)

    def test_observations_get_product_list(self, capsys):
        observations = Observations.query_criteria(object_name='M8', obs_collection=['K2', 'IUE'])
        test_obs_id = str(observations[0]['obsid'])
        mult_obs_ids = str(observations[0]['obsid']) + ',' + str(observations[1]['obsid'])

        result1 = Observations.get_product_list(test_obs_id)
        result2 = Observations.get_product_list(observations[0])
        filenames1 = list(result1['productFilename'])
        filenames2 = list(result2['productFilename'])
        assert isinstance(result1, Table)
        assert len(result1) == len(result2)
        assert set(filenames1) == set(filenames2)

        result1 = Observations.get_product_list(mult_obs_ids)
        result2 = Observations.get_product_list(observations[0:2])
        filenames1 = result1['productFilename']
        filenames2 = result2['productFilename']
        assert isinstance(result1, Table)
        assert len(result1) == len(result2)
        assert set(filenames1) == set(filenames2)

        obsLoc = np.where(observations['obs_id'] == 'ktwo200071160-c92_lc')
        result = Observations.get_product_list(observations[obsLoc])
        assert isinstance(result, Table)
        assert len(result) == 1

        obsLocs = np.where((observations['target_name'] == 'NGC6523') & (observations['obs_collection'] == "IUE"))
        result = Observations.get_product_list(observations[obsLocs])
        obs_collection = np.unique(list(result['obs_collection']))
        assert isinstance(result, Table)
        assert len(obs_collection) == 1
        assert obs_collection[0] == 'IUE'

        # Test batching
        result_batch = Observations.get_product_list(observations[:2], batch_size=1)
        out, _ = capsys.readouterr()
        assert isinstance(result_batch, Table)
        assert len(result_batch) == len(result1)
        assert set(result_batch['productFilename']) == set(filenames1)
        assert 'Fetching products for 2 unique observations in 2 batches' in out

    def test_observations_get_product_list_tess_tica(self, caplog):
        # Get observations and products with both TESS and TICA FFIs
        obs = Observations.query_criteria(target_name=['TESS FFI', 'TICA FFI', '429031146'])
        prods = Observations.get_product_list(obs)

        # Check that WARNING messages about FFIs were logged
        with caplog.at_level("WARNING", logger="astroquery"):
            assert "TESS FFI products" in caplog.text
            assert "TICA FFI products" in caplog.text

        # Should only return products corresponding to target 429031146
        assert len(prods) > 0
        assert (np.char.find(np.char.asarray(prods['obs_id']), '429031146') != -1).all()

    def test_observations_get_unique_product_list(self, caplog):
        # Check that no rows are filtered out when all products are unique
        obsids = ['24832668']
        products = Observations.get_product_list(obsids)
        unique_products = Observations.get_unique_product_list(obsids)

        # Should have the same length
        assert len(products) == len(unique_products)
        # No INFO messages should be logged
        with caplog.at_level('INFO', logger='astroquery'):
            assert caplog.text == ''

        # Check that rows are filtered out when products are not unique
        obsids.append('26421364')
        products = Observations.get_product_list(obsids)
        unique_products = Observations.get_unique_product_list(obsids)

        # Unique product list should have fewer rows
        assert len(products) > len(unique_products)
        # Rows should be unique based on dataURI
        assert (len(unique_products) == len(unique(unique_products, keys='dataURI')))
        # Check that INFO messages were logged
        with caplog.at_level('INFO', logger='astroquery'):
            assert 'products were duplicates' in caplog.text
            assert 'To return all products' in caplog.text

    def test_observations_filter_products(self):
        observations = Observations.query_object("M8", radius=".04 deg")
        obsLoc = np.where(observations["obs_id"] == 'ktwo200071160-c92_lc')
        products = Observations.get_product_list(observations[obsLoc])
        result = Observations.filter_products(products,
                                              productType=["SCIENCE"],
                                              mrp_only=False)
        assert isinstance(result, Table)
        assert len(result) == sum(products['productType'] == "SCIENCE")

    def test_observations_download_products(self, tmp_path):
        def check_filepath(path):
            assert os.path.isfile(path)

        test_obs_id = '25119363'
        result = Observations.download_products(test_obs_id,
                                                download_dir=tmp_path,
                                                productType=["SCIENCE"],
                                                mrp_only=False)
        assert isinstance(result, Table)
        for row in result:
            if row['Status'] == 'COMPLETE':
                check_filepath(row['Local Path'])

        # just get the curl script
        result = Observations.download_products(test_obs_id,
                                                download_dir=tmp_path,
                                                curl_flag=True,
                                                productType=["SCIENCE"],
                                                mrp_only=False)
        assert isinstance(result, Table)
        check_filepath(result['Local Path'][0])

        # check for row input
        result1 = Observations.get_product_list(test_obs_id)
        result2 = Observations.download_products(result1[0],
                                                 download_dir=tmp_path)
        assert isinstance(result2, Table)
        check_filepath(result2['Local Path'][0])
        assert len(result2) == 1

    def test_observations_download_products_flat(self, tmp_path, msa_product_table):

        # Get a product list with 6 duplicate JWST MSA config files
        products = msa_product_table

        assert len(products) == 6

        # Download with flat=True
        manifest = Observations.download_products(products, flat=True, download_dir=tmp_path)

        assert Path(manifest["Local Path"][0]).parent == tmp_path

    def test_observations_download_products_flat_curl(self, tmp_path, msa_product_table):

        # Get a product list with 6 duplicate JWST MSA config files
        products = msa_product_table

        assert len(products) == 6

        # Download with flat=True, curl_flag=True, look for warning
        with pytest.warns(InputWarning):
            Observations.download_products(products, flat=True,
                                           curl_flag=True,
                                           download_dir=tmp_path)

    def test_observations_download_products_no_duplicates(self, tmp_path, caplog, msa_product_table):

        # Get a product list with 6 duplicate JWST MSA config files
        products = msa_product_table

        assert len(products) == 6

        # Download the product
        manifest = Observations.download_products(products, download_dir=tmp_path)

        # Check that it downloads the MSA config file only once
        assert len(manifest) == 1

        # Check that an INFO message about duplicates was logged
        with caplog.at_level("INFO", logger="astroquery"):
            assert "products were duplicates" in caplog.text

    def test_observations_download_products_cloud(self, tmp_path, msa_product_table):
        pytest.importorskip('boto3')

        Observations.enable_cloud_dataset()

        # Adding a product that's not in the cloud to test mixed downloads
        in_uri = "mast:IUE/url/pub/vospectra/iue2/swp18830mxlo_vo.fits"
        new_row = {col: msa_product_table[col][0] for col in msa_product_table.colnames}
        new_row["dataURI"] = in_uri
        new_row["productFilename"] = Path(in_uri).name
        msa_product_table = msa_product_table.copy()
        msa_product_table.add_row(new_row)

        with pytest.warns(NoResultsWarning, match='Skipping download'):
            result = Observations.download_products(msa_product_table,
                                                    download_dir=tmp_path,
                                                    cloud_only=True)
        assert isinstance(result, Table)
        assert len(result) == 2
        # First product should be downloaded from cloud, second should be skipped
        assert result['Status'][0] == 'COMPLETE'
        assert result['Status'][1] == 'SKIPPED'
        assert Path(result['Local Path'][0]).exists()
        assert not Path(result['Local Path'][1]).exists()
        Path.unlink(result['Local Path'][0])  # clean up file

        # Should fall back and download if cloud_only is False
        with pytest.warns(InputWarning, match='Falling back to MAST download.'):
            result = Observations.download_products(msa_product_table,
                                                    download_dir=tmp_path)
        assert isinstance(result, Table)
        assert len(result) == 2
        assert result['Status'][0] == 'COMPLETE'
        assert result['Status'][1] == 'COMPLETE'
        assert Path(result['Local Path'][0]).exists()
        assert Path(result['Local Path'][1]).exists()

        Observations.disable_cloud_dataset()

    def test_observations_download_file(self, tmp_path):

        def check_result(result, path):
            assert result == ('COMPLETE', None, None)
            assert os.path.exists(path)

        # get observations from GALEX instrument with query_criteria
        observations = Observations.query_criteria(object_name='M10',
                                                   radius=0.001,
                                                   instrument_name='GALEX')

        assert len(observations) > 0, 'No results found for GALEX query.'

        # pull data products from a single observation
        products = Observations.get_product_list(observations['obsid'][0])

        # pull the URI of a single product
        uri = products['dataURI'][0]
        filename = Path(uri).name

        # download with unspecified local_path parameter
        # should download to current working directory
        result = Observations.download_file(uri)
        check_result(result, Path(os.getcwd(), filename))
        Path.unlink(filename)  # clean up file

        # download with directory as local_path parameter
        local_path = Path(tmp_path, filename)
        result = Observations.download_file(uri, local_path=tmp_path)
        check_result(result, local_path)

        # download with filename as local_path parameter
        local_path_file = Path(tmp_path, "test.fits")
        result = Observations.download_file(uri, local_path=local_path_file)
        check_result(result, local_path_file)

    @pytest.mark.parametrize("in_uri", [
        'mast:GALEX/url/data/GR6/pipe/01-vsn/03329-MISDR1_18916_0459/d/01-main/0001-img/07-try/'
        'MISDR1_18916_0459-fd-flagstar.fits.gz',
        'mast:HST/product/u24r0102t_c3m.fits'
    ])
    def test_observations_download_file_cloud(self, tmp_path, in_uri):
        pytest.importorskip("boto3")

        Observations.enable_cloud_dataset()

        filename = Path(in_uri).name
        result = Observations.download_file(uri=in_uri, cloud_only=True, local_path=tmp_path)
        assert result == ('COMPLETE', None, None)
        assert Path(tmp_path, filename).exists()

        Observations.disable_cloud_dataset()

    def test_observations_download_file_cloud_not_found(self, tmp_path):
        pytest.importorskip("boto3")
        in_uri = 'mast:IUE/url/pub/vospectra/iue2/swp18830mxlo_vo.fits'

        Observations.enable_cloud_dataset()

        # Warn and fallback
        with pytest.warns(InputWarning, match='Falling back to MAST download.'):
            result = Observations.download_file(uri=in_uri, local_path=tmp_path)
            assert result == ('COMPLETE', None, None)
            assert Path(tmp_path, Path(in_uri).name).exists()
            Path.unlink(Path(tmp_path, Path(in_uri).name))  # clean up file

        # Skip if cloud_only is set
        with pytest.warns(NoResultsWarning, match='Skipping download'):
            result = Observations.download_file(uri=in_uri, cloud_only=True, local_path=tmp_path)
            assert result == ('SKIPPED', None, None)
            assert not Path(tmp_path, Path(in_uri).name).exists()

        Observations.disable_cloud_dataset()

    def test_observations_download_file_escaped(self, tmp_path):
        # test that `download_file` correctly escapes a URI
        in_uri = 'mast:HLA/url/cgi-bin/fitscut.cgi?' \
                 'red=hst_04819_65_wfpc2_f814w_pc&blue=hst_04819_65_wfpc2_f555w_pc&size=ALL&format=fits'
        filename = Path(in_uri).name
        result = Observations.download_file(uri=in_uri, local_path=tmp_path)
        assert result == ('COMPLETE', None, None)
        assert Path(tmp_path, filename).exists()

        # check that downloaded file is a valid FITS file
        f = fits.open(Path(tmp_path, filename))
        f.close()

    def test_observations_download_file_no_length(self, tmp_path, caplog):
        # test that `download_file` correctly handles the case where the server
        # does not return a Content-Length header for a cached file
        # initial download
        in_uri = "mast:HLA/url/cgi-bin/getdata.cgi?filename=hst_05206_01_wfpc2_f375n_wf_daophot_trm.cat"
        filename = Path(in_uri).name
        result = Observations.download_file(uri=in_uri, local_path=tmp_path)
        assert result == ("COMPLETE", None, None)
        assert Path(tmp_path, filename).exists()

        # download again, should warn and re-download file
        with caplog.at_level(logging.WARNING):
            result = Observations.download_file(uri=in_uri, local_path=tmp_path)
        assert "Could not verify length of cached file" in caplog.text
        assert result == ("COMPLETE", None, None)
        assert Path(tmp_path, filename).exists()

    def test_observations_list_cloud_missions(self):
        pytest.importorskip('boto3')
        Observations.enable_cloud_dataset()
        missions = Observations.list_cloud_datasets()
        assert isinstance(missions, list)
        assert len(missions) > 0
        for m in ['hst', 'jwst', 'panstarrs', 'galex', 'tess']:
            assert m in missions
        Observations.disable_cloud_dataset()

    @pytest.mark.parametrize("test_data_uri, expected_cloud_uri", [
        ("mast:HST/product/u24r0102t_c1f.fits",
         "s3://stpubdata/hst/public/u24r/u24r0102t/u24r0102t_c1f.fits"),
        ("mast:PS1/product/rings.v3.skycell.1334.061.stk.r.unconv.exp.fits",
         "s3://stpubdata/panstarrs/ps1/public/rings.v3.skycell/1334/061/"
         "rings.v3.skycell.1334.061.stk.r.unconv.exp.fits")
    ])
    def test_observations_get_cloud_uri(self, test_data_uri, expected_cloud_uri):
        pytest.importorskip("boto3")
        # get a product list
        product = Table()
        product['dataURI'] = [test_data_uri]
        # enable access to public AWS S3 bucket
        Observations.enable_cloud_dataset()

        # get uri
        uri = Observations.get_cloud_uri(product[0])

        assert len(uri) > 0, f'Product for dataURI {test_data_uri} was not found in the cloud.'
        assert uri == expected_cloud_uri, f'Cloud URI does not match expected. ({uri} != {expected_cloud_uri})'

        # pass the URI as a string
        uri = Observations.get_cloud_uri(test_data_uri)
        assert uri == expected_cloud_uri, f'Cloud URI does not match expected. ({uri} != {expected_cloud_uri})'

        Observations.disable_cloud_dataset()

    @pytest.mark.parametrize("test_obs_id", ["25568122", "31411", "107604081"])
    def test_observations_get_cloud_uris(self, test_obs_id):
        pytest.importorskip("boto3")

        # get a product list
        index = 24 if test_obs_id == '25568122' else 0
        products = Observations.get_product_list(test_obs_id)[index:index + 2]

        assert len(products) > 0, (f'No products found for OBSID {test_obs_id}. '
                                   'Unable to move forward with getting URIs from the cloud.')

        # enable access to public AWS S3 bucket
        Observations.enable_cloud_dataset()

        # get uris
        uris = Observations.get_cloud_uris(products)

        assert len(uris) > 0, f'Products for OBSID {test_obs_id} were not found in the cloud.'

        # check for warning if no data products match filters
        with pytest.warns(NoResultsWarning):
            Observations.get_cloud_uris(products,
                                        extension='png')

        Observations.disable_cloud_dataset()

    def test_observations_get_cloud_uris_list_input(self):
        pytest.importorskip("boto3")
        uri_list = ['mast:HST/product/u24r0102t_c1f.fits',
                    'mast:PS1/product/rings.v3.skycell.1334.061.stk.r.unconv.exp.fits']
        expected = ['s3://stpubdata/hst/public/u24r/u24r0102t/u24r0102t_c1f.fits',
                    's3://stpubdata/panstarrs/ps1/public/rings.v3.skycell/1334/061/rings.v3.skycell.1334.'
                    '061.stk.r.unconv.exp.fits']

        # enable access to public AWS S3 bucket
        Observations.enable_cloud_dataset()

        # list of URI strings as input
        uris = Observations.get_cloud_uris(uri_list)
        assert len(uris) > 0, f'Products for URI list {uri_list} were not found in the cloud.'
        assert uris == expected

        # return map of dataURI to cloud URI
        uri_map = Observations.get_cloud_uris(uri_list, return_uri_map=True)
        assert isinstance(uri_map, dict)
        assert len(uri_map) == 2
        for i, uri in enumerate(uri_list):
            assert uri_map[uri] == expected[i]

        # check for warning if filters are provided with list input
        with pytest.warns(InputWarning, match='Filtering is not supported'):
            Observations.get_cloud_uris(uri_list,
                                        extension='png')

        # check for warning if one of the URIs is not found
        with pytest.warns(NoResultsWarning, match='Failed to retrieve cloud path'):
            Observations.get_cloud_uris(['mast:HST/product/does_not_exist.fits'])

        Observations.disable_cloud_dataset()

    def test_observations_get_cloud_uris_query(self):
        pytest.importorskip("boto3")

        # enable access to public AWS S3 bucket
        Observations.enable_cloud_dataset()

        # get uris with other functions
        obs = Observations.query_criteria(target_name=234295610)
        prod = Observations.get_product_list(obs)
        filt = Observations.filter_products(prod, calib_level=[2])
        s3_uris = Observations.get_cloud_uris(filt)

        # get uris with streamlined function
        uris = Observations.get_cloud_uris(target_name=234295610,
                                           filter_products={'calib_level': [2]})
        assert s3_uris == uris

        # check that InvalidQueryError is thrown if neither data_products or **criteria are defined
        with pytest.raises(InvalidQueryError):
            Observations.get_cloud_uris(filter_products={'calib_level': [2]})

        # check for warning if query returns no observations
        with pytest.warns(NoResultsWarning):
            Observations.get_cloud_uris(target_name=234295611)

        Observations.disable_cloud_dataset()

    def test_observations_get_cloud_uris_no_duplicates(self, msa_product_table):
        pytest.importorskip("boto3")

        # Get a product list with 6 duplicate JWST MSA config files
        products = msa_product_table

        assert len(products) == 6

        # enable access to public AWS S3 bucket
        Observations.enable_cloud_dataset(provider='AWS')

        # Check that only one URI is returned
        uris = Observations.get_cloud_uris(products)
        assert len(uris) == 1

        Observations.disable_cloud_dataset()

    ######################
    # CatalogClass tests #
    ######################

    # query functions
    def test_catalogs_query_region_async(self):
        in_rad = 0.001 * u.deg
        responses = Catalogs.query_region_async("158.47924 -7.30962",
                                                radius=in_rad,
                                                catalog="Galex")
        assert isinstance(responses, list)

        # Default catalog is HSC
        responses = Catalogs.query_region_async("322.49324 12.16683",
                                                radius=in_rad)
        assert isinstance(responses, list)

        responses = Catalogs.query_region_async("322.49324 12.16683",
                                                radius=in_rad,
                                                catalog="panstarrs",
                                                table="mean")
        assert isinstance(responses, Response)

    def test_catalogs_query_region(self):
        def check_result(result, row, exp_values):
            assert isinstance(result, Table)
            for k, v in exp_values.items():
                assert result[row][k] == v

        in_radius = 0.1 * u.deg
        result = Catalogs.query_region("158.47924 -7.30962",
                                       radius=in_radius,
                                       catalog="Gaia")
        row = np.where(result['source_id'] == '3774902350511581696')
        check_result(result, row, {'solution_id': '1635721458409799680'})

        result = Catalogs.query_region("322.49324 12.16683",
                                       radius=0.001*u.deg,
                                       catalog="HSC",
                                       magtype=2)
        row = np.where(result['MatchID'] == '8150896')

        with pytest.warns(MaxResultsWarning):
            result = Catalogs.query_region("322.49324 12.16683", catalog="HSC", magtype=2, nr=5)

        check_result(result, row, {'NumImages': 14, 'TargetName': 'M15'})

        result = Catalogs.query_region("322.49324 12.16683",
                                       radius=0.001*u.deg,
                                       catalog="HSC",
                                       version=2,
                                       magtype=2)
        row = np.where(result['MatchID'] == '82361658')
        check_result(result, row, {'NumImages': 11, 'TargetName': 'NGC7078'})

        result = Catalogs.query_region("322.49324 12.16683",
                                       radius=in_radius,
                                       catalog="Gaia",
                                       version=1)
        row = np.where(result['source_id'] == '1745948323734098688')
        check_result(result, row, {'solution_id': '1635378410781933568'})
        result = Catalogs.query_region("322.49324 12.16683",
                                       radius=0.01*u.deg,
                                       catalog="Gaia",
                                       version=2)

        row = np.where(result['source_id'] == '1745947739618544000')
        check_result(result, row, {'solution_id': '1635721458409799680'})

        result = Catalogs.query_region("322.49324 12.16683",
                                       radius=0.01*u.deg, catalog="panstarrs",
                                       table="mean",
                                       columns=['objName', 'objID', 'yFlags', 'distance'])
        row = np.where((result['objName'] == 'PSO J322.4622+12.1920') & (result['yFlags'] == 16777496))
        assert isinstance(result, Table)
        np.testing.assert_allclose(result[row]['distance'], 0.039381703406789904)

        result = Catalogs.query_region("158.47924 -7.30962",
                                       radius=in_radius,
                                       catalog="Galex")
        in_radius_arcmin = 0.1*u.deg.to(u.arcmin)
        distances = list(result['distance_arcmin'])
        assert isinstance(result, Table)
        assert max(distances) <= in_radius_arcmin

        result = Catalogs.query_region("158.47924 -7.30962",
                                       radius=in_radius,
                                       catalog="tic")
        row = np.where(result['ID'] == '841736289')
        second_id = result[1]['ID']
        check_result(result, row, {'gaiaqflag': 1})
        np.testing.assert_allclose(result[row]['RA_orig'], 158.475246786483)

        result = Catalogs.query_region("158.47924 -7.30962",
                                       radius=in_radius,
                                       catalog="tic",
                                       pagesize=1,
                                       page=2)
        assert isinstance(result, Table)
        assert len(result) == 1
        assert second_id == result[0]['ID']

        result = Catalogs.query_region("158.47924 -7.30962",
                                       radius=in_radius,
                                       catalog="ctl")
        row = np.where(result['ID'] == '56662064')
        check_result(result, row, {'TYC': '4918-01335-1'})

        result = Catalogs.query_region("210.80227 54.34895",
                                       radius=1*u.deg,
                                       catalog="diskdetective")
        row = np.where(result['designation'] == 'J140544.95+535941.1')
        check_result(result, row, {'ZooniverseID': 'AWI0000r57'})

    def test_catalogs_query_object_async(self):
        responses = Catalogs.query_object_async("M10",
                                                radius=.02,
                                                catalog="TIC")
        assert isinstance(responses, list)

    def test_catalogs_query_object(self):
        def check_result(result, exp_values):
            assert isinstance(result, Table)
            for k, v in exp_values.items():
                assert v in result[k]

        result = Catalogs.query_object("M10",
                                       radius=.001,
                                       catalog="TIC")
        check_result(result, {'ID': '1305764225'})
        second_id = result[1]['ID']

        result = Catalogs.query_object("M10",
                                       radius=.001,
                                       catalog="TIC",
                                       pagesize=1,
                                       page=2)
        assert isinstance(result, Table)
        assert len(result) == 1
        assert second_id == result[0]['ID']

        result = Catalogs.query_object("M10",
                                       radius=.001,
                                       catalog="HSC",
                                       magtype=1)
        check_result(result, {'MatchID': '667727'})

        result = Catalogs.query_object("M10",
                                       radius=.001,
                                       catalog="panstarrs",
                                       table="mean",
                                       columns=['objName', 'objID'])
        check_result(result, {'objName': 'PSO J254.2873-04.1006'})

        result = Catalogs.query_object("M10",
                                       radius=0.18,
                                       catalog="diskdetective")
        check_result(result, {'designation': 'J165749.79-040315.1'})

        result = Catalogs.query_object("M10",
                                       radius=0.001,
                                       catalog="Gaia",
                                       version=1)
        distances = list(result['distance'])
        radius_arcmin = 0.01 * u.deg.to(u.arcmin)
        assert isinstance(result, Table)
        assert max(distances) < radius_arcmin

        result = Catalogs.query_object("TIC 441662144",
                                       radius=0.001,
                                       catalog="ctl")
        check_result(result, {'ID': '441662144'})

        result = Catalogs.query_object('M10',
                                       radius=0.08,
                                       catalog='plato')
        assert 'PICidDR1' in result.colnames

    def test_catalogs_query_criteria_async(self):
        # without position
        responses = Catalogs.query_criteria_async(catalog="Tic",
                                                  Bmag=[30, 50],
                                                  objType="STAR")
        assert isinstance(responses, list)

        responses = Catalogs.query_criteria_async(catalog="ctl",
                                                  Bmag=[30, 50],
                                                  objType="STAR")
        assert isinstance(responses, list)

        responses = Catalogs.query_criteria_async(catalog="DiskDetective",
                                                  state=["inactive", "disabled"],
                                                  oval=[8, 10],
                                                  multi=[3, 7])
        assert isinstance(responses, list)

        # with position
        responses = Catalogs.query_criteria_async(catalog="Tic",
                                                  object_name="M10",
                                                  objType="EXTENDED")
        assert isinstance(responses, list)

        responses = Catalogs.query_criteria_async(catalog="CTL",
                                                  object_name="M10",
                                                  objType="EXTENDED")
        assert isinstance(responses, list)

        responses = Catalogs.query_criteria_async(catalog="DiskDetective",
                                                  object_name="M10",
                                                  radius=2,
                                                  state="complete")
        assert isinstance(responses, list)

        responses = Catalogs.query_criteria_async(catalog="panstarrs",
                                                  table="mean",
                                                  object_name="M10",
                                                  radius=.02,
                                                  qualityFlag=48)
        assert isinstance(responses, Response)

    def test_catalogs_query_criteria(self):
        def check_result(result, exp_vals):
            assert isinstance(result, Table)
            for k, v in exp_vals.items():
                assert v in result[k]

        # without position
        result = Catalogs.query_criteria(catalog="Tic",
                                         Bmag=[30, 50],
                                         objType="STAR")
        check_result(result, {'ID': '81609218'})
        second_id = result[1]['ID']

        result = Catalogs.query_criteria(catalog="Tic",
                                         Bmag=[30, 50],
                                         objType="STAR",
                                         pagesize=1,
                                         page=2)
        assert isinstance(result, Table)
        assert len(result) == 1
        assert second_id == result[0]['ID']

        result = Catalogs.query_criteria(catalog="ctl",
                                         Tmag=[10.5, 11],
                                         POSflag="2mass")
        check_result(result, {'ID': '291067184'})

        result = Catalogs.query_criteria(catalog="DiskDetective",
                                         state=["inactive", "disabled"],
                                         oval=[8, 10],
                                         multi=[3, 7])
        check_result(result, {'designation': 'J003920.04-300132.4'})

        # with position
        result = Catalogs.query_criteria(catalog="Tic",
                                         object_name="M10", objType="EXTENDED")
        check_result(result, {'ID': '10000732589'})

        result = Catalogs.query_criteria(object_name='TIC 291067184',
                                         catalog="ctl",
                                         Tmag=[10.5, 11],
                                         POSflag="2mass")
        check_result(result, {'Tmag': 10.893})

        result = Catalogs.query_criteria(catalog="DiskDetective",
                                         object_name="M10",
                                         radius=2,
                                         state="complete")
        check_result(result, {'designation': 'J165628.40-054630.8'})

        result = Catalogs.query_criteria(catalog="panstarrs",
                                         object_name="M10",
                                         radius=.01,
                                         qualityFlag=32,
                                         zoneID=10306,
                                         columns=['objName', 'objID'])
        check_result(result, {'objName': 'PSO J254.2861-04.1091'})

        result = Catalogs.query_criteria(coordinates="158.47924 -7.30962",
                                         radius=0.01,
                                         catalog="PANSTARRS",
                                         table="mean",
                                         data_release="dr2",
                                         nStackDetections=[("gte", "1")],
                                         columns=["objName", "distance"],
                                         sort_by=[("asc", "distance")])
        assert isinstance(result, Table)
        assert result['distance'][0] <= result['distance'][1]

        # with case-insensitive keyword arguments
        result = Catalogs.query_criteria(catalog="Tic",
                                         bMAG=[30, 50],
                                         objtype="STAR")
        check_result(result, {'ID': '81609218'})

        result = Catalogs.query_criteria(catalog="DiskDetective",
                                         STATE=["inactive", "disabled"],
                                         oVaL=[8, 10],
                                         Multi=[3, 7])
        check_result(result, {'designation': 'J003920.04-300132.4'})

    def test_catalogs_query_criteria_invalid_keyword(self):
        # attempt to make a criteria query with invalid keyword
        with pytest.raises(InvalidQueryError) as err_no_alt:
            Catalogs.query_criteria(catalog='tic', not_a_keyword='TESS')
        assert "Filter 'not_a_keyword' does not exist." in str(err_no_alt.value)

        # keyword is close enough for difflib to offer alternative
        with pytest.raises(InvalidQueryError) as err_with_alt:
            Catalogs.query_criteria(catalog='ctl', objectType="STAR")
        assert 'objType' in str(err_with_alt.value)

        # region query with invalid keyword
        with pytest.raises(InvalidQueryError) as err_region:
            Catalogs.query_region('322.49324 12.16683',
                                  radius=0.001*u.deg,
                                  catalog='HSC',
                                  invalid=2)
        assert "Filter 'invalid' does not exist for catalog HSC." in str(err_region.value)

        # panstarrs criteria query with invalid keyword
        with pytest.raises(InvalidQueryError) as err_ps_criteria:
            Catalogs.query_criteria(coordinates="158.47924 -7.30962",
                                    catalog="PANSTARRS",
                                    table="mean",
                                    data_release="dr2",
                                    columns=["objName", "distance"],
                                    sort_by=[("asc", "distance")],
                                    obj_name='invalid')
        assert 'objName' in str(err_ps_criteria.value)

    def test_catalogs_query_hsc_matchid_async(self):
        catalogData = Catalogs.query_object("M10",
                                            radius=.001,
                                            catalog="HSC",
                                            magtype=1)

        responses = Catalogs.query_hsc_matchid_async(catalogData[0])
        assert isinstance(responses, list)

        responses = Catalogs.query_hsc_matchid_async(catalogData[0]["MatchID"])
        assert isinstance(responses, list)

    def test_catalogs_query_hsc_matchid(self):
        catalogData = Catalogs.query_object("M10",
                                            radius=.001,
                                            catalog="HSC",
                                            magtype=1)
        matchid = catalogData[0]["MatchID"]

        result = Catalogs.query_hsc_matchid(catalogData[0])
        assert isinstance(result, Table)
        assert (result['MatchID'] == matchid).all()

        result2 = Catalogs.query_hsc_matchid(matchid)
        assert isinstance(result2, Table)
        assert len(result2) == len(result)
        assert (result2['MatchID'] == matchid).all()

    def test_catalogs_get_hsc_spectra_async(self):
        responses = Catalogs.get_hsc_spectra_async()
        assert isinstance(responses, list)

    def test_catalogs_get_hsc_spectra(self):
        result = Catalogs.get_hsc_spectra()
        assert isinstance(result, Table)
        assert result[np.where(result['MatchID'] == '19657846')]
        assert result[np.where(result['DatasetName'] == 'HAG_J072657.06+691415.5_J8HPAXAEQ_V01.SPEC1D')]

    def test_catalogs_download_hsc_spectra(self, tmpdir):
        allSpectra = Catalogs.get_hsc_spectra()

        # actually download the products
        result = Catalogs.download_hsc_spectra(allSpectra[10],
                                               download_dir=str(tmpdir))
        assert isinstance(result, Table)

        for row in result:
            if row['Status'] == 'COMPLETE':
                assert os.path.isfile(row['Local Path'])

        # just get the curl script
        result = Catalogs.download_hsc_spectra(allSpectra[20:24],
                                               download_dir=str(tmpdir), curl_flag=True)
        assert isinstance(result, Table)
        assert os.path.isfile(result['Local Path'][0])

    ######################
    # TesscutClass tests #
    ######################

    def test_tesscut_get_sectors(self):
        def check_sector_table(sector_table):
            assert isinstance(sector_table, Table)
            assert len(sector_table) >= 1
            assert "tess-s00" in sector_table['sectorName'][0]
            assert sector_table['sector'][0] > 0
            assert sector_table['camera'][0] > 0
            assert sector_table['ccd'][0] > 0

        coord = SkyCoord(349.62609, -47.12424, unit="deg")
        sector_table = Tesscut.get_sectors(coordinates=coord)
        check_sector_table(sector_table)

        sector_table = Tesscut.get_sectors(object_name="M104")
        check_sector_table(sector_table)

    def test_tesscut_get_sectors_mt(self):
        # Moving target functionality testing
        moving_target_name = 'Eleonora'
        sector_table = Tesscut.get_sectors(object_name=moving_target_name,
                                           moving_target=True)
        assert isinstance(sector_table, Table)
        assert len(sector_table) >= 1
        assert sector_table['sectorName'][0] == "tess-s0006-1-1"
        assert sector_table['sector'][0] == 6
        assert sector_table['camera'][0] == 1
        assert sector_table['ccd'][0] == 1

        error_nameresolve = f"Could not resolve \"{moving_target_name}\" to a sky position."
        with pytest.raises(ResolverError) as error_msg:
            Tesscut.get_sectors(object_name=moving_target_name)
        assert error_nameresolve in str(error_msg.value)

    def test_tesscut_download_cutouts(self, tmpdir):
        def check_manifest(manifest, ext="fits"):
            assert isinstance(manifest, Table)
            assert len(manifest) >= 1
            assert manifest["Local Path"][0][-4:] == ext
            for row in manifest:
                assert os.path.isfile(row['Local Path'])

        coord = SkyCoord(349.62609, -47.12424, unit="deg")
        manifest = Tesscut.download_cutouts(coordinates=coord, size=1, path=str(tmpdir))
        check_manifest(manifest)

        coord = SkyCoord(107.18696, -70.50919, unit="deg")
        manifest = Tesscut.download_cutouts(coordinates=coord, size=1, sector=27,
                                            path=str(tmpdir))
        check_manifest(manifest)

        manifest = Tesscut.download_cutouts(coordinates=coord, size=[1, 1]*u.pix, sector=33,
                                            path=str(tmpdir))
        check_manifest(manifest)

        manifest = Tesscut.download_cutouts(coordinates=coord, size=1, sector=33,
                                            path=str(tmpdir), inflate=False)
        check_manifest(manifest, ".zip")

        manifest = Tesscut.download_cutouts(object_name="TIC 32449963", size=1, path=str(tmpdir))
        check_manifest(manifest, "fits")

    def test_tesscut_download_cutouts_mt(self, tmpdir):
        # Moving target functionality testing
        moving_target_name = 'Eleonora'

        manifest = Tesscut.download_cutouts(object_name=moving_target_name,
                                            moving_target=True,
                                            sector=6,
                                            size=1,
                                            path=str(tmpdir))
        assert isinstance(manifest, Table)
        assert len(manifest) == 1
        assert manifest["Local Path"][0][-4:] == "fits"
        for row in manifest:
            assert os.path.isfile(row['Local Path'])

        error_nameresolve = f"Could not resolve \"{moving_target_name}\" to a sky position."
        with pytest.raises(ResolverError) as error_msg:
            Tesscut.download_cutouts(object_name=moving_target_name)
        assert error_nameresolve in str(error_msg.value)

    def test_tesscut_get_cutouts(self):
        def check_cutout_hdu(cutout_hdus_list):
            assert isinstance(cutout_hdus_list, list)
            assert len(cutout_hdus_list) >= 1
            assert isinstance(cutout_hdus_list[0], fits.HDUList)

        coord = SkyCoord(107.18696, -70.50919, unit="deg")
        cutout_hdus_list = Tesscut.get_cutouts(coordinates=coord,
                                               size=1,
                                               sector=33)
        check_cutout_hdu(cutout_hdus_list)

        coord = SkyCoord(349.62609, -47.12424, unit="deg")
        cutout_hdus_list = Tesscut.get_cutouts(coordinates=coord,
                                               size=[1, 1]*u.arcmin,
                                               sector=[28, 68])
        check_cutout_hdu(cutout_hdus_list)

        cutout_hdus_list = Tesscut.get_cutouts(object_name="TIC 32449963",
                                               size=1,
                                               sector=37)
        check_cutout_hdu(cutout_hdus_list)

    def test_tesscut_get_cutouts_mt(self):
        # Moving target functionality testing
        moving_target_name = 'Eleonora'
        cutout_hdus_list = Tesscut.get_cutouts(object_name=moving_target_name,
                                               moving_target=True,
                                               sector=6,
                                               size=1)
        assert isinstance(cutout_hdus_list, list)
        assert len(cutout_hdus_list) == 1
        assert isinstance(cutout_hdus_list[0], fits.HDUList)

        error_nameresolve = f"Could not resolve \"{moving_target_name}\" to a sky position."
        with pytest.raises(ResolverError) as error_msg:
            Tesscut.get_cutouts(object_name=moving_target_name)
        assert error_nameresolve in str(error_msg.value)

    def test_tesscut_get_cutouts_mt_no_sector(self):
        """Test get_cutouts with moving target but no sector specified.

        When sector is not specified for moving targets, the method should
        automatically fetch available sectors and make individual requests per sector.
        """
        moving_target_name = "Eleonora"
        # Moving target without specifying sector - should automatically fetch sectors
        cutout_hdus_list = Tesscut.get_cutouts(object_name=moving_target_name, moving_target=True, size=1)
        assert isinstance(cutout_hdus_list, list)
        # Should return cutouts for all available sectors
        assert len(cutout_hdus_list) >= 1
        assert isinstance(cutout_hdus_list[0], fits.HDUList)

    def test_tesscut_download_cutouts_mt_no_sector(self, tmpdir):
        """Test download_cutouts with moving target but no sector specified.

        When sector is not specified for moving targets, the method should
        automatically fetch available sectors and make individual requests per sector.
        """
        moving_target_name = "Eleonora"
        # Moving target without specifying sector - should automatically fetch sectors
        manifest = Tesscut.download_cutouts(object_name=moving_target_name,
                                            moving_target=True, size=1, path=str(tmpdir))
        assert isinstance(manifest, Table)
        # Should return files for all available sectors
        assert len(manifest) >= 1
        assert manifest["Local Path"][0][-4:] == "fits"
        for row in manifest:
            assert os.path.isfile(row["Local Path"])

    ###################
    # ZcutClass tests #
    ###################
    def test_zcut_get_surveys(self):
        def check_survey_list(survery_list, no_results=True):
            assert isinstance(survery_list, list)
            assert len(survery_list) == 0 if no_results else len(survery_list) >= 1

        coord = SkyCoord(189.49206, 62.20615, unit="deg")
        survey_list = Zcut.get_surveys(coordinates=coord)
        check_survey_list(survey_list, False)
        assert survey_list[0] == 'candels_gn_60mas'
        assert survey_list[1] == 'candels_gn_30mas'
        assert survey_list[2] == 'goods_north'

        # This should always return no results
        coord = SkyCoord(57.10523, -30.08085, unit="deg")
        with pytest.warns(NoResultsWarning):
            survey_list = Zcut.get_surveys(coordinates=coord, radius=0)
            check_survey_list(survey_list)

        coord = SkyCoord(57.10523, -30.08085, unit="deg")
        with pytest.warns(NoResultsWarning):
            survey_list = Zcut.get_surveys(coordinates=coord, radius=0)
            check_survey_list(survey_list)

    def test_zcut_download_cutouts(self, tmpdir):

        def check_cutout(cutout_table, ext):
            assert isinstance(cutout_table, Table)
            assert len(cutout_table) >= 1
            assert cutout_table["Local Path"][0][-4:] == ext
            assert os.path.isfile(cutout_table[0]['Local Path'])

        coord = SkyCoord(34.47320, -5.24271, unit="deg")
        cutout_table = Zcut.download_cutouts(coordinates=coord, size=1, path=str(tmpdir))
        check_cutout(cutout_table, "fits")

        coord = SkyCoord(189.28065571, 62.17415175, unit="deg")
        cutout_table = Zcut.download_cutouts(coordinates=coord, size=[1, 1], cutout_format="jpg", path=str(tmpdir))
        check_cutout(cutout_table, ".jpg")

        cutout_table = Zcut.download_cutouts(
            coordinates=coord, size=1, units='1*u.arcsec', cutout_format="png", path=str(tmpdir))
        check_cutout(cutout_table, ".png")

        # Intentionally returns no results
        with pytest.warns(NoResultsWarning):
            cutout_table = Zcut.download_cutouts(coordinates=coord,
                                                 survey='candels_gn_30mas',
                                                 cutout_format="jpg",
                                                 path=str(tmpdir))
            assert isinstance(cutout_table, Table)
            assert len(cutout_table) == 0

        cutout_table = Zcut.download_cutouts(
            coordinates=coord, size=1, survey='goods_north', cutout_format="jpg", path=str(tmpdir))
        check_cutout(cutout_table, ".jpg")

        cutout_table = Zcut.download_cutouts(
            coordinates=coord, size=1, cutout_format="jpg", path=str(tmpdir), stretch='asinh', invert=True)
        check_cutout(cutout_table, ".jpg")

    def test_zcut_get_cutouts(self):
        def check_cutout_list(cutout_list, multi=True):
            assert isinstance(cutout_list, list)
            assert isinstance(cutout_list[0], fits.HDUList)
            assert len(cutout_list) >= 1 if multi else len(cutout_list) == 1

        coord = SkyCoord(189.28065571, 62.17415175, unit="deg")

        cutout_list = Zcut.get_cutouts(coordinates=coord, size=1)
        check_cutout_list(cutout_list)

        cutout_list = Zcut.get_cutouts(coordinates=coord, size=[1, 1])
        check_cutout_list(cutout_list)

        # Intentionally returns no results
        with pytest.warns(NoResultsWarning):
            cutout_list = Zcut.get_cutouts(coordinates=coord,
                                           survey='candels_gn_30mas')
            assert isinstance(cutout_list, list)
            assert len(cutout_list) == 0

        cutout_list = Zcut.get_cutouts(coordinates=coord,
                                       survey='3dhst_goods-n',
                                       size=1)
        check_cutout_list(cutout_list, multi=False)

    ###################
    # HapcutClass tests #
    ###################

    def test_hapcut_download_cutouts(self, tmpdir):
        def check_cutout_table(cutout_table, check_shape=True, data_shape=None):
            assert isinstance(cutout_table, Table)
            assert len(cutout_table) >= 1
            for row in cutout_table:
                assert os.path.isfile(row['Local Path'])
                if check_shape and 'fits' in os.path.basename(row['Local Path']):
                    assert fits.getdata(row['Local Path']).shape == data_shape

        # Test 1: Simple API call with expected results
        coord = SkyCoord(351.347812, 28.497808, unit="deg")

        cutout_table = Hapcut.download_cutouts(coordinates=coord, size=5, path=str(tmpdir))
        check_cutout_table(cutout_table, data_shape=(5, 5))

        # Test 2: Make input size a list
        cutout_table = Hapcut.download_cutouts(coordinates=coord, size=[2, 3], path=str(tmpdir))
        check_cutout_table(cutout_table, data_shape=(3, 2))

        # Test 3: Specify unit for input size
        cutout_table = Hapcut.download_cutouts(coordinates=coord, size=5*u.arcsec, path=str(tmpdir))
        check_cutout_table(cutout_table, check_shape=False)

        # Test 4: Intentional API call with no results
        bad_coord = SkyCoord(102.7, 70.50, unit="deg")
        with pytest.warns(NoResultsWarning, match='Missing HAP files for input target. Cutout not performed.'):
            cutout_table = Hapcut.download_cutouts(coordinates=bad_coord, size=5, path=str(tmpdir))
            assert isinstance(cutout_table, Table)
            assert len(cutout_table) == 0

    def test_hapcut_get_cutouts(self):
        def check_cutout_list(cutout_list, data_shape):
            assert isinstance(cutout_list, list)
            assert len(cutout_list) >= 1
            assert isinstance(cutout_list[0], fits.HDUList)
            assert cutout_list[0][1].data.shape == data_shape

        # Test 1: Simple API call with expected results
        coord = SkyCoord(351.347812, 28.497808, unit="deg")

        cutout_list = Hapcut.get_cutouts(coordinates=coord)
        check_cutout_list(cutout_list, (5, 5))

        # Test 2: Make input size a list
        cutout_list = Hapcut.get_cutouts(coordinates=coord, size=[2, 3])
        check_cutout_list(cutout_list, (3, 2))

        # Test 3: Specify unit for input size
        cutout_list = Hapcut.get_cutouts(coordinates=coord, size=5*u.arcsec)
        check_cutout_list(cutout_list, (42, 42))

        # Test 4: Intentional API call with no results
        bad_coord = SkyCoord(102.7, 70.50, unit="deg")

        with pytest.warns(NoResultsWarning, match='Missing HAP files for input target. Cutout not performed.'):
            cutout_list = Hapcut.get_cutouts(coordinates=bad_coord)
            assert isinstance(cutout_list, list)
            assert len(cutout_list) == 0
