Tests without network connection

upath/tests/implementations/test_github.py

@@ -1,3 +1,4 @@
+import functools
 import os
 import platform
 import sys
@@ -17,6 +18,56 @@
 )
 
 
+def has_internet_connection():
+    import requests
+
+    try:
+        requests.get("http://example.com")
+    except requests.exceptions.ConnectionError:
+        return False
+    else:
+        return True
+
+
+def xfail_on_github_rate_limit(func):
+    """
+    Method decorator to mark test as xfail when GitHub rate limit is exceeded.
+    """
+
+    @functools.wraps(func)
+    def wrapped_method(self, *args, **kwargs):
+        import requests
+
+        try:
+            return func(self, *args, **kwargs)
+        except AssertionError as e:
+            if "nodename nor servname provided, or not known" in str(e):
+                pytest.xfail(reason="No internet connection")
+            raise
+        except requests.exceptions.ConnectionError:
+            pytest.xfail(reason="No internet connection")
+        except Exception as e:
+            if "rate limit exceeded" in str(e):
+                pytest.xfail("GitHub API rate limit exceeded")
+            else:
+                raise
+
+    return wrapped_method
+
+
+def wrap_github_rate_limit_check(cls):
+    """
+    Class decorator to wrap all test methods with the
+    xfail_on_github_rate_limit decorator.
+    """
+    for attr_name in dir(cls):
+        if attr_name.startswith("test_"):
+            orig_method = getattr(cls, attr_name)
+            setattr(cls, attr_name, xfail_on_github_rate_limit(orig_method))
+    return cls
+
+
+@wrap_github_rate_limit_check
 class TestUPathGitHubPath(BaseTests):
     """
     Unit-tests for the GitHubPath implementation of UPath.

upath/tests/implementations/test_hf.py

@@ -1,3 +1,5 @@
+import functools
+
 import pytest
 from fsspec import get_filesystem_class
 
@@ -12,13 +14,44 @@
     pytestmark = pytest.mark.skip
 
 
+def xfail_on_hf_service_unavailable(func):
+    """
+    Method decorator to mark test as xfail when HuggingFace service is unavailable.
+    """
+    from httpx import HTTPStatusError
+
+    @functools.wraps(func)
+    def wrapped_method(self, *args, **kwargs):
+        try:
+            return func(self, *args, **kwargs)
+        except HTTPStatusError as err:
+            if err.response.status_code == 503:
+                pytest.xfail("HuggingFace API not reachable")
+            raise
+
+    return wrapped_method
+
+
 def test_hfpath():
     path = UPath("hf://HuggingFaceTB/SmolLM2-135M")
     assert isinstance(path, HfPath)
-    assert path.exists()
-
-
-class TestUPathHttp(BaseTests):
+    try:
+        assert path.exists()
+    except AssertionError:
+        from httpx import ConnectError
+        from huggingface_hub import HfApi
+
+        try:
+            HfApi().repo_info("HuggingFaceTB/SmolLM2-135M")
+        except ConnectError:
+            pytest.xfail("No internet connection")
+        except Exception as err:
+            if "Service Unavailable" in str(err):
+                pytest.xfail("HuggingFace API not reachable")
+            raise
+
+
+class TestUPathHf(BaseTests):
     @pytest.fixture(autouse=True, scope="function")
     def path(self, hf_fixture_with_readonly_mocked_hf_api):
         self.path = UPath(hf_fixture_with_readonly_mocked_hf_api)

upath/tests/implementations/test_http.py

@@ -17,7 +17,19 @@
     pytestmark = pytest.mark.skip
 
 
-def test_httppath():
+@pytest.fixture
+def internet_connection():
+    import requests
+
+    try:
+        requests.get("http://example.com")
+    except requests.exceptions.ConnectionError:
+        pytest.xfail(reason="No internet connection")
+    else:
+        yield
+
+
+def test_httppath(internet_connection):
     path = UPath("http://example.com")
     assert isinstance(path, HTTPPath)
     assert path.exists()