Implement TarFS.geturl and ZipFS.geturl and Fix #329, #333, #340 (#330)

* ✨ provide geturl for ReadZipFS. As a user of zipsf, I need geturl to provide FSURL string. * 🐛 on windows and python 3, fs.open_fs(osfs(~/).geturl('myfolder/subfolder')) triggers CreateFailed 🐛 osfs.geturl() cannot be opened by itself * 🔬 all test cases are in and ✨ support geturl for read tar file system * 🔥 remove unwanted comment in code * 📖 update change log and contributor md * :short: update code with black * 📖 update change log * 👕 provide type info * 💚 update unit tests * 🔥 remove dead code * 💚 update tarfs unit test * 🔥 remove unwanted change * :short: run black over osfs.py * 🐛 fix hidden exception at fs.close() when opening an absent zip/tar file URL. fix #333 * 📝 update the behavior of geturl of zipfs and tarfs * 👕 address review feedback ✨ url quote the files for proper url string * 💚 fix broken tests * ♿ add helpful exception info to help developers, who create pypifs, gitfs, fs.datalake et al. fix #340 * 🐛 fix windows path test * ✨ uniformly support fs purpose * 🔨 quote around the root path. #340 * 🚜 alternative file uri implementation * 🔬 try windows path test case where unicode characters stays as they are * 🐛 fix unit test expectation because of the difference between windows and linux file uri * 🚜 avoid Windows File URI for fs purpose * 🐛 before quote, utf8 string needs to be encoded. https://stackoverflow.com/questions/15115588/urllib-quote-throws-keyerror * 🚜 respect rfc 3986, where unicode will be quoted * 💚 🔨 code refactor and fix broken unit tests * 👕 address review feedback from @lurch * 💚 fix typo in code and 👕 update assertions * 🔥 remove unused variable * 👕 address further comments from @lurch * 💚 update windows test case. fix the typo * 🐛 colon:tmp is bad path under windows * 🐛 forward slash on Windows is a valid path separator * 💚 fix unit tests on travis-ci * 👕 address review comments * 👕 mypy compliance * 👕 dot the i and cross the t

Author: chfw

CHANGELOG.md

@@ -7,20 +7,26 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 
 ## [2.4.11] - Unreleased
 
+### Added
+
+- Added geturl for TarFS and ZipFS for 'fs' purpose. NoURL for 'download' purpose.
+- Added helpful root path in CreateFailed exception [#340](https://github.com/PyFilesystem/pyfilesystem2/issues/340)
+
 ### Fixed
 
 - Fixed tests leaving tmp files
 - Fixed typing issues
 - Fixed link namespace returning bytes
+- Fixed broken FSURL in windows [#329](https://github.com/PyFilesystem/pyfilesystem2/issues/329)
+- Fixed hidden exception at fs.close() when opening an absent zip/tar file URL [#333](https://github.com/PyFilesystem/pyfilesystem2/issues/333)
 - Fixed abstract class import from `collections` which would break on Python 3.8
 - Fixed incorrect imports of `mock` on Python 3
 - Removed some unused imports and unused `requirements.txt` file
 - Added mypy checks to Travis
 
 ### Changed
 
-Entire test suite has been migrated to [pytest](https://docs.pytest.org/en/latest/).
-Closes [#327](https://github.com/PyFilesystem/pyfilesystem2/issues/327).
+- Entire test suite has been migrated to [pytest](https://docs.pytest.org/en/latest/). Closes [#327](https://github.com/PyFilesystem/pyfilesystem2/issues/327).
 
 ## [2.4.10] - 2019-07-29
 

CONTRIBUTORS.md

@@ -2,6 +2,7 @@
 
 Many thanks to the following developers for contributing to this project:
 
+- [C. W.](https://github.com/chfw)
 - [Diego Argueta](https://github.com/dargueta)
 - [Geoff Jukes](https://github.com/geoffjukes)
 - [Giampaolo](https://github.com/gpcimino)

fs/_url_tools.py

@@ -0,0 +1,49 @@
+import re
+import six
+import platform
+
+if False: # typing.TYPE_CHECKING
+ from typing import Text, Union, BinaryIO
+
+_WINDOWS_PLATFORM = platform.system() == "Windows"
+
+
+def url_quote(path_snippet):
+ # type: (Text) -> Text
+ """
+ On Windows, it will separate drive letter and quote windows
+ path alone. No magic on Unix-alie path, just pythonic
+ `pathname2url`
+
+ Arguments:
+ path_snippet: a file path, relative or absolute.
+ """
+ if _WINDOWS_PLATFORM and _has_drive_letter(path_snippet):
+ drive_letter, path = path_snippet.split(":", 1)
+ if six.PY2:
+ path = path.encode("utf-8")
+ path = six.moves.urllib.request.pathname2url(path)
+ path_snippet = "{}:{}".format(drive_letter, path)
+ else:
+ if six.PY2:
+ path_snippet = path_snippet.encode("utf-8")
+ path_snippet = six.moves.urllib.request.pathname2url(path_snippet)
+ return path_snippet
+
+
+def _has_drive_letter(path_snippet):
+ # type: (Text) -> bool
+ """
+ The following path will get True
+ D:/Data
+ C:\\My Dcouments\\ test
+
+ And will get False
+
+ /tmp/abc:test
+
+ Arguments:
+ path_snippet: a file path, relative or absolute.
+ """
+ windows_drive_pattern = ".:[/\\\\].*$"
+ return re.match(windows_drive_pattern, path_snippet) is not None

fs/base.py

@@ -1633,7 +1633,7 @@ def hash(self, path, name):
  fs.errors.UnsupportedHash: If the requested hash is not supported.
 
  """
- _path = self.validatepath(path)
+ self.validatepath(path)
  try:
  hash_object = hashlib.new(name)
  except ValueError:

fs/osfs.py

@@ -39,7 +39,6 @@
  sendfile = None # type: ignore # pragma: no cover
 
 from . import errors
-from .errors import FileExists
 from .base import FS
 from .enums import ResourceType
 from ._fscompat import fsencode, fsdecode, fspath
@@ -49,6 +48,7 @@
 from .error_tools import convert_os_errors
 from .mode import Mode, validate_open_mode
 from .errors import FileExpected, NoURL
+from ._url_tools import url_quote
 
 if False: # typing.TYPE_CHECKING
  from typing import (
@@ -137,7 +137,8 @@ def __init__(
  )
  else:
  if not os.path.isdir(_root_path):
- raise errors.CreateFailed("root path does not exist")
+ message = "root path '{}' does not exist".format(_root_path)
+ raise errors.CreateFailed(message)
 
  _meta = self._meta = {
  "network": False,
@@ -526,7 +527,6 @@ def _scandir(self, path, namespaces=None):
  namespaces = namespaces or ()
  _path = self.validatepath(path)
  sys_path = self.getsyspath(_path)
- _sys_path = fsencode(sys_path)
  with convert_os_errors("scandir", path, directory=True):
  for entry_name in os.listdir(sys_path):
  _entry_name = fsdecode(entry_name)
@@ -584,9 +584,14 @@ def getsyspath(self, path):
 
  def geturl(self, path, purpose="download"):
  # type: (Text, Text) -> Text
- if purpose != "download":
+ sys_path = self.getsyspath(path)
+ if purpose == "download":
+ return "file://" + sys_path
+ elif purpose == "fs":
+ url_path = url_quote(sys_path)
+ return "osfs://" + url_path
+ else:
  raise NoURL(path, purpose)
- return "file://" + self.getsyspath(path)
 
  def gettype(self, path):
  # type: (Text) -> ResourceType

fs/tarfs.py

@@ -4,7 +4,6 @@
 from __future__ import print_function
 from __future__ import unicode_literals
 
-import copy
 import os
 import tarfile
 import typing
@@ -17,14 +16,14 @@
 from .base import FS
 from .compress import write_tar
 from .enums import ResourceType
-from .errors import IllegalBackReference
+from .errors import IllegalBackReference, NoURL
 from .info import Info
 from .iotools import RawWrapper
 from .opener import open_fs
-from .path import dirname, relpath, basename, isbase, normpath, parts, frombase
+from .path import relpath, basename, isbase, normpath, parts, frombase
 from .wrapfs import WrapFS
 from .permissions import Permissions
-
+from ._url_tools import url_quote
 
 if False: # typing.TYPE_CHECKING
  from tarfile import TarInfo
@@ -461,16 +460,25 @@ def removedir(self, path):
  def close(self):
  # type: () -> None
  super(ReadTarFS, self).close()
- self._tar.close()
+ if hasattr(self, "_tar"):
+ self._tar.close()
 
  def isclosed(self):
  # type: () -> bool
  return self._tar.closed # type: ignore
 
+ def geturl(self, path, purpose="download"):
+ # type: (Text, Text) -> Text
+ if purpose == "fs" and isinstance(self._file, six.string_types):
+ quoted_file = url_quote(self._file)
+ quoted_path = url_quote(path)
+ return "tar://{}!/{}".format(quoted_file, quoted_path)
+ else:
+ raise NoURL(path, purpose)
+
 
 if __name__ == "__main__": # pragma: no cover
  from fs.tree import render
- from fs.opener import open_fs
 
  with TarFS("tests.tar") as tar_fs:
  print(tar_fs.listdir("/"))

fs/zipfs.py

@@ -22,6 +22,7 @@
 from .path import dirname, forcedir, normpath, relpath
 from .time import datetime_to_epoch
 from .wrapfs import WrapFS
+from ._url_tools import url_quote
 
 if False: # typing.TYPE_CHECKING
  from typing import (
@@ -434,7 +435,8 @@ def removedir(self, path):
  def close(self):
  # type: () -> None
  super(ReadZipFS, self).close()
- self._zip.close()
+ if hasattr(self, "_zip"):
+ self._zip.close()
 
  def readbytes(self, path):
  # type: (Text) -> bytes
@@ -444,3 +446,12 @@ def readbytes(self, path):
  zip_name = self._path_to_zip_name(path)
  zip_bytes = self._zip.read(zip_name)
  return zip_bytes
+
+ def geturl(self, path, purpose="download"):
+ # type: (Text, Text) -> Text
+ if purpose == "fs" and isinstance(self._file, six.string_types):
+ quoted_file = url_quote(self._file)
+ quoted_path = url_quote(path)
+ return "zip://{}!/{}".format(quoted_file, quoted_path)
+ else:
+ raise errors.NoURL(path, purpose)

tests/test_osfs.py

@@ -7,13 +7,11 @@
 import shutil
 import tempfile
 import unittest
-
 import pytest
 
-from fs import osfs
-from fs.path import relpath
+from fs import osfs, open_fs
+from fs.path import relpath, dirname
 from fs import errors
-
 from fs.test import FSTestCases
 
 from six import text_type
@@ -77,7 +75,7 @@ def assert_text(self, path, contents):
 
  def test_not_exists(self):
  with self.assertRaises(errors.CreateFailed):
- fs = osfs.OSFS("/does/not/exists/")
+ osfs.OSFS("/does/not/exists/")
 
  def test_expand_vars(self):
  self.fs.makedir("TYRIONLANISTER")
@@ -162,3 +160,40 @@ def test_validatepath(self):
  with self.assertRaises(errors.InvalidCharsInPath):
  with self.fs.open("13 – Marked Register.pdf", "wb") as fh:
  fh.write(b"foo")
+
+ def test_consume_geturl(self):
+ self.fs.create("foo")
+ try:
+ url = self.fs.geturl("foo", purpose="fs")
+ except errors.NoURL:
+ self.assertFalse(self.fs.hasurl("foo"))
+ else:
+ self.assertTrue(self.fs.hasurl("foo"))
+
+ # Should not throw an error
+ base_dir = dirname(url)
+ open_fs(base_dir)
+
+ def test_complex_geturl(self):
+ self.fs.makedirs("foo/bar ha")
+ test_fixtures = [
+ # test file, expected url path
+ ["foo", "foo"],
+ ["foo-bar", "foo-bar"],
+ ["foo_bar", "foo_bar"],
+ ["foo/bar ha/barz", "foo/bar%20ha/barz"],
+ ["example b.txt", "example%20b.txt"],
+ ["exampleㄓ.txt", "example%E3%84%93.txt"],
+ ]
+ file_uri_prefix = "osfs://"
+ for test_file, relative_url_path in test_fixtures:
+ self.fs.create(test_file)
+ expected = file_uri_prefix + self.fs.getsyspath(relative_url_path).replace(
+ "\\", "/"
+ )
+ actual = self.fs.geturl(test_file, purpose="fs")
+
+ self.assertEqual(actual, expected)
+
+ def test_geturl_return_no_url(self):
+ self.assertRaises(errors.NoURL, self.fs.geturl, "test/path", "upload")

tests/test_tarfs.py

@@ -7,14 +7,14 @@
 import tarfile
 import tempfile
 import unittest
-
 import pytest
 
 from fs import tarfs
 from fs.enums import ResourceType
 from fs.compress import write_tar
 from fs.opener import open_fs
 from fs.opener.errors import NotWriteable
+from fs.errors import NoURL
 from fs.test import FSTestCases
 
 from .test_archives import ArchiveTestCases
@@ -93,15 +93,6 @@ def destroy_fs(self, fs):
  os.remove(fs._tar_file)
  del fs._tar_file
 
- def assert_is_bzip(self):
- try:
- tarfile.open(fs._tar_file, "r:gz")
- except tarfile.ReadError:
- self.fail("{} is not a valid gz archive".format(fs._tar_file))
- for other_comps in ["xz", "bz2", ""]:
- with self.assertRaises(tarfile.ReadError):
- tarfile.open(fs._tar_file, "r:{}".format(other_comps))
-
 
 @pytest.mark.skipif(six.PY2, reason="Python2 does not support LZMA")
 class TestWriteXZippedTarFS(FSTestCases, unittest.TestCase):
@@ -181,11 +172,44 @@ def test_read_from_filename(self):
  except:
  self.fail("Couldn't open tarfs from filename")
 
+ def test_read_non_existent_file(self):
+ fs = tarfs.TarFS(open(self._temp_path, "rb"))
+ # it has been very difficult to catch exception in __del__()
+ del fs._tar
+ try:
+ fs.close()
+ except AttributeError:
+ self.fail("Could not close tar fs properly")
+ except Exception:
+ self.fail("Strange exception in closing fs")
+
  def test_getinfo(self):
  super(TestReadTarFS, self).test_getinfo()
  top = self.fs.getinfo("top.txt", ["tar"])
  self.assertTrue(top.get("tar", "is_file"))
 
+ def test_geturl_for_fs(self):
+ test_fixtures = [
+ # test_file, expected
+ ["foo/bar/egg/foofoo", "foo/bar/egg/foofoo"],
+ ["foo/bar egg/foo foo", "foo/bar%20egg/foo%20foo"],
+ ]
+ tar_file_path = self._temp_path.replace("\\", "/")
+ for test_file, expected_file in test_fixtures:
+ expected = "tar://{tar_file_path}!/{file_inside_tar}".format(
+ tar_file_path=tar_file_path, file_inside_tar=expected_file
+ )
+ self.assertEqual(self.fs.geturl(test_file, purpose="fs"), expected)
+
+ def test_geturl_for_fs_but_file_is_binaryio(self):
+ self.fs._file = six.BytesIO()
+ self.assertRaises(NoURL, self.fs.geturl, "test", "fs")
+
+ def test_geturl_for_download(self):
+ test_file = "foo/bar/egg/foofoo"
+ with self.assertRaises(NoURL):
+ self.fs.geturl(test_file)
+
 
 class TestBrokenPaths(unittest.TestCase):
  @classmethod