Spaces:
Sleeping
Sleeping
File size: 4,200 Bytes
8fd238c |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 |
import sys
from importlib.metadata import EntryPoint
from unittest.mock import create_autospec, patch
import pytest
import fsspec
from fsspec.implementations.zip import ZipFileSystem
from fsspec.registry import (
_registry,
filesystem,
get_filesystem_class,
known_implementations,
register_implementation,
registry,
)
from fsspec.spec import AbstractFileSystem
@pytest.fixture()
def clear_registry():
try:
yield
finally:
_registry.clear()
known_implementations.pop("test", None)
@pytest.fixture()
def clean_imports():
try:
real_module = sys.modules["fsspec"]
del sys.modules["fsspec"]
yield
finally:
sys.modules["fsspec"] = real_module
def test_registry_readonly():
get_filesystem_class("file")
assert "file" in registry
assert "file" in list(registry)
with pytest.raises(TypeError):
del registry["file"]
with pytest.raises(TypeError):
registry["file"] = None
with pytest.raises(AttributeError):
registry.clear()
def test_register_cls(clear_registry):
with pytest.raises(ValueError):
get_filesystem_class("test")
register_implementation("test", AbstractFileSystem)
cls = get_filesystem_class("test")
assert cls is AbstractFileSystem
def test_register_str(clear_registry):
with pytest.raises(ValueError):
get_filesystem_class("test")
register_implementation("test", "fsspec.AbstractFileSystem")
assert "test" not in registry
cls = get_filesystem_class("test")
assert cls is AbstractFileSystem
assert "test" in registry
def test_register_fail(clear_registry):
register_implementation("test", "doesntexist.AbstractFileSystem")
with pytest.raises(ImportError):
get_filesystem_class("test")
# NOOP
register_implementation("test", "doesntexist.AbstractFileSystem", clobber=False)
with pytest.raises(ValueError):
register_implementation(
"test", "doesntexist.AbstractFileSystemm", clobber=False
)
# by default we do not allow clobbering
with pytest.raises(ValueError):
register_implementation("test", "doesntexist.AbstractFileSystemm")
register_implementation(
"test", "doesntexist.AbstractFileSystem", errtxt="hiho", clobber=True
)
with pytest.raises(ImportError) as e:
get_filesystem_class("test")
assert "hiho" in str(e.value)
register_implementation("test", AbstractFileSystem)
# NOOP
register_implementation("test", AbstractFileSystem)
with pytest.raises(ValueError):
register_implementation("test", ZipFileSystem)
register_implementation("test", AbstractFileSystem, clobber=True)
assert isinstance(fsspec.filesystem("test"), AbstractFileSystem)
def test_entry_points_registered_on_import(clear_registry, clean_imports):
mock_ep = create_autospec(EntryPoint, module="fsspec.spec.AbstractFileSystem")
mock_ep.name = "test" # this can't be set in the constructor...
mock_ep.value = "fsspec.spec.AbstractFileSystem"
import_location = "importlib.metadata.entry_points"
with patch(import_location, return_value={"fsspec.specs": [mock_ep]}):
assert "test" not in registry
import fsspec # noqa
get_filesystem_class("test")
assert "test" in registry
def test_filesystem_warning_arrow_hdfs_deprecated(clear_registry, clean_imports):
mock_ep = create_autospec(EntryPoint, module="fsspec.spec.AbstractFileSystem")
mock_ep.name = "arrow_hdfs" # this can't be set in the constructor...
mock_ep.value = "fsspec.spec.AbstractFileSystem"
import_location = "importlib.metadata.entry_points"
with patch(import_location, return_value={"fsspec.specs": [mock_ep]}):
import fsspec # noqa
with pytest.warns(DeprecationWarning):
filesystem("arrow_hdfs")
def test_old_s3(monkeypatch):
from fsspec.registry import _import_class
s3fs = pytest.importorskip("s3fs")
monkeypatch.setattr(s3fs, "__version__", "0.4.2")
with pytest.warns():
_import_class("s3fs:S3FileSystem")
with pytest.warns():
_import_class("s3fs.S3FileSystem")
|