Initial symlink creation support

fs/base.py

@@ -310,6 +310,21 @@ def setinfo(self, path, info):
 
  """
 
+ def symlink(self, src, dst):
+ # type: (Text, Text) -> None
+ """Create a symbolic link pointing to src named dst.
+
+ Arguments:
+ src (str): Path to target
+ dst (str): Path to symlink
+
+ Raises:
+ fs.errors.ResourceNotFound: If ``path`` does not exist
+ on the filesystem
+ NotImplementedError: If not supported by underlying filesystem.
+ """
+ raise NotImplementedError()
+
  # ---------------------------------------------------------------- #
  # Optional methods #
  # Filesystems *may* implement these methods. #
@@ -709,6 +724,8 @@ def getmeta(self, namespace="standard"):
  read_only `True` if this filesystem is read only.
  supports_rename `True` if this filesystem supports an
  `os.rename` operation.
+ symlink `True` if this filesystem supports an
+ `os.symlink` operation.
  =================== ============================================
 
  Most builtin filesystems will provide all these keys, and third-

fs/osfs.py

@@ -146,6 +146,7 @@ def __init__(
  "thread_safe": True,
  "unicode_paths": os.path.supports_unicode_filenames,
  "virtual": False,
+ "symlink": bool(getattr(os, 'symlink', None)),
  }
 
  try:
@@ -676,3 +677,8 @@ def validatepath(self, path):
  " env var); {error}".format(path=path, error=error),
  )
  return super(OSFS, self).validatepath(path)
+
+ def symlink(self, src, dst):
+ _dst = self._to_sys_path(self.validatepath(dst))
+ with convert_os_errors("symlink", _dst):
+ return os.symlink(src, _dst)

fs/test.py

@@ -1852,3 +1852,35 @@ def test_hash(self):
  self.assertEqual(
  foo_fs.hash("hashme.txt", "md5"), "9fff4bb103ab8ce4619064109c54cb9c"
  )
+
+ def test_create_symlink(self):
+ if not self.fs.getmeta().get("symlink", False):
+ raise unittest.SkipTest("the filesystem does not support symlinks.")
+ self.fs.writetext('a', 'hello')
+ self.fs.symlink('a', 'b')
+ info = self.fs.getinfo('b', namespaces=['link'])
+ self.assertTrue(info.is_link)
+ self.assertEqual(info.target, 'a')
+ read_text = self.fs.readtext('b', 'utf8')
+ self.assertEqual('hello', read_text)
+
+ def test_create_symlink_dir(self):
+ if not self.fs.getmeta().get("symlink", False):
+ raise unittest.SkipTest("the filesystem does not support symlinks.")
+ self.fs.makedirs('a1/a2')
+ self.fs.writetext('a1/a2/file.txt', 'hello')
+ self.fs.makedir('b1')
+ self.fs.symlink('../a1/a2', 'b1/b2')
+ b1 = self.fs.opendir('b1')
+ self.assertTrue(b1.islink('b2'))
+ b2 = b1.opendir('b2')
+ self.assertEqual('hello', b2.readtext('file.txt', 'utf8'))
+
+ def test_create_symlink_target_exists(self):
+ # Check it behaves like os.symlink and ln -T:
+ # if target is directory don't create inside directory
+ if not self.fs.getmeta().get("symlink", False):
+ raise unittest.SkipTest("the filesystem does not support symlinks.")
+ self.fs.makedir('dst')
+ with self.assertRaises(errors.FileExists):
+ self.fs.symlink('src', 'dst')

fs/wrapfs.py

@@ -538,3 +538,7 @@ def hash(self, path, name):
  def walk(self):
  # type: () -> BoundWalker
  return self._wrap_fs.walker_class.bind(self)
+
+ def symlink(self, src, dst):
+ _fs, _dst = self.delegate_path(dst)
+ return _fs.symlink(src, _dst)