From 776e8cf68f4dd1a8144a0f0c945602033314a975 Mon Sep 17 00:00:00 2001 From: Markus Demleitner Date: Wed, 10 Jul 2024 16:33:53 +0200 Subject: [PATCH] RegistryResults.get_tables now preserves table utypes. In addition, while creating a test for this, I noticed that pytest has scope="function" for fixtures by default. I take the liberty of sneaking in scope="session" for all the regtap fixtures, which speeds up the tests between a bit and (for remote fixtures) dramatically. --- CHANGES.rst | 5 +- pyvo/registry/regtap.py | 4 +- pyvo/registry/tests/test_regtap.py | 82 +++++++++++++++--------------- 3 files changed, 49 insertions(+), 42 deletions(-) diff --git a/CHANGES.rst b/CHANGES.rst index a2b5db93..2cfd9479 100644 --- a/CHANGES.rst +++ b/CHANGES.rst @@ -41,7 +41,10 @@ Enhancements and Fixes done via subqueries for less duplication of interfaces. [#562] - Where datalink records are made from table rows, the table row is - now accessible as datalinks.original_row. [] + now accessible as datalinks.original_row. [#559] + +- Tables returned by RegistryResource.get_tables() now have a utype + attribute [#576] Deprecations and Removals ------------------------- diff --git a/pyvo/registry/regtap.py b/pyvo/registry/regtap.py index 9d2c8309..e6a1b532 100644 --- a/pyvo/registry/regtap.py +++ b/pyvo/registry/regtap.py @@ -1088,6 +1088,7 @@ def _build_vosi_table(self, table_row, columns): res.name = table_row["table_name"] res.title = table_row["table_title"] res.description = table_row["table_description"] + res.utype = table_row["table_utype"] res._columns = [ self._build_vosi_column(row) for row in columns] @@ -1115,7 +1116,8 @@ def get_tables(self, *, table_limit=20): svc = get_RegTAP_service() tables = svc.run_sync( - """SELECT table_name, table_description, table_index, table_title + """SELECT table_name, table_description, table_index, table_title, + table_utype FROM rr.res_table WHERE ivoid={}""".format( rtcons.make_sql_literal(self.ivoid))) diff --git a/pyvo/registry/tests/test_regtap.py b/pyvo/registry/tests/test_regtap.py index 6051447d..425d3747 100644 --- a/pyvo/registry/tests/test_regtap.py +++ b/pyvo/registry/tests/test_regtap.py @@ -31,7 +31,7 @@ @pytest.fixture(name='capabilities') -def _capabilities(mocker): +def _capabilities(mocker, scope="session"): def callback(request, context): return get_pkg_data_contents('data/capabilities.xml') @@ -52,7 +52,7 @@ def callback(request, context): @pytest.fixture(name='regtap_pulsar_distance_response') -def _regtap_pulsar_distance_response(mocker): +def _regtap_pulsar_distance_response(mocker, scope="session"): with mocker.register_uri( 'POST', REGISTRY_BASEURL + '/sync', content=get_pkg_data_contents('data/regtap.xml')) as matcher: @@ -60,7 +60,7 @@ def _regtap_pulsar_distance_response(mocker): @pytest.fixture() -def keywords_fixture(mocker): +def keywords_fixture(mocker, scope="session"): def keywordstest_callback(request, context): data = dict(parse_qsl(request.body)) query = data['QUERY'] @@ -83,7 +83,7 @@ def keywordstest_callback(request, context): @pytest.fixture() -def single_keyword_fixture(mocker): +def single_keyword_fixture(mocker, scope="session"): def keywordstest_callback(request, context): data = dict(parse_qsl(request.body)) query = data['QUERY'] @@ -102,7 +102,7 @@ def keywordstest_callback(request, context): @pytest.fixture() -def servicetype_fixture(mocker): +def servicetype_fixture(mocker, scope="session"): def servicetypetest_callback(request, context): data = dict(parse_qsl(request.body)) query = data['QUERY'] @@ -123,7 +123,7 @@ def servicetypetest_callback(request, context): @pytest.fixture() -def waveband_fixture(mocker): +def waveband_fixture(mocker, scope="session"): def wavebandtest_callback(request, content): data = dict(parse_qsl(request.body)) query = data['QUERY'] @@ -140,7 +140,7 @@ def wavebandtest_callback(request, content): @pytest.fixture() -def datamodel_fixture(mocker): +def datamodel_fixture(mocker, scope="session"): def datamodeltest_callback(request, content): data = dict(parse_qsl(request.body)) query = data['QUERY'] @@ -162,7 +162,7 @@ def datamodeltest_callback(request, content): @pytest.fixture() -def aux_fixture(mocker): +def aux_fixture(mocker, scope="session"): def auxtest_callback(request, context): data = dict(parse_qsl(request.body)) query = data['QUERY'] @@ -179,7 +179,7 @@ def auxtest_callback(request, context): @pytest.fixture(name='multi_interface_fixture') -def _multi_interface_fixture(mocker): +def _multi_interface_fixture(mocker, scope="session"): # to update this, run # import requests # from pyvo.registry import regtap @@ -197,7 +197,7 @@ def _multi_interface_fixture(mocker): @pytest.fixture(name='flash_service') -def _flash_service(multi_interface_fixture): +def _flash_service(multi_interface_fixture, scope="session"): return regtap.search( ivoid="ivo://org.gavo.dc/flashheros/q/ssa")[0] @@ -849,14 +849,14 @@ def test_ambiguous_access_url_warns(self, recwarn): # TODO: While I suppose the contact test should keep requiring network, # I think we should can the network responses involved in the following; # the stuff might change upstream any time and then break our unit tests. -@pytest.fixture(name='flash_tables') -def _flash_tables(): +@pytest.fixture(name='obscore_tables') +def _obscore_tables(scope="session"): rsc = _makeRegistryRecord( - ivoid="ivo://org.gavo.dc/flashheros/q/ssa") + ivoid="ivo://org.gavo.dc/__system__/obscore/obscore") return rsc.get_tables() -@pytest.mark.usefixtures("flash_tables") +@pytest.mark.usefixtures("obscore_tables") class TestGetTables: @pytest.mark.remote_data def test_get_tables_limit_enforced(self): @@ -869,47 +869,49 @@ def test_get_tables_limit_enforced(self): " Pass a higher table_limit to see them all.", str(excinfo.value)) @pytest.mark.remote_data - def test_get_tables_names(self, flash_tables): - assert (list(sorted(flash_tables.keys())) - == ["flashheros.data", "ivoa.obscore"]) + def test_get_tables_names(self, obscore_tables): + assert (list(sorted(obscore_tables.keys())) + == ["ivoa.obscore"]) @pytest.mark.remote_data - def test_get_tables_table_instance(self, flash_tables): - assert (flash_tables["ivoa.obscore"].name + def test_get_tables_table_instance(self, obscore_tables): + assert (obscore_tables["ivoa.obscore"].name == "ivoa.obscore") - assert (flash_tables["ivoa.obscore"].description - == "This data collection is queryable in GAVO Data" - " Center's obscore table.") - assert (flash_tables["flashheros.data"].title - == "Flash/Heros SSA table") + assert (obscore_tables["ivoa.obscore"].description[:42] + == "The IVOA-defined obscore table, containing") + assert (obscore_tables["ivoa.obscore"].title + == "GAVO Data Center Obscore Table") - assert (flash_tables["flashheros.data"].origin.ivoid - == "ivo://org.gavo.dc/flashheros/q/ssa") + assert (obscore_tables["ivoa.obscore"].origin.ivoid + == "ivo://org.gavo.dc/__system__/obscore/obscore") @pytest.mark.remote_data - def test_get_tables_column_meta(self, flash_tables): - def getflashcol(name): - for col in flash_tables["flashheros.data"].columns: + def test_get_tables_column_meta(self, obscore_tables): + def getcol(name): + for col in obscore_tables["ivoa.obscore"].columns: if name == col.name: return col raise KeyError(name) - assert getflashcol("accref").datatype.content == "char" - assert getflashcol("accref").datatype.arraysize == "*" + assert getcol("access_url").datatype.content == "char" + assert getcol("access_url").datatype.arraysize == "*" + + assert getcol("s_region").datatype._extendedtype == "adql:region" -# TODO: upstream bug: the following needs to fixed in DaCHS before -# the assertion passes - # assert getflashcol("ssa_region").datatype._extendedtype == "point" + assert getcol("access_format").ucd == 'meta.code.mime' - assert getflashcol("mime").ucd == 'meta.code.mime' + assert getcol("em_min").unit == "m" - assert getflashcol("ssa_specend").unit == "m" + assert (getcol("t_max").utype + == "obscore:char.timeaxis.coverage.bounds.limits.stoptime") - assert (getflashcol("ssa_specend").utype - == "ssa:char.spectralaxis.coverage.bounds.stop") + assert (getcol("t_exptime").description + == "Total exposure time") - assert (getflashcol("ssa_fluxcalib").description - == "Type of flux calibration") + @pytest.mark.remote_data + def test_get_tables_utype(self, obscore_tables): + assert (obscore_tables["ivoa.obscore"].utype + == "ivo://ivoa.net/std/obscore#table-1.1") @pytest.mark.remote_data