def test_cachedir(self):
        mem_fs = open_fs("mem://")
        mem_fs.makedirs("foo/bar/baz")
        mem_fs.touch("egg")

        fs = wrap.cache_directory(mem_fs)
        self.assertEqual(sorted(fs.listdir("/")), ["egg", "foo"])
        self.assertEqual(sorted(fs.listdir("/")), ["egg", "foo"])
        self.assertTrue(fs.isdir("foo"))
        self.assertTrue(fs.isdir("foo"))
        self.assertTrue(fs.isfile("egg"))
        self.assertTrue(fs.isfile("egg"))

        self.assertEqual(fs.getinfo("foo"), mem_fs.getinfo("foo"))
        self.assertEqual(fs.getinfo("foo"), mem_fs.getinfo("foo"))

        self.assertEqual(fs.getinfo("/"), mem_fs.getinfo("/"))
        self.assertEqual(fs.getinfo("/"), mem_fs.getinfo("/"))

        with self.assertRaises(errors.ResourceNotFound):
            fs.getinfo("/foofoo") 

def test_entry_point_create_error(self):
        class BadOpener(opener.Opener):
            def __init__(self, *args, **kwargs):
                raise ValueError("some creation error")

            def open_fs(self, *args, **kwargs):
                pass

        entry_point = mock.MagicMock()
        entry_point.load = mock.MagicMock(return_value=BadOpener)
        iter_entry_points = mock.MagicMock(return_value=iter([entry_point]))

        with mock.patch("pkg_resources.iter_entry_points", iter_entry_points):
            with self.assertRaises(errors.EntryPointError) as ctx:
                opener.open_fs("test://")
            self.assertEqual(
                "could not instantiate opener; some creation error", str(ctx.exception)
            ) 

def test_copy_large(self):
        data1 = b"foo" * 512 * 1024
        data2 = b"bar" * 2 * 512 * 1024
        data3 = b"baz" * 3 * 512 * 1024
        data4 = b"egg" * 7 * 512 * 1024
        with open_fs("temp://") as src_fs:
            src_fs.writebytes("foo", data1)
            src_fs.writebytes("bar", data2)
            src_fs.makedir("dir1").writebytes("baz", data3)
            src_fs.makedirs("dir2/dir3").writebytes("egg", data4)
            for workers in (0, 1, 2, 4):
                with open_fs("temp://") as dst_fs:
                    fs.copy.copy_fs(src_fs, dst_fs, workers=workers)
                    self.assertEqual(dst_fs.readbytes("foo"), data1)
                    self.assertEqual(dst_fs.readbytes("bar"), data2)
                    self.assertEqual(dst_fs.readbytes("dir1/baz"), data3)
                    self.assertEqual(dst_fs.readbytes("dir2/dir3/egg"), data4) 

def test_copy_file_if_newer_same_fs(self):
        src_fs = open_fs("mem://")
        src_fs.makedir("foo2").touch("exists")
        src_fs.makedir("foo1").touch("test1.txt")
        src_fs.settimes(
            "foo2/exists", datetime.datetime.utcnow() + datetime.timedelta(hours=1)
        )
        self.assertTrue(
            fs.copy.copy_file_if_newer(
                src_fs, "foo1/test1.txt", src_fs, "foo2/test1.txt.copy"
            )
        )
        self.assertFalse(
            fs.copy.copy_file_if_newer(src_fs, "foo1/test1.txt", src_fs, "foo2/exists")
        )
        self.assertTrue(src_fs.exists("foo2/test1.txt.copy")) 

def test_copy_file_if_newer_dst_is_newer(self):
        try:
            src_dir = self._create_sandbox_dir()
            src_file1 = self._touch(src_dir, "file1.txt")
            self._write_file(src_file1)

            dst_dir = self._create_sandbox_dir()
            dst_file1 = self._touch(dst_dir, "file1.txt")
            self._write_file(dst_file1)

            src_fs = open_fs("osfs://" + src_dir)
            dst_fs = open_fs("osfs://" + dst_dir)

            self.assertTrue(dst_fs.exists("/file1.txt"))

            copied = fs.copy.copy_file_if_newer(
                src_fs, "/file1.txt", dst_fs, "/file1.txt"
            )

            self.assertEqual(copied, False)
        finally:
            shutil.rmtree(src_dir)
            shutil.rmtree(dst_dir) 

def test_copy_dir_if_newer_multiple_files(self):
        try:
            src_dir = self._create_sandbox_dir()
            src_fs = open_fs("osfs://" + src_dir)
            src_fs.makedirs("foo/bar")
            src_fs.makedirs("foo/empty")
            src_fs.touch("test.txt")
            src_fs.touch("foo/bar/baz.txt")

            dst_dir = self._create_sandbox_dir()
            dst_fs = open_fs("osfs://" + dst_dir)

            fs.copy.copy_dir_if_newer(src_fs, "/foo", dst_fs, "/")

            self.assertTrue(dst_fs.isdir("bar"))
            self.assertTrue(dst_fs.isdir("empty"))
            self.assertTrue(dst_fs.isfile("bar/baz.txt"))
        finally:
            shutil.rmtree(src_dir)
            shutil.rmtree(dst_dir) 

def open_fs(cls, *args, **kwargs):
        return cls(open_fs(*args, **kwargs)) 

def __init__(self, pyfs, *args, **kwargs):
        if isinstance(pyfs, str):
            # pyfs is an opener url
            self._pyfilesystem_instance = open_fs(pyfs, *args, **kwargs)
        elif isinstance(pyfs, type) and issubclass(pyfs, FS):
            # pyfs is an FS subclass
            self._pyfilesystem_instance = pyfs(*args, **kwargs)
        elif isinstance(pyfs, FS):
            # pyfs is a FS instance
            self._pyfilesystem_instance = pyfs
        else:
            raise TypeError("pyfs must be a url, an FS subclass, or an FS instance") 

def test_open_fs_url_strict_parameter_works(query_param, strict):
    fs = open_fs("gs://{}?{}".format(TEST_BUCKET, query_param))
    assert fs.strict == strict 

def test_open_fs_project_parameter_works(query_param, project):
    fs = open_fs("gs://{}?{}".format(TEST_BUCKET, query_param))
    assert fs.client.project == project 

def test_open_fs_api_endpoint_parameter_works():
    fs = open_fs("gs://{}?api_endpoint=http%3A//localhost%3A8888".format(TEST_BUCKET))
    assert fs.client.client_options == {"api_endpoint": "http://localhost:8888"} 

def default_fs_factory(path: str) -> FS:
    from fs import open_fs

    return open_fs(path, create=True) 

def __init__(
        self,
        src_path: str,
        src_fs_factory: Callable[..., FS] = default_fs_factory,
        output_prefix: str = "",
        dry_run: bool = False,
    ):
        """
        
        Args:
            src_path (str): root path of the directory to modify via staging technique
            src_fs_factory (Callable[..., FS], optional): Factory method for returning
def default_fs_factory(path: str) -> FS:
                PyFileSystem object. Defaults to default_fs_factory.
        """
        self.dry_run = dry_run
        self._deleted_files: List[str] = []
        self.src_path = src_path
        # self.src_fs = src_fs_factory(".")

        # The src_fs root is the primary reference path from which files are created,
        # diffed, etc
        self.src_fs = src_fs_factory(src_path or ".")

        # The stg_fs mirrors src_fs as an in-memory buffer of changes to be made
        self.stg_fs = fs.open_fs(f"mem://")

        # The render_fs is contained within stg_fs at the relative path `output_prefix`.
        # Rendering file system is from the frame-of-reference of the output_prefix
        if not self.stg_fs.isdir(output_prefix):
            self.stg_fs.makedirs(output_prefix)
        self.render_fs = self.stg_fs.opendir(output_prefix) 

def __init__(
        self,
        schematic_path: str,
        src_path: str = ".",
        output_prefix: str = "",
        dry_run: bool = False,
    ):
        from jinja2 import Environment
        from flaskerize.fileio import StagedFileSystem

        self.src_path = src_path
        self.output_prefix = output_prefix
        self.schematic_path = schematic_path
        self.schematic_files_path = os.path.join(
            self.schematic_path, self.DEFAULT_FILES_DIRNAME
        )

        self.schema_path = self._get_schema_path()
        self._load_schema()

        self.arg_parser = self._check_get_arg_parser()
        self.env = Environment()
        self.fs = StagedFileSystem(
            src_path=self.src_path, output_prefix=output_prefix, dry_run=dry_run
        )
        self.sch_fs = fs.open_fs(f"osfs://{self.schematic_files_path}")
        self.dry_run = dry_run 

def get_filesystem(path, create=False, **kwargs):
    """ A utility function for initializing any type of filesystem object with PyFilesystem2 package

    :param path: A filesystem path
    :type path: str
    :param create: If the filesystem path doesn't exist this flag indicates to either create it or raise an error
    :type create: bool
    :param kwargs: Any keyword arguments to be passed forward
    :return: A filesystem object
    :rtype: fs.FS
    """
    if path.startswith('s3://'):
        return load_s3_filesystem(path, *kwargs)

    return fs.open_fs(path, create=create, **kwargs) 

def test_import_path(self):
        """Test import fs also imports other symbols."""
        restore_fs = sys.modules.pop("fs")
        sys.modules.pop("fs.path")
        try:
            import fs

            fs.path
            fs.Seek
            fs.ResourceType
            fs.open_fs
        finally:
            sys.modules["fs"] = restore_fs 

def test_move_fs(self):
        src_fs = open_fs("mem://")
        src_fs.makedirs("foo/bar")
        src_fs.touch("test.txt")
        src_fs.touch("foo/bar/baz.txt")

        dst_fs = open_fs("mem://")
        fs.move.move_fs(src_fs, dst_fs)

        self.assertTrue(dst_fs.isdir("foo/bar"))
        self.assertTrue(dst_fs.isfile("test.txt"))
        self.assertTrue(src_fs.isempty("/")) 

def test_copy_dir(self):
        src_fs = open_fs("mem://")
        src_fs.makedirs("foo/bar")
        src_fs.touch("test.txt")
        src_fs.touch("foo/bar/baz.txt")

        dst_fs = open_fs("mem://")
        fs.move.move_dir(src_fs, "/foo", dst_fs, "/")

        self.assertTrue(dst_fs.isdir("bar"))
        self.assertTrue(dst_fs.isfile("bar/baz.txt"))
        self.assertFalse(src_fs.exists("foo")) 

def test_unknown_protocol(self):
        with self.assertRaises(errors.UnsupportedProtocol):
            opener.open_fs("unknown://") 

def test_entry_point_type_error(self):
        class NotAnOpener(object):
            pass

        entry_point = mock.MagicMock()
        entry_point.load = mock.MagicMock(return_value=NotAnOpener)
        iter_entry_points = mock.MagicMock(return_value=iter([entry_point]))

        with mock.patch("pkg_resources.iter_entry_points", iter_entry_points):
            with self.assertRaises(errors.EntryPointError) as ctx:
                opener.open_fs("test://")
            self.assertEqual("entry point did not return an opener", str(ctx.exception)) 

def test_install(self):
        """Test Registry.install works as a decorator."""
        registry = Registry()
        self.assertNotIn("foo", registry.protocols)

        @registry.install
        class FooOpener(opener.Opener):
            protocols = ["foo"]

            def open_fs(self, *args, **kwargs):
                pass

        self.assertIn("foo", registry.protocols) 

def test_open_osfs(self):
        fs = opener.open_fs("osfs://.")
        self.assertIsInstance(fs, OSFS)

        # test default protocol
        fs = opener.open_fs("./")
        self.assertIsInstance(fs, OSFS) 

def test_open_memfs(self):
        fs = opener.open_fs("mem://")
        self.assertIsInstance(fs, MemoryFS) 

def test_open_tarfs(self):
        fh, tar_name = tempfile.mkstemp(suffix=".tar.gz")
        os.close(fh)
        try:
            # Test creating tar
            with opener.open_fs("tar://" + tar_name, create=True) as make_tar:
                self.assertEqual(make_tar.compression, "gz")
                make_tar.writetext("foo.txt", "foofoo")
            # Test opening tar
            with opener.open_fs("tar://" + tar_name, writeable=False) as tar_fs:
                self.assertEqual(tar_fs.readtext("foo.txt"), "foofoo")
        finally:
            os.remove(tar_name) 

def test_open_fs(self):
        mem_fs = opener.open_fs("mem://")
        mem_fs_2 = opener.open_fs(mem_fs)
        self.assertEqual(mem_fs, mem_fs_2) 

def test_open_userdata(self):
        with self.assertRaises(errors.OpenerError):
            opener.open_fs("userdata://foo:bar:baz:egg")

        app_fs = opener.open_fs("userdata://fstest:willmcgugan:1.0", create=True)
        self.assertEqual(app_fs.app_dirs.appname, "fstest")
        self.assertEqual(app_fs.app_dirs.appauthor, "willmcgugan")
        self.assertEqual(app_fs.app_dirs.version, "1.0") 

def test_open_userdata_no_version(self):
        app_fs = opener.open_fs("userdata://fstest:willmcgugan", create=True)
        self.assertEqual(app_fs.app_dirs.appname, "fstest")
        self.assertEqual(app_fs.app_dirs.appauthor, "willmcgugan")
        self.assertEqual(app_fs.app_dirs.version, None) 

def test_user_data_opener(self):
        user_data_fs = open_fs("userdata://fstest:willmcgugan:1.0", create=True)
        self.assertIsInstance(user_data_fs, UserDataFS)
        user_data_fs.makedir("foo", recreate=True)
        user_data_fs.writetext("foo/bar.txt", "baz")
        user_data_fs_foo_dir = open_fs("userdata://fstest:willmcgugan:1.0/foo/")
        self.assertEqual(user_data_fs_foo_dir.readtext("bar.txt"), "baz") 

def test_open_ftp_proxy(self, mock_FTPFS):
        open_fs("ftp://foo:bar@ftp.example.org?proxy=ftp.proxy.org")
        mock_FTPFS.assert_called_once_with(
            "ftp.example.org",
            passwd="bar",
            port=21,
            user="foo",
            proxy="ftp.proxy.org",
            timeout=10,
        ) 

def setUp(self):
        fs = self.fs = open_fs("mem://")
        fs.writetext("foo.py", "Hello, World")
        fs.touch("bar.py")
        fs.touch("baz.py")
        fs.makedirs("egg")
        fs.writetext("egg/foo.py", "from fs import open_fs")
        fs.touch("egg/foo.pyc")
        fs.makedirs("a/b/c/").writetext("foo.py", "import fs")
        repr(fs.glob)