FM / fsspec

/usr/local/lib/python3.6/site-packages/fsspec/

/usr/local/lib/python3.6/site-packages/fsspec

Name	Size	Mode	Actions
implementations/	-	0755	rm
__pycache__/	-	0755	rm
archive.py	2150	0644	edit dl rm
asyn.py	27386	0644	edit dl rm
caching.py	16731	0644	edit dl rm
callbacks.py	4827	0644	edit dl rm
compression.py	4884	0644	edit dl rm
config.py	2900	0644	edit dl rm
conftest.py	1183	0644	edit dl rm
core.py	22102	0644	edit dl rm
dircache.py	2684	0644	edit dl rm
exceptions.py	348	0644	edit dl rm
fuse.py	10048	0644	edit dl rm
gui.py	13766	0644	edit dl rm
mapping.py	7517	0644	edit dl rm
parquet.py	19530	0644	edit dl rm
registry.py	9054	0644	edit dl rm
spec.py	56033	0644	edit dl rm
transaction.py	2195	0644	edit dl rm
utils.py	16848	0644	edit dl rm
_version.py	501	0644	edit dl rm
__init__.py	1557	0644	edit dl rm

Edit: /usr/local/lib/python3.6/site-packages/fsspec/conftest.py (1183B)

import os
import shutil
import subprocess
import sys
import time

import pytest

import fsspec
from fsspec.implementations.cached import CachingFileSystem

@pytest.fixture()
def m():
    """
    Fixture providing a memory filesystem.
    """
    m = fsspec.filesystem("memory")
    m.store.clear()
    m.pseudo_dirs.clear()
    try:
        yield m
    finally:
        m.store.clear()
        m.pseudo_dirs.clear()

@pytest.fixture
def ftp_writable(tmpdir):
    """
    Fixture providing a writable FTP filesystem.
    """
    pytest.importorskip("pyftpdlib")
    from fsspec.implementations.ftp import FTPFileSystem

FTPFileSystem.clear_instance_cache()  # remove lingering connections
    CachingFileSystem.clear_instance_cache()
    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

import os import shutil import subprocess import sys import time import pytest import fsspec from fsspec.implementations.cached import CachingFileSystem @pytest.fixture() def m(): """ Fixture providing a memory filesystem. """ m = fsspec.filesystem("memory") m.store.clear() m.pseudo_dirs.clear() try: yield m finally: m.store.clear() m.pseudo_dirs.clear() @pytest.fixture def ftp_writable(tmpdir): """ Fixture providing a writable FTP filesystem. """ pytest.importorskip("pyftpdlib") from fsspec.implementations.ftp import FTPFileSystem FTPFileSystem.clear_instance_cache() # remove lingering connections CachingFileSystem.clear_instance_cache() 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