Added checks for content type for getdatalink

pyvo/dal/adhoc.py

@@ -307,6 +307,32 @@ class DatalinkRecordMixin:
 
     - ``getdataset()`` considers datalink.
     """
+    def getdatalinkurl(self):
+        """
+        look for and return the URL contained in record columns
+        to retrieve the datalink associated with this record.
+        DALServiceError is raised if no datalink column is found.
+        """
+        # First, check if the main url of the record is a datalink
+        access_format = self.getdataformat()
+        if ('datalink' in access_format) or ('votable' in access_format):
+            return self.getdataurl()
+
+        # Next, check if any columns are datalinks
+        for fieldname in self._results.fieldnames:
+            field = self._results.getdesc(fieldname)
+            if "meta.ref.url" in field.ucd:
+                out = self[fieldname]
+                if isinstance(out, bytes):
+                    out = out.decode('utf-8')
+                try:
+                    content = requests.get(out, stream=True).headers['Content-Type']
+                    if ('datalink' in content) or ('votable' in content):
+                        return out
+                except requests.exceptions.MissingSchema:
+                    pass
+
+        raise DALServiceError("No datalink found for record.")
 
     def getdatalink(self):
         try:
@@ -315,7 +341,11 @@ def getdatalink(self):
             query = DatalinkQuery.from_resource(self, datalink, session=self._session)
             return query.execute()
         except DALServiceError:
-            return DatalinkResults.from_result_url(self.getdataurl(), session=self._session)
+            datalink_url = self.getdatalinkurl()
+            return DatalinkResults.from_result_url(datalink_url, session=self._session)
+
+
+
 
     @stream_decode_content
     def getdataset(self, timeout=None):

pyvo/dal/tests/test_datalink.py

@@ -8,10 +8,11 @@
 import pytest
 
 import pyvo as vo
-from pyvo.dal.adhoc import DatalinkResults
+from pyvo.dal.adhoc import DatalinkResults, DALServiceError
 from pyvo.dal.sia2 import SIA2Results
 from pyvo.dal.tap import TAPResults
 from pyvo.utils import testing, vocabularies
+from pyvo.dal.sia import search
 
 from astropy.utils.data import get_pkg_data_contents, get_pkg_data_filename
 
@@ -307,3 +308,14 @@ def test_generic_record(self):
         assert len(links) == 1
         assert (next(links[0].bysemantics("#this"))["access_url"]
             == "http://dc.zah.uni-heidelberg.de/getproduct/flashheros/data/ca90/f0011.mt")
+
+
+@pytest.mark.usefixtures('sia', 'register_mocks')
+@pytest.mark.filterwarnings("ignore::astropy.io.votable.exceptions.W06")
+def test_search():
+    # for issue #328 getdatalink() exits messily when there isn't a datalink
+
+    results = search('http://example.com/sia', pos=(288, 15), format="all")
+    result = results[0]
+    with pytest.raises(DALServiceError, match="No datalink found for record."):
+        result.getdatalink()