Now properly warning when an update of a cached vocabulary fails.

pyvo/utils/tests/test_vocabularies_remote.py

@@ -8,12 +8,14 @@
 """
 
 import os
+import pathlib
 import time
 
 import pytest
 
 from astropy.utils import data
 
+from pyvo.dal.exceptions import PyvoUserWarning
 from pyvo.utils import vocabularies
 
 
@@ -56,3 +58,31 @@ def test_refreshing(self):
     def test_non_existing_voc(self):
         with pytest.raises(vocabularies.VocabularyError):
             vocabularies.get_vocabulary("not_an_ivoa_vocabulary")
+
+    def test_failed_update(self):
+        # Create a fake vocabulary and make it so old the machine
+        # will want to refresh it.
+        fake_voc = "http://www.ivoa.net/rdf/astropy-test-failure"
+
+        cache_dir = pathlib.Path(data._get_download_cache_loc()
+            )/data._url_to_dirname(fake_voc)
+        cache_dir.mkdir(exist_ok=True)
+
+        cache_name = cache_dir/"contents"
+        with open(cache_name, "w") as f:
+            f.write("{}")
+        with open(cache_dir/"url", "w") as f:
+            f.write(fake_voc)
+        os.utime(cache_name, (1000000000, 1000000000))
+
+        with pytest.warns(PyvoUserWarning) as msgs:
+            voc = vocabularies.get_vocabulary("astropy-test-failure")
+        # this sometimes catches a warning about an unclosed socket that,
+        # I think, originates somewhere else; let me work around it for
+        # the moment.
+        for msg in msgs:
+            if str(msg.message) == ("Updating cache for the vocabulary"
+                " astropy-test-failure failed: HTTP Error 404: Not Found"):
+                break
+        else:
+            raise AssertionError("No warning about failed cache update")

pyvo/utils/vocabularies.py

@@ -10,10 +10,14 @@
 import json
 import os
 import time
+import warnings
 from urllib import request
 
 from astropy.utils.data import download_file, clear_download_cache
 
+from pyvo.dal.exceptions import PyvoUserWarning
+
+
 IVOA_VOCABULARY_ROOT = "http://www.ivoa.net/rdf/"
 
 
@@ -54,8 +58,9 @@ def get_vocabulary(voc_name, force_update=False):
                 cache="update", show_progress=False,
                 http_headers={"accept": "application/x-desise+json"})
         except Exception as msg:
-            base.ui.notifyWarning("Updating cache for the vocabulary"
-                " {} failed: {}".format(voc_name, msg))
+            warnings.warn("Updating cache for the vocabulary"
+                f" {voc_name} failed: {msg}",
+                category=PyvoUserWarning)
 
     with open(src_name, "r", encoding="utf-8") as f:
         return json.load(f)