Relaxed baseurl requirements in SIA2Service

CHANGES.rst

@@ -1,6 +1,8 @@
 1.5 (unreleased)
 ================
 
+- Made SIA2Service accept access urls without finding them in the service capabilities [#500]
+
 - Add intersect modes for the spatial constraint in the registry module ``pyvo.registry.Spatial`` [#495]
 
 - Added ``alt_identifier``, ``created``, ``updated`` and ``rights`` to the

pyvo/dal/sia2.py

@@ -156,7 +156,7 @@ class SIA2Service(DALService, AvailabilityMixin, CapabilityMixin):
     generally not notice that, though.
     """
 
-    def __init__(self, baseurl, session=None):
+    def __init__(self, baseurl, session=None, check_baseurl=True):
         """
         instantiate an SIA2 service
 
@@ -166,6 +166,9 @@ def __init__(self, baseurl, session=None):
            url - URL of the SIA2service (base or query endpoint)
         session : object
            optional session to use for network requests
+        check_baseurl : bool
+           True - use the capabilities end point of the service to get the
+           query end point, False - baseurl is the query end point
         """
 
         super().__init__(baseurl, session=session)
@@ -177,22 +180,23 @@ def __init__(self, baseurl, session=None):
         if hasattr(self._session, 'update_from_capabilities'):
             self._session.update_from_capabilities(self.capabilities)
 
-        self.query_ep = None  # service query end point
-        for cap in self.capabilities:
-            # assumes that the access URL is the same regardless of the
-            # authentication method except BasicAA which is not supported
-            # in pyvo. So pick any access url as long as it's not
-            if cap.standardid.lower() == SIA2_STANDARD_ID.lower():
-                for interface in cap.interfaces:
-                    if interface.accessurls and \
-                            not (len(interface.securitymethods) == 1
-                                 and interface.securitymethods[0].standardid
-                                 == 'ivo://ivoa.net/sso#BasicAA'):
-                        self.query_ep = interface.accessurls[0].content
-                        break
-
-        if not self.query_ep:
-            raise AttributeError('No valid end point found')
+        self.query_ep = baseurl.strip('&')  # default service end point
+        if check_baseurl:
+            for cap in self.capabilities:
+                # assumes that the access URL is the same regardless of the
+                # authentication method except BasicAA which is not supported
+                # in pyvo. So pick any access url as long as it's not
+                if cap.standardid.lower() == SIA2_STANDARD_ID.lower():
+                    for interface in cap.interfaces:
+                        if interface.accessurls and \
+                                not (len(interface.securitymethods) == 1
+                                     and interface.securitymethods[0].standardid
+                                     == 'ivo://ivoa.net/sso#BasicAA'):
+                            self.query_ep = interface.accessurls[0].content
+                            break
+                    else:
+                        continue
+                    break
 
     def search(self, pos=None, band=None, time=None, pol=None,
                field_of_view=None, spatial_resolution=None,

pyvo/dal/tests/test_sia2.py

@@ -63,9 +63,6 @@ class TestSIA2Service:
     def test_capabilities(self):
         # this tests the SIA2 capabilities with various combinations:
 
-        # - sia-basicauth - one interfaces with BasicAA security method - this
-        #         should fail because pyvo does not support BasicAA
-
         with requests_mock.Mocker() as cm:
             cm.get('https://example.com/sia/capabilities',
                    content=get_pkg_data_contents('data/sia2/capabilities.xml'))
@@ -77,18 +74,15 @@ def test_capabilities(self):
                        'data/sia2/capabilities-newformat.xml'))
             cm.get('https://example.com/sia-priv/capabilities',
                    content=get_pkg_data_contents(
-                       'data/sia2/capabilities-priv.xml'))
+                       'data/sia2/capabilities-priv.xml')),
+            cm.get('https://example.com/sia/myquery/capabilities',
+                   content=get_pkg_data_contents('data/sia2/capabilities.xml'))
 
             # multiple interfaces with single security method each and
             # anonymous access.
             service = SIA2Service('https://example.com/sia')
             assert service.query_ep == 'https://example.com/sia/v2query'
 
-            # one interfaces with BasicAA security method - this should fail
-            # because pyvo does not support BasicAA
-            with pytest.raises(AttributeError):
-                service = SIA2Service('https://example.com/sia-basicauth')
-
             # one interface with multiple security methods
             service = SIA2Service('https://example.com/sia-newformat')
             assert service.query_ep == 'https://example.com/sia/v2query'
@@ -97,6 +91,16 @@ def test_capabilities(self):
             service = SIA2Service('https://example.com/sia-priv')
             assert service.query_ep == 'https://example.com/sia/v2query'
 
+            # any access point will be valid even when it contains query params
+            service = SIA2Service('https://example.com/sia/myquery?param=1')
+            assert service.query_ep == 'https://example.com/sia/v2query'
+
+            # capabilities checking is bypassed all together with the
+            # check_baseurl=False flag
+            service = SIA2Service('https://example.com/sia/myquery?param=1&',
+                                  check_baseurl=False)
+            assert service.query_ep == 'https://example.com/sia/myquery?param=1'
+
     POSITIONS = [(2, 4, 0.0166 * u.deg),
                  (12, 12.5, 34, 36),
                  (12.0 * u.deg, 34.0 * u.deg,

pyvo/dal/tests/test_sia2_remote.py

@@ -11,6 +11,7 @@
 
 from pyvo.dal.sia2 import search, SIA2Service
 from pyvo.dal.adhoc import DatalinkResults
+from pyvo import regsearch
 
 
 CADC_SIA_URL = 'https://ws.cadc-ccda.hia-iha.nrc-cnrc.gc.ca/sia'
@@ -254,3 +255,12 @@ def test_res_format(self):
         for rr in results:
             assert rr.access_format == \
                 'application/x-votable+xml;content=datalink'
+
+    @pytest.mark.filterwarnings("ignore::pyvo.dal.exceptions.DALOverflowWarning")
+    def test_reg_sia2(self):
+        image_services = regsearch(servicetype='sia2')
+        irsa_seip = \
+            [s for s in image_services if
+             'irsa' in s.ivoid and 'seip' in s.ivoid][0]
+        result = irsa_seip.search(pos=(2.8425, 74.4846, 10), maxrec=1)
+        assert len(result) == 1

pyvo/dal/vosi.py

@@ -4,6 +4,7 @@
 """
 from itertools import chain
 import requests
+from urllib.parse import urlparse
 
 from astropy.utils.decorators import lazyproperty, deprecated
 
@@ -20,12 +21,18 @@ class EndpointMixin():
     def _get_endpoint(self, endpoint):
         # finds the endpoint relative to the base url or its parent
         # and returns its content in raw format
+
+        # do not trust baseurl as it might contain query or fragments
+        urlcomp = urlparse(self.baseurl)
+        curated_baseurl = '{}://{}{}'.format(urlcomp.scheme,
+                                             urlcomp.hostname,
+                                             urlcomp.path)
         if not endpoint:
             raise AttributeError('endpoint required')
         ep_urls = [
-            '{baseurl}/{endpoint}'.format(baseurl=self.baseurl,
+            '{baseurl}/{endpoint}'.format(baseurl=curated_baseurl,
                                           endpoint=endpoint),
-            url_sibling(self.baseurl, endpoint)
+            url_sibling(curated_baseurl, endpoint)
         ]
 
         for ep_url in ep_urls:

pyvo/registry/regtap.py

@@ -371,7 +371,10 @@ def to_service(self):
             raise ValueError("PyVO has no support for interfaces with"
                              f" standard id {self.standard_id}.")
 
-        return service_class(self.access_url)
+        if service_class == sia2.SIA2Service:
+            return service_class(self.access_url, check_baseurl=False)
+        else:
+            return service_class(self.access_url)
 
     def supports(self, standard_id):
         """returns true if we believe the interface should be able to talk

pyvo/registry/tests/test_regtap.py

@@ -19,7 +19,7 @@
 from pyvo.registry import search as regsearch
 from pyvo.dal import DALOverflowWarning
 from pyvo.dal import query as dalq
-from pyvo.dal import tap
+from pyvo.dal import tap, sia2
 
 from astropy.utils.data import get_pkg_data_contents
 
@@ -239,6 +239,12 @@ def test_known_standard(self):
         assert isinstance(intf.to_service(), tap.TAPService)
         assert not intf.is_vosi
 
+    def test_sia2_standard(self):
+        intf = regtap.Interface("http://example.org",
+                                "ivo://ivoa.net/std/sia2", "vs:paramhttp", "std")
+        assert isinstance(intf.to_service(), sia2.SIA2Service)
+        assert not intf.is_vosi
+
     def test_secondary_interface(self):
         intf = regtap.Interface("http://example.org",
                                 "ivo://ivoa.net/std/tap#aux",