Function offline

tests/utils.py:455–505

Simulate offline mode. There are three offline simulation modes: CONNECTION_FAILS (default mode): a ConnectionError is raised for each network call. CONNECTION_TIMES_OUT: a ReadTimeout or ConnectTimeout is raised for each network call. HF_HUB_OFFLINE_SET_TO_1: the HF_HUB_OFFLI

(mode: OfflineSimulationMode)

@contextmanager
def offline(mode: OfflineSimulationMode):
    """
    Simulate offline mode.

    There are three offline simulation modes:

    CONNECTION_FAILS (default mode): a ConnectionError is raised for each network call.
    CONNECTION_TIMES_OUT: a ReadTimeout or ConnectTimeout is raised for each network call.
    HF_HUB_OFFLINE_SET_TO_1: the HF_HUB_OFFLINE_SET_TO_1 environment variable is set to 1.
        This makes the http/ftp calls of the library instantly fail and raise an OfflineModeEnabled error.

    The raised exceptions are either from the `requests` library (if `huggingface_hub<1.0.0`)
    or from the `httpx` library (if `huggingface_hub>=1.0.0`).
    """
    # Enable offline mode
    if mode is OfflineSimulationMode.HF_HUB_OFFLINE_SET_TO_1:
        with patch("datasets.config.HF_HUB_OFFLINE", True):
            yield
        return

    # Determine which exception to raise based on mode

    def error_response(*args, **kwargs):
        if mode is OfflineSimulationMode.CONNECTION_FAILS:
            exc = httpx.ConnectError if IS_HF_HUB_1_x else requests.ConnectionError
        elif mode is OfflineSimulationMode.CONNECTION_TIMES_OUT:
            if kwargs.get("timeout") is None:
                raise RequestWouldHangIndefinitelyError(
                    "Tried an HTTP call in offline mode with no timeout set. Please set a timeout."
                )
            exc = httpx.ReadTimeout if IS_HF_HUB_1_x else requests.ConnectTimeout
        else:
            raise ValueError("Please use a value from the OfflineSimulationMode enum.")
        raise exc(f"Offline mode {mode}")

    # Patch all client methods to raise the appropriate error
    client_mock = Mock()
    for method in ["head", "get", "post", "put", "delete", "request", "stream"]:
        setattr(client_mock, method, Mock(side_effect=error_response))

    # Patching is slightly different depending on hfh internals
    patch_target = (
        {"target": "huggingface_hub.utils._http._GLOBAL_CLIENT", "new": client_mock}
        if IS_HF_HUB_1_x
        else {
            "target": "huggingface_hub.utils._http._get_session_from_cache",
            "return_value": client_mock,
        }
    )
    with patch(**patch_target):
        yield


@contextmanager