File size: 3,446 Bytes
838f737 |
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 |
import os
import shutil
import subprocess
import sys
import time
from collections import deque
from collections.abc import Generator, Sequence
import pytest
import fsspec
@pytest.fixture()
def m():
"""
Fixture providing a memory filesystem.
"""
m = fsspec.filesystem("memory")
m.store.clear()
m.pseudo_dirs.clear()
m.pseudo_dirs.append("")
try:
yield m
finally:
m.store.clear()
m.pseudo_dirs.clear()
m.pseudo_dirs.append("")
class InstanceCacheInspector:
"""
Helper class to inspect instance caches of filesystem classes in tests.
"""
def clear(self) -> None:
"""
Clear instance caches of all currently imported filesystem classes.
"""
classes = deque([fsspec.spec.AbstractFileSystem])
while classes:
cls = classes.popleft()
cls.clear_instance_cache()
classes.extend(cls.__subclasses__())
def gather_counts(self, *, omit_zero: bool = True) -> dict[str, int]:
"""
Gather counts of filesystem instances in the instance caches
of all currently imported filesystem classes.
Parameters
----------
omit_zero:
Whether to omit instance types with no cached instances.
"""
out: dict[str, int] = {}
classes = deque([fsspec.spec.AbstractFileSystem])
while classes:
cls = classes.popleft()
count = len(cls._cache) # there is no public interface for the cache
# note: skip intermediate AbstractFileSystem subclasses
# if they proxy the protocol attribute via a property.
if isinstance(cls.protocol, (Sequence, str)):
key = cls.protocol if isinstance(cls.protocol, str) else cls.protocol[0]
if count or not omit_zero:
out[key] = count
classes.extend(cls.__subclasses__())
return out
@pytest.fixture(scope="function", autouse=True)
def instance_caches() -> Generator[InstanceCacheInspector, None, None]:
"""
Fixture to ensure empty filesystem instance caches before and after a test.
Used by default for all tests.
Clears caches of all imported filesystem classes.
Can be used to write test assertions about instance caches.
Usage:
def test_something(instance_caches):
# Test code here
fsspec.open("file://abc")
fsspec.open("memory://foo/bar")
# Test assertion
assert instance_caches.gather_counts() == {"file": 1, "memory": 1}
Returns
-------
instance_caches: An instance cache inspector for clearing and inspecting caches.
"""
ic = InstanceCacheInspector()
ic.clear()
try:
yield ic
finally:
ic.clear()
@pytest.fixture(scope="function")
def ftp_writable(tmpdir):
"""
Fixture providing a writable FTP filesystem.
"""
pytest.importorskip("pyftpdlib")
d = str(tmpdir)
with open(os.path.join(d, "out"), "wb") as f:
f.write(b"hello" * 10000)
P = subprocess.Popen(
[sys.executable, "-m", "pyftpdlib", "-d", d, "-u", "user", "-P", "pass", "-w"]
)
try:
time.sleep(1)
yield "localhost", 2121, "user", "pass"
finally:
P.terminate()
P.wait()
try:
shutil.rmtree(tmpdir)
except Exception:
pass
|