# Licensed under a 3-clause BSD style license - see LICENSE.rst
"""Tests for `astropy.vo.client.conesearch` and `astropy.vo.client.async`."""
from __future__ import (absolute_import, division, print_function,
                        unicode_literals)

# STDLIB
import os
import time
import warnings

# THIRD-PARTY
import pytest
import numpy as np

# LOCAL
from .. import conesearch, vos_catalog
from ..exceptions import VOSError, ConeSearchError
from ... import conf
from .... import units as u
from ....coordinates import ICRS, SkyCoord
from ....tests.helper import remote_data
from ....utils.data import get_pkg_data_filename
from ....utils.exceptions import AstropyDeprecationWarning
from ....utils import data


__doctest_skip__ = ['*']


# Global variables for TestConeSearch
SCS_RA = 0
SCS_DEC = 0
SCS_SR = 0.1
SCS_CENTER = ICRS(SCS_RA * u.degree, SCS_DEC * u.degree)
SCS_RADIUS = SCS_SR * u.degree


def setup_module():
    """Ignore all deprecation warnings here."""
    warnings.simplefilter('ignore', AstropyDeprecationWarning)


def teardown_module():
    warnings.resetwarnings()


@remote_data
class TestConeSearch(object):
    """Test Cone Search on a pre-defined access URL.

    .. note::

        This test will fail if the URL becomes inaccessible,
        which is beyond AstroPy's control. When this happens,
        change the test to use a different URL.

        At the time this was written, ``pedantic=True`` will
        not yield any successful search.

    """

    def setup_class(self):
        # If this link is broken, use the next in database that works
        self.url = ('http://vizier.u-strasbg.fr/viz-bin/votable/-A?-out.all&'
                    '-source=I/252/out&')
        self.catname = 'USNO-A2'

        # Avoid downloading the full database
        conf.conesearch_dbname = 'conesearch_simple'

        # Sometimes 3s is not enough
        data.conf.remote_timeout = 10

        self.verbose = False
        self.pedantic = False

    def test_cat_listing(self):
        assert (conesearch.list_catalogs() ==
                ['BROKEN', 'USNO ACT', 'USNO NOMAD', 'USNO-A2', 'USNO-B1'])
        assert (conesearch.list_catalogs(pattern='usno*a') ==
                ['USNO ACT', 'USNO NOMAD', 'USNO-A2'])

    def test_no_result(self):
        with pytest.raises(VOSError):
            conesearch.conesearch(
                SCS_CENTER, 0.001, catalog_db=self.url,
                pedantic=self.pedantic, verbose=self.verbose)

    @pytest.mark.parametrize(('center', 'radius'),
                             [((SCS_RA, SCS_DEC), SCS_SR),
                              (SCS_CENTER, SCS_RADIUS)])
    def test_one_search(self, center, radius):
        """This does not necessarily uses ``self.url`` because of
        unordered dict in JSON tree.

        """
        tab_1 = conesearch.conesearch(
            center, radius, pedantic=None, verbose=self.verbose)

        assert tab_1.array.size > 0

    def test_sky_coord(self):
        """
        Check that searching with a SkyCoord works too
        """
        sc_cen = SkyCoord(SCS_CENTER)
        tab = conesearch.conesearch(
            sc_cen, SCS_RADIUS, catalog_db=self.url,
            pedantic=self.pedantic, verbose=self.verbose)

        assert tab.array.size > 0

    def test_timeout(self):
        """Test time out error."""
        with pytest.raises(VOSError) as e:
            with data.conf.set_temp('remote_timeout', 0.001):
                conesearch.conesearch(
                    SCS_CENTER, SCS_RADIUS, pedantic=self.pedantic,
                    verbose=self.verbose, catalog_db=self.url, cache=False)
            assert 'timed out' in e.info, 'test_timeout failed'

    def test_searches(self):
        tab_2 = conesearch.conesearch(
            SCS_CENTER, SCS_RADIUS, catalog_db=self.url,
            pedantic=self.pedantic, verbose=self.verbose)

        tab_3 = conesearch.conesearch(
            SCS_CENTER, SCS_RADIUS,
            catalog_db=[self.catname, self.url],
            pedantic=self.pedantic, verbose=self.verbose)

        tab_4 = conesearch.conesearch(
            SCS_CENTER, SCS_RADIUS,
            catalog_db=vos_catalog.get_remote_catalog_db(
                conf.conesearch_dbname),
            pedantic=self.pedantic, verbose=self.verbose)

        assert tab_2.url == tab_3.url
        np.testing.assert_array_equal(tab_2.array, tab_3.array)

        # If this fails, it is because of dict hashing, no big deal.
        if tab_2.url == tab_4.url:
            np.testing.assert_array_equal(tab_2.array, tab_4.array)
        else:
            pytest.xfail('conesearch_simple.json used a different URL')

    @pytest.mark.parametrize(('center', 'radius'),
                             [((SCS_RA, SCS_DEC), SCS_SR),
                              (SCS_CENTER, SCS_RADIUS)])
    def test_search_all(self, center, radius):
        all_results = conesearch.search_all(
            center, radius, catalog_db=['BROKEN', self.url],
            pedantic=self.pedantic, verbose=self.verbose)

        assert len(all_results) == 1

        tab_1 = all_results[self.url]

        assert tab_1.array.size > 0

    def test_async(self):
        async_search = conesearch.AsyncConeSearch(
            SCS_CENTER, SCS_RADIUS, pedantic=self.pedantic)

        # Wait a little for the instance to set up properly
        time.sleep(1)

        tab = async_search.get(timeout=data.conf.remote_timeout)

        try:
            assert async_search.done()
        except AssertionError as exc:
            pytest.xfail(str(exc))
        else:
            assert tab.array.size > 0

    def test_async_all(self):
        async_search_all = conesearch.AsyncSearchAll(
            SCS_CENTER, SCS_RADIUS, pedantic=self.pedantic)

        # Wait a little for the instance to set up properly
        time.sleep(1)

        all_results = async_search_all.get(timeout=data.conf.remote_timeout*3)

        try:
            assert async_search_all.done()
        except AssertionError as exc:
            pytest.xfail(str(exc))
        else:
            for tab in all_results.values():
                assert tab.array.size > 0

    @pytest.mark.parametrize(('center', 'radius'),
                             [((SCS_RA, SCS_DEC), 0.8),
                              (SCS_CENTER, 0.8 * u.degree)])
    def test_prediction(self, center, radius):
        """Prediction tests are not very accurate but will have to do."""
        t_1, tab_1 = conesearch.conesearch_timer(
            center, radius, catalog_db=self.url,
            pedantic=self.pedantic, verbose=self.verbose)
        n_1 = tab_1.array.size

        t_2, n_2 = conesearch.predict_search(
            self.url, center, radius,
            pedantic=self.pedantic, verbose=self.verbose)

        assert n_2 > 0 and n_2 <= n_1 * 1.5

        # Timer depends on network latency as well, so upper limit is very lax.
        assert t_2 > 0 and t_2 <= t_1 * 10

    def test_prediction_neg_radius(self):
        with pytest.raises(ConeSearchError):
            t, n = conesearch.predict_search(
                self.url, SCS_CENTER, -1, pedantic=self.pedantic,
                verbose=self.verbose)

    def teardown_class(self):
        conf.reset('conesearch_dbname')
        data.conf.reset('remote_timeout')


class TestErrorResponse(object):
    """Test Cone Search error response handling.

    This is defined in Section 2.3 of Simple Cone Search Version 1.03,
    IVOA Recommendation, 22 February 2008.

    Also see https://github.com/astropy/astropy/issues/1001

    """

    def setup_class(self):
        self.datadir = 'data'
        self.pedantic = False
        self.conesearch_errmsg = {
            'conesearch_error1.xml': 'Error in input RA value: as3f',
            'conesearch_error2.xml': 'Error in input RA value: as3f',
            'conesearch_error3.xml': 'Invalid data type: text/html',
            'conesearch_error4.xml': 'Invalid data type: text/html'}

    def conesearch_compare(self, xmlfile, msg):
        """Bypassing Cone Search query and just imitating the reply,
        then check if appropriate error message is caught.

        """
        # conesearch_error4.xml is a wont-fix for now
        if xmlfile == 'conesearch_error4.xml':
            pytest.xfail('Currently not supported, '
                         'see astropy.io.votable.exceptions.W22')

        url = get_pkg_data_filename(os.path.join(self.datadir, xmlfile))
        try:
            vos_catalog._vo_service_request(url, self.pedantic, {})
        except VOSError as e:
            assert msg in str(e)

    @pytest.mark.parametrize(('id'), [1, 2, 3, 4])
    def test_conesearch_response(self, id):
        xml = 'conesearch_error{0}.xml'.format(id)
        msg = self.conesearch_errmsg[xml]
        self.conesearch_compare(xml, msg)


def test_validate_coord():
    """Valid coordinates should not raise an error."""
    result = conesearch._validate_coord(
        ICRS(6.02233 * u.degree, -72.08144 * u.degree))
    np.testing.assert_allclose(result, [6.022330000000011, -72.08144])

    result = conesearch._validate_coord(
        SkyCoord(6.02233 * u.degree, -72.08144 * u.degree, frame='icrs'))
    np.testing.assert_allclose(result, [6.022330000000011, -72.08144])

    result = conesearch._validate_coord((-0, 0))
    np.testing.assert_allclose(result, [0, 0])

    result = conesearch._validate_coord((-1, -1))
    np.testing.assert_allclose(result, [359, -1])
