codekingpro/portable-devtools

1import os2import shutil3import subprocess4import sys5import time6from collections import deque7from collections.abc import Generator, Sequence8 9import pytest10 11import fsspec12 13 14@pytest.fixture()15def m():16    """17    Fixture providing a memory filesystem.18    """19    m = fsspec.filesystem("memory")20    m.store.clear()21    m.pseudo_dirs.clear()22    m.pseudo_dirs.append("")23    try:24        yield m25    finally:26        m.store.clear()27        m.pseudo_dirs.clear()28        m.pseudo_dirs.append("")29 30 31class InstanceCacheInspector:32    """33    Helper class to inspect instance caches of filesystem classes in tests.34    """35 36    def clear(self) -> None:37        """38        Clear instance caches of all currently imported filesystem classes.39        """40        classes = deque([fsspec.spec.AbstractFileSystem])41        while classes:42            cls = classes.popleft()43            cls.clear_instance_cache()44            classes.extend(cls.__subclasses__())45 46    def gather_counts(self, *, omit_zero: bool = True) -> dict[str, int]:47        """48        Gather counts of filesystem instances in the instance caches49        of all currently imported filesystem classes.50 51        Parameters52        ----------53        omit_zero:54            Whether to omit instance types with no cached instances.55        """56        out: dict[str, int] = {}57        classes = deque([fsspec.spec.AbstractFileSystem])58        while classes:59            cls = classes.popleft()60            count = len(cls._cache)  # there is no public interface for the cache61            # note: skip intermediate AbstractFileSystem subclasses62            #   if they proxy the protocol attribute via a property.63            if isinstance(cls.protocol, (Sequence, str)):64                key = cls.protocol if isinstance(cls.protocol, str) else cls.protocol[0]65                if count or not omit_zero:66                    out[key] = count67            classes.extend(cls.__subclasses__())68        return out69 70 71@pytest.fixture(scope="function", autouse=True)72def instance_caches() -> Generator[InstanceCacheInspector, None, None]:73    """74    Fixture to ensure empty filesystem instance caches before and after a test.75 76    Used by default for all tests.77    Clears caches of all imported filesystem classes.78    Can be used to write test assertions about instance caches.79 80    Usage:81 82        def test_something(instance_caches):83            # Test code here84            fsspec.open("file://abc")85            fsspec.open("memory://foo/bar")86 87            # Test assertion88            assert instance_caches.gather_counts() == {"file": 1, "memory": 1}89 90    Returns91    -------92    instance_caches: An instance cache inspector for clearing and inspecting caches.93    """94    ic = InstanceCacheInspector()95 96    ic.clear()97    try:98        yield ic99    finally:100        ic.clear()101 102 103@pytest.fixture(scope="function")104def ftp_writable(tmpdir):105    """106    Fixture providing a writable FTP filesystem.107    """108    pytest.importorskip("pyftpdlib")109 110    d = str(tmpdir)111    with open(os.path.join(d, "out"), "wb") as f:112        f.write(b"hello" * 10000)113    P = subprocess.Popen(114        [sys.executable, "-m", "pyftpdlib", "-d", d, "-u", "user", "-P", "pass", "-w"]115    )116    try:117        time.sleep(1)118        yield "localhost", 2121, "user", "pass"119    finally:120        P.terminate()121        P.wait()122        try:123            shutil.rmtree(tmpdir)124        except Exception:125            pass126