From a320e046594c6a2022767cb70a660d1bf2ea4a74 Mon Sep 17 00:00:00 2001 From: Ashley Barnes <30494539+ashleythomasbarnes@users.noreply.github.com> Date: Wed, 10 Jun 2026 15:42:20 +0200 Subject: [PATCH 1/2] Add TAP endpoint and query behavior tests Add comprehensive tests for TAP endpoint selection and ADQL/query behavior. Tests cover: invalid tap_endpoint names, detection of schema prefixes in TAP_CAT table names, ROW_LIMIT validation, _try_retrieve_pyvo_table error messages and overflow handling, selecting the correct TAP endpoint/session for authenticated and unauthenticated calls, propagation of tap_endpoint through query_tap to the retriever, correct metadata queries for columns_table on tap_obs vs tap_cat (including prefix-stripping), list/help flows using the catalog endpoint, and dry-run get_query_payload behavior for both _query_on_allowed_values and query_catalog. Also import eso.core as eso_core to allow monkeypatching TAPService and add missing pytest import in catalog tests. --- astroquery/eso/tests/test_eso.py | 249 ++++++++++++++++++++++ astroquery/eso/tests/test_eso_catalogs.py | 39 ++++ 2 files changed, 288 insertions(+) diff --git a/astroquery/eso/tests/test_eso.py b/astroquery/eso/tests/test_eso.py index f5f49761de..373dc38f5d 100644 --- a/astroquery/eso/tests/test_eso.py +++ b/astroquery/eso/tests/test_eso.py @@ -18,6 +18,7 @@ from astroquery.utils.mocks import MockResponse from ...eso import Eso +from ...eso import core as eso_core from ...eso.utils import _UserParams, \ _build_adql_string, _adql_sanitize_op_val, _reorder_columns, \ DEFAULT_LEAD_COLS_RAW @@ -281,6 +282,37 @@ def test_tap_endpoint_invalid_url(): eso_instance._tap_endpoint("https://archive.eso.org/not-a-tap") +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("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_instance = 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_instance, "query_tap", fake_query_tap) + + result = eso_instance._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("input_val, expected", [ # Numeric values (1, "= 1"), @@ -366,6 +398,15 @@ def test_maxrec(): assert maxrec == EXPECTED_MAX_ROW_LIMIT +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 +428,99 @@ 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 + + +@pytest.mark.parametrize("authenticated", [False, True]) +def test_tap_uses_selected_endpoint(monkeypatch, authenticated): + eso_instance = Eso() + calls = [] + + class FakeTAPService: + def __init__(self, url, session=None): + calls.append((url, session)) + + # TAPService is monkeypatched so this verifies endpoint/session selection + # without opening a real pyvo connection to the ESO archive. + monkeypatch.setattr(eso_core, "TAPService", FakeTAPService) + monkeypatch.setattr(eso_instance, "authenticated", lambda: True) + monkeypatch.setattr(eso_instance, "_get_auth_header", lambda: {"Authorization": "Bearer token"}) + + eso_instance.tap(authenticated=authenticated, tap_endpoint="tap_cat") + + expected_session = eso_instance._session if authenticated else None + assert calls == [(eso_instance._tap_url("tap_cat"), expected_session)] + + +def test_query_tap_passes_endpoint_to_tap_and_retriever(monkeypatch): + eso_instance = 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 public free-ADQL entry point, so it must preserve the + # selected endpoint all the way down to the pyvo retrieval helper. + monkeypatch.setattr(eso_instance, "tap", fake_tap) + monkeypatch.setattr(eso_instance, "_try_retrieve_pyvo_table", fake_retrieve) + + result = eso_instance.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_issue_table_length_warnings(): eso_instance = Eso() @@ -405,6 +539,121 @@ def test_issue_table_length_warnings(): eso_instance._maybe_warn_about_table_length(t, EXPECTED_MAXREC+1) +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_columns_table_uses_tap_cat_metadata_query_without_schema_prefix(monkeypatch): + eso_instance = 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_instance, "_tap_cat_table_names_have_schema_prefix", + lambda *, tap_endpoint: False) + monkeypatch.setattr(eso_instance, "query_tap", fake_query_tap) + + result = eso_instance._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_selected_endpoint_for_help_and_count(monkeypatch): + eso_instance = 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_instance, "_columns_table", fake_columns_table) + monkeypatch.setattr(eso_instance, "query_tap", fake_query_tap) + + eso_instance._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_print_help_uses_selected_endpoint(monkeypatch): + eso_instance = Eso() + calls = [] + + def fake_list_column(table_name, *, tap_endpoint): + calls.append((table_name, tap_endpoint)) + + # help=True should print/list valid columns and stop there; it should not + # build or submit a data query to TAP. + monkeypatch.setattr(eso_instance, "_list_column", fake_list_column) + monkeypatch.setattr(eso_instance, "query_tap", lambda *args, **kwargs: pytest.fail("unexpected TAP query")) + + result = eso_instance._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_get_query_payload_does_not_query_tap(monkeypatch): + eso_instance = Eso() + + # get_query_payload=True is a dry-run mode for inspecting ADQL, so the + # method should return the generated query without contacting TAP. + monkeypatch.setattr(eso_instance, "query_tap", lambda *args, **kwargs: pytest.fail("unexpected TAP query")) + + result = eso_instance._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_reorder_columns(monkeypatch): eso = Eso() monkeypatch.setattr(eso, 'query_tap', monkey_tap) diff --git a/astroquery/eso/tests/test_eso_catalogs.py b/astroquery/eso/tests/test_eso_catalogs.py index e72645d982..893791b4f9 100644 --- a/astroquery/eso/tests/test_eso_catalogs.py +++ b/astroquery/eso/tests/test_eso_catalogs.py @@ -11,6 +11,7 @@ import os import astropy.io.ascii +import pytest from astropy.table import Table from ...eso import Eso @@ -289,3 +290,41 @@ def test_query_catalogs(monkeypatch): result = eso.query_catalog("KiDS_DR4_1_ugriZYJHKs_cat_fits") assert isinstance(result, Table) assert len(result) <= 5 + + +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")] From 006454ff0ea0f1c79f956b3e3b9e8808445b0015 Mon Sep 17 00:00:00 2001 From: Ashley Barnes <30494539+ashleythomasbarnes@users.noreply.github.com> Date: Wed, 10 Jun 2026 20:46:41 +0100 Subject: [PATCH 2/2] move some catalog tests --- astroquery/eso/tests/test_eso.py | 171 +--------------------- astroquery/eso/tests/test_eso_catalogs.py | 170 +++++++++++++++++++++ 2 files changed, 176 insertions(+), 165 deletions(-) diff --git a/astroquery/eso/tests/test_eso.py b/astroquery/eso/tests/test_eso.py index 373dc38f5d..f801ca8cb5 100644 --- a/astroquery/eso/tests/test_eso.py +++ b/astroquery/eso/tests/test_eso.py @@ -18,7 +18,6 @@ from astroquery.utils.mocks import MockResponse from ...eso import Eso -from ...eso import core as eso_core from ...eso.utils import _UserParams, \ _build_adql_string, _adql_sanitize_op_val, _reorder_columns, \ DEFAULT_LEAD_COLS_RAW @@ -282,6 +281,8 @@ 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() @@ -291,28 +292,6 @@ def test_tap_url_invalid_endpoint(): eso_instance._tap_url("not-a-tap") -@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_instance = 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_instance, "query_tap", fake_query_tap) - - result = eso_instance._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("input_val, expected", [ # Numeric values (1, "= 1"), @@ -398,6 +377,8 @@ 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() @@ -472,55 +453,6 @@ def search(self, **kwargs): assert len(result) == 0 -@pytest.mark.parametrize("authenticated", [False, True]) -def test_tap_uses_selected_endpoint(monkeypatch, authenticated): - eso_instance = Eso() - calls = [] - - class FakeTAPService: - def __init__(self, url, session=None): - calls.append((url, session)) - - # TAPService is monkeypatched so this verifies endpoint/session selection - # without opening a real pyvo connection to the ESO archive. - monkeypatch.setattr(eso_core, "TAPService", FakeTAPService) - monkeypatch.setattr(eso_instance, "authenticated", lambda: True) - monkeypatch.setattr(eso_instance, "_get_auth_header", lambda: {"Authorization": "Bearer token"}) - - eso_instance.tap(authenticated=authenticated, tap_endpoint="tap_cat") - - expected_session = eso_instance._session if authenticated else None - assert calls == [(eso_instance._tap_url("tap_cat"), expected_session)] - - -def test_query_tap_passes_endpoint_to_tap_and_retriever(monkeypatch): - eso_instance = 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 public free-ADQL entry point, so it must preserve the - # selected endpoint all the way down to the pyvo retrieval helper. - monkeypatch.setattr(eso_instance, "tap", fake_tap) - monkeypatch.setattr(eso_instance, "_try_retrieve_pyvo_table", fake_retrieve) - - result = eso_instance.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_issue_table_length_warnings(): eso_instance = Eso() @@ -539,6 +471,8 @@ 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 = [] @@ -561,99 +495,6 @@ def fake_query_tap(query, *, tap_endpoint): )] -def test_columns_table_uses_tap_cat_metadata_query_without_schema_prefix(monkeypatch): - eso_instance = 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_instance, "_tap_cat_table_names_have_schema_prefix", - lambda *, tap_endpoint: False) - monkeypatch.setattr(eso_instance, "query_tap", fake_query_tap) - - result = eso_instance._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_selected_endpoint_for_help_and_count(monkeypatch): - eso_instance = 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_instance, "_columns_table", fake_columns_table) - monkeypatch.setattr(eso_instance, "query_tap", fake_query_tap) - - eso_instance._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_print_help_uses_selected_endpoint(monkeypatch): - eso_instance = Eso() - calls = [] - - def fake_list_column(table_name, *, tap_endpoint): - calls.append((table_name, tap_endpoint)) - - # help=True should print/list valid columns and stop there; it should not - # build or submit a data query to TAP. - monkeypatch.setattr(eso_instance, "_list_column", fake_list_column) - monkeypatch.setattr(eso_instance, "query_tap", lambda *args, **kwargs: pytest.fail("unexpected TAP query")) - - result = eso_instance._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_get_query_payload_does_not_query_tap(monkeypatch): - eso_instance = Eso() - - # get_query_payload=True is a dry-run mode for inspecting ADQL, so the - # method should return the generated query without contacting TAP. - monkeypatch.setattr(eso_instance, "query_tap", lambda *args, **kwargs: pytest.fail("unexpected TAP query")) - - result = eso_instance._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_reorder_columns(monkeypatch): eso = Eso() monkeypatch.setattr(eso, 'query_tap', monkey_tap) diff --git a/astroquery/eso/tests/test_eso_catalogs.py b/astroquery/eso/tests/test_eso_catalogs.py index 893791b4f9..bbc4c73d21 100644 --- a/astroquery/eso/tests/test_eso_catalogs.py +++ b/astroquery/eso/tests/test_eso_catalogs.py @@ -15,6 +15,8 @@ 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 @@ -267,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) @@ -292,6 +460,8 @@ def test_query_catalogs(monkeypatch): 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()