Add TAP endpoint and query behavior tests

astroquery/eso/tests/test_eso.py

@@ -281,6 +281,17 @@ def test_tap_endpoint_invalid_url():
         eso_instance._tap_endpoint("https://archive.eso.org/not-a-tap")
 
 
+# This TAP test is deliberately offline. It checks endpoint validation before
+# pyvo could construct a real remote service.
+def test_tap_url_invalid_endpoint():
+    eso_instance = Eso()
+
+    # Unknown endpoint names should fail before constructing a TAP service,
+    # so callers get a local configuration error instead of a remote failure.
+    with pytest.raises(ValueError, match="tap_endpoint must be one of"):
+        eso_instance._tap_url("not-a-tap")
+
+
 @pytest.mark.parametrize("input_val, expected", [
     # Numeric values
     (1, "= 1"),
@@ -366,6 +377,17 @@ def test_maxrec():
     assert maxrec == EXPECTED_MAX_ROW_LIMIT
 
 
+# The next retrieval tests exercise local guardrails and pyvo error handling.
+# Fake TAP objects keep the tests deterministic and document expected fallbacks.
+def test_row_limit_rejects_values_above_service_limit():
+    eso_instance = Eso()
+
+    # The ESO TAP service has a hard maximum; reject larger values immediately
+    # instead of sending impossible maxrec values to pyvo.
+    with pytest.raises(ValueError, match="ROW_LIMIT cannot be higher"):
+        eso_instance.ROW_LIMIT = EXPECTED_MAX_ROW_LIMIT + 1
+
+
 def test_retrieve_pyvo_table(monkeypatch):
     eso_instance = Eso()
     dal = pyvo.dal.TAPService(eso_instance._tap_url())
@@ -387,6 +409,50 @@ def mock_search(*args, **kwargs):
         table = eso_instance._try_retrieve_pyvo_table(q_str, dal)
 
 
+def test_retrieve_pyvo_table_reports_custom_tap_url_in_errors():
+    eso_instance = Eso()
+    q_str = "select * from custom.Table"
+
+    class FakeResult:
+        def to_table(self):
+            raise pyvo.dal.exceptions.DALFormatError(reason="bad format")
+
+    class FakeTap:
+        baseurl = "https://example.invalid/custom_tap"
+
+        def search(self, **kwargs):
+            return FakeResult()
+
+    # Custom TAP URLs are not one of the configured ESO endpoints, but the error
+    # message should still show the URL so users can tell which service failed.
+    with pytest.raises(pyvo.dal.exceptions.DALFormatError) as exc:
+        eso_instance._try_retrieve_pyvo_table(q_str, FakeTap())
+
+    assert 'tap_endpoint="https://example.invalid/custom_tap"' in str(exc.value)
+
+
+def test_retrieve_pyvo_table_handles_overflow_warning_as_partial_result():
+    eso_instance = Eso()
+
+    class FakeResult:
+        def to_table(self):
+            raise pyvo.dal.exceptions.DALOverflowWarning("overflow")
+
+    class FakeTap:
+        baseurl = eso_instance._tap_url()
+
+        def search(self, **kwargs):
+            return FakeResult()
+
+    # pyvo raises DALOverflowWarning when the TAP result is incomplete. The ESO
+    # wrapper logs that condition and then applies its normal no-result warning.
+    with pytest.warns(NoResultsWarning):
+        result = eso_instance._try_retrieve_pyvo_table("select * from ivoa.ObsCore", FakeTap())
+
+    assert isinstance(result, Table)
+    assert len(result) == 0
+
+
 def test_issue_table_length_warnings():
     eso_instance = Eso()
 
@@ -405,6 +471,30 @@ def test_issue_table_length_warnings():
     eso_instance._maybe_warn_about_table_length(t, EXPECTED_MAXREC+1)
 
 
+# This helper test is observation-TAP specific. It documents the metadata query
+# used for observation table help without touching the catalogue TAP endpoint.
+def test_columns_table_uses_tap_obs_metadata_query(monkeypatch):
+    eso_instance = Eso()
+    calls = []
+
+    def fake_query_tap(query, *, tap_endpoint):
+        calls.append((query, tap_endpoint))
+        return Table({"column_name": ["dp_id"]})
+
+    # Observation TAP exposes xtype, so the help query for tap_obs should use
+    # the observation metadata columns and stay on the observation endpoint.
+    monkeypatch.setattr(eso_instance, "query_tap", fake_query_tap)
+
+    result = eso_instance._columns_table("ivoa.ObsCore", tap_endpoint="tap_obs")
+
+    assert result["column_name"][0] == "dp_id"
+    assert calls == [(
+        "select column_name, datatype, unit, xtype "
+        "from TAP_SCHEMA.columns where table_name = 'ivoa.ObsCore'",
+        "tap_obs",
+    )]
+
+
 def test_reorder_columns(monkeypatch):
     eso = Eso()
     monkeypatch.setattr(eso, 'query_tap', monkey_tap)

astroquery/eso/tests/test_eso_catalogs.py

@@ -11,9 +11,12 @@
 import os
 
 import astropy.io.ascii
+import pytest
 from astropy.table import Table
 
 from ...eso import Eso
+from ...eso import core as eso_core
+from ...eso.utils import _UserParams
 
 DATA_DIR = os.path.join(os.path.dirname(__file__), "data")
 EXPECTED_MAXREC = 1000
@@ -266,6 +269,172 @@ def monkey_tap(query, **kwargs):
     return table
 
 
+# These catalogue TAP tests are deliberately offline. They cover TAP_CAT
+# endpoint routing and metadata compatibility without contacting the ESO archive.
+@pytest.mark.parametrize("table_name, expected", [
+    ("safcat.KiDS_DR4_1_ugriZYJHKs_cat_fits", True),
+    ("KiDS_DR4_1_ugriZYJHKs_cat_fits", False),
+])
+def test_tap_cat_table_names_have_schema_prefix(monkeypatch, table_name, expected):
+    eso = Eso()
+    calls = []
+
+    def fake_query_tap(query, *, tap_endpoint):
+        calls.append((query, tap_endpoint))
+        return Table({"table_name": [table_name]})
+
+    # TAP_CAT can expose table names with or without the schema prefix depending
+    # on TAP version, so this helper decides which query shape to use later.
+    monkeypatch.setattr(eso, "query_tap", fake_query_tap)
+
+    result = eso._tap_cat_table_names_have_schema_prefix(tap_endpoint="tap_cat")
+
+    assert result is expected
+    assert calls == [("select top 1 table_name from TAP_SCHEMA.tables", "tap_cat")]
+
+
+@pytest.mark.parametrize("authenticated", [False, True])
+def test_tap_uses_catalog_endpoint(monkeypatch, authenticated):
+    eso = Eso()
+    calls = []
+
+    class FakeTAPService:
+        def __init__(self, url, session=None):
+            calls.append((url, session))
+
+    # TAPService is monkeypatched so this verifies TAP_CAT endpoint/session
+    # selection without opening a real pyvo connection to the ESO archive.
+    monkeypatch.setattr(eso_core, "TAPService", FakeTAPService)
+    monkeypatch.setattr(eso, "authenticated", lambda: True)
+    monkeypatch.setattr(eso, "_get_auth_header", lambda: {"Authorization": "Bearer token"})
+
+    eso.tap(authenticated=authenticated, tap_endpoint="tap_cat")
+
+    expected_session = eso._session if authenticated else None
+    assert calls == [(eso._tap_url("tap_cat"), expected_session)]
+
+
+def test_query_tap_passes_catalog_endpoint_to_tap_and_retriever(monkeypatch):
+    eso = Eso()
+    fake_tap_service = object()
+    fake_table = Table({"col": [1]})
+    calls = []
+
+    def fake_tap(authenticated=False, *, tap_endpoint="tap_obs"):
+        calls.append(("tap", authenticated, tap_endpoint))
+        return fake_tap_service
+
+    def fake_retrieve(query, tap_service):
+        calls.append(("retrieve", query, tap_service))
+        return fake_table
+
+    # query_tap is the free-ADQL entry point for catalogue queries, so it must
+    # preserve tap_endpoint="tap_cat" down to the pyvo retrieval helper.
+    monkeypatch.setattr(eso, "tap", fake_tap)
+    monkeypatch.setattr(eso, "_try_retrieve_pyvo_table", fake_retrieve)
+
+    result = eso.query_tap("select 1", authenticated=True, tap_endpoint="tap_cat")
+
+    assert result is fake_table
+    assert calls == [
+        ("tap", True, "tap_cat"),
+        ("retrieve", "select 1", fake_tap_service),
+    ]
+
+
+def test_columns_table_uses_tap_cat_metadata_query_without_schema_prefix(monkeypatch):
+    eso = Eso()
+    calls = []
+
+    def fake_query_tap(query, *, tap_endpoint):
+        calls.append((query, tap_endpoint))
+        return Table({"column_name": ["ID"]})
+
+    # Some TAP_CAT deployments list table names without "safcat.". When that is
+    # detected, the helper strips the prefix before querying TAP_SCHEMA.columns.
+    monkeypatch.setattr(eso, "_tap_cat_table_names_have_schema_prefix",
+                        lambda *, tap_endpoint: False)
+    monkeypatch.setattr(eso, "query_tap", fake_query_tap)
+
+    result = eso._columns_table(
+        "safcat.KiDS_DR4_1_ugriZYJHKs_cat_fits",
+        tap_endpoint="tap_cat",
+    )
+
+    assert result["column_name"][0] == "ID"
+    assert calls == [(
+        "select column_name, datatype, unit, ucd "
+        "from TAP_SCHEMA.columns "
+        "where table_name = 'KiDS_DR4_1_ugriZYJHKs_cat_fits'",
+        "tap_cat",
+    )]
+
+
+def test_list_column_uses_catalog_endpoint_for_help_and_count(monkeypatch):
+    eso = Eso()
+    calls = []
+
+    def fake_columns_table(table_name, *, tap_endpoint):
+        calls.append(("columns", table_name, tap_endpoint))
+        return Table({"column_name": ["ID"], "datatype": ["int"], "unit": [""], "ucd": [""]})
+
+    def fake_query_tap(query, *, tap_endpoint):
+        calls.append(("count", query, tap_endpoint))
+        return Table({"count": [7]})
+
+    # list/help output needs two TAP calls: one for column metadata and one for
+    # row count. Both must use the catalogue endpoint when helping catalogues.
+    monkeypatch.setattr(eso, "_columns_table", fake_columns_table)
+    monkeypatch.setattr(eso, "query_tap", fake_query_tap)
+
+    eso._list_column("safcat.KiDS_DR4_1_ugriZYJHKs_cat_fits", tap_endpoint="tap_cat")
+
+    assert calls == [
+        ("columns", "safcat.KiDS_DR4_1_ugriZYJHKs_cat_fits", "tap_cat"),
+        ("count", "select count(*) from safcat.KiDS_DR4_1_ugriZYJHKs_cat_fits", "tap_cat"),
+    ]
+
+
+def test_query_on_allowed_values_catalog_help_uses_catalog_endpoint(monkeypatch):
+    eso = Eso()
+    calls = []
+
+    def fake_list_column(table_name, *, tap_endpoint):
+        calls.append((table_name, tap_endpoint))
+
+    # help=True should print/list valid catalogue columns and stop there; it
+    # should not build or submit a data query to TAP.
+    monkeypatch.setattr(eso, "_list_column", fake_list_column)
+    monkeypatch.setattr(eso, "query_tap", lambda *args, **kwargs: pytest.fail("unexpected TAP query"))
+
+    result = eso._query_on_allowed_values(_UserParams(
+        table_name="safcat.KiDS_DR4_1_ugriZYJHKs_cat_fits",
+        print_help=True,
+        tap_endpoint="tap_cat",
+    ))
+
+    assert result is None
+    assert calls == [("safcat.KiDS_DR4_1_ugriZYJHKs_cat_fits", "tap_cat")]
+
+
+def test_query_on_allowed_values_catalog_payload_does_not_query_tap(monkeypatch):
+    eso = Eso()
+
+    # get_query_payload=True is a dry-run mode for inspecting catalogue ADQL, so
+    # the method should return the generated query without contacting TAP_CAT.
+    monkeypatch.setattr(eso, "query_tap", lambda *args, **kwargs: pytest.fail("unexpected TAP query"))
+
+    result = eso._query_on_allowed_values(_UserParams(
+        table_name="safcat.KiDS_DR4_1_ugriZYJHKs_cat_fits",
+        columns="ID",
+        column_filters={"MAG_AUTO": "< 10"},
+        get_query_payload=True,
+        tap_endpoint="tap_cat",
+    ))
+
+    assert result == "select ID from safcat.KiDS_DR4_1_ugriZYJHKs_cat_fits where MAG_AUTO < 10"
+
+
 def test_list_catalogs_latest_versions(monkeypatch):
     eso = Eso()
     monkeypatch.setattr(eso, "query_tap", monkey_tap)
@@ -289,3 +458,43 @@ def test_query_catalogs(monkeypatch):
     result = eso.query_catalog("KiDS_DR4_1_ugriZYJHKs_cat_fits")
     assert isinstance(result, Table)
     assert len(result) <= 5
+
+
+# These catalogue wrapper tests document the non-network paths. They verify
+# query dry-runs and help routing without asking TAP_CAT for live data.
+def test_query_catalog_get_query_payload_prefixes_catalog_schema(monkeypatch):
+    eso = Eso()
+
+    # get_query_payload=True is a dry-run mode. It should add the safcat schema
+    # prefix for catalogue tables and return ADQL without making a TAP request.
+    monkeypatch.setattr(eso, "query_tap", lambda *args, **kwargs: pytest.fail("unexpected TAP query"))
+
+    query = eso.query_catalog(
+        "KiDS_DR4_1_ugriZYJHKs_cat_fits",
+        columns="ID",
+        column_filters={"MAG_AUTO": "< 10"},
+        get_query_payload=True,
+    )
+
+    assert query == (
+        "select ID from safcat.KiDS_DR4_1_ugriZYJHKs_cat_fits "
+        "where MAG_AUTO < 10"
+    )
+
+
+def test_query_catalog_help_uses_catalog_endpoint(monkeypatch):
+    eso = Eso()
+    calls = []
+
+    def fake_list_column(table_name, *, tap_endpoint):
+        calls.append((table_name, tap_endpoint))
+
+    # help=True should route through the catalogue TAP endpoint so column help
+    # describes catalogue columns rather than observation-table columns.
+    monkeypatch.setattr(eso, "_list_column", fake_list_column)
+    monkeypatch.setattr(eso, "query_tap", lambda *args, **kwargs: pytest.fail("unexpected TAP query"))
+
+    result = eso.query_catalog("KiDS_DR4_1_ugriZYJHKs_cat_fits", help=True)
+
+    assert result is None
+    assert calls == [("safcat.KiDS_DR4_1_ugriZYJHKs_cat_fits", "tap_cat")]