Merge pull request #95 from sruggier/pr/add-link-support

Implement hard link support

fake_filesystem_test.py

@@ -1740,6 +1740,88 @@ def testSymlink(self):
     self.assertTrue(self.os.path.lexists(file_path))
     self.assertTrue(self.os.path.exists(file_path))
 
+  # hard link related tests
+
+  def testLinkBogus(self):
+    # trying to create a link from a non-existent file should fail
+    self.assertRaises(OSError,
+                     self.os.link, '/nonexistent_source', '/link_dest')
+
+  def testLinkDelete(self):
+    fake_open = fake_filesystem.FakeFileOpen(self.filesystem)
+
+    file1_path = 'test_file1'
+    file2_path = 'test_file2'
+    contents1 = 'abcdef'
+    # Create file
+    self.filesystem.CreateFile(file1_path, contents=contents1)
+    # link to second file
+    self.os.link(file1_path, file2_path)
+    # delete first file
+    self.os.unlink(file1_path)
+    # assert that second file exists, and its contents are the same
+    self.assertTrue(self.os.path.exists(file2_path))
+    with fake_open(file2_path) as f:
+        self.assertEqual(f.read(), contents1)
+
+  def testLinkUpdate(self):
+    fake_open = fake_filesystem.FakeFileOpen(self.filesystem)
+
+    file1_path = 'test_file1'
+    file2_path = 'test_file2'
+    contents1 = 'abcdef'
+    contents2 = 'ghijkl'
+    # Create file and link
+    self.filesystem.CreateFile(file1_path, contents=contents1)
+    self.os.link(file1_path, file2_path)
+    # assert that the second file contains contents1
+    with fake_open(file2_path) as f:
+        self.assertEqual(f.read(), contents1)
+    # update the first file
+    with fake_open(file1_path, 'w') as f:
+        f.write(contents2)
+    # assert that second file contains contents2
+    with fake_open(file2_path) as f:
+        self.assertEqual(f.read(), contents2)
+
+  def testLinkNonExistentParent(self):
+    fake_open = fake_filesystem.FakeFileOpen(self.filesystem)
+
+    file1_path = 'test_file1'
+    breaking_link_path = 'nonexistent/test_file2'
+    contents1 = 'abcdef'
+    # Create file and link
+    self.filesystem.CreateFile(file1_path, contents=contents1)
+
+    # trying to create a link under a non-existent directory should fail
+    self.assertRaises(OSError,
+                     self.os.link, file1_path, breaking_link_path)
+
+  def testLinkCount1(self):
+    """Test that hard link counts are updated correctly."""
+    file1_path = 'test_file1'
+    file2_path = 'test_file2'
+    file3_path = 'test_file3'
+    self.filesystem.CreateFile(file1_path)
+    # initial link count should be one
+    self.assertEqual(self.os.stat(file1_path).st_nlink, 1)
+    self.os.link(file1_path, file2_path)
+    # the count should be incremented for each hard link created
+    self.assertEqual(self.os.stat(file1_path).st_nlink, 2)
+    self.assertEqual(self.os.stat(file2_path).st_nlink, 2)
+    # Check that the counts are all updated together
+    self.os.link(file2_path, file3_path)
+    self.assertEqual(self.os.stat(file1_path).st_nlink, 3)
+    self.assertEqual(self.os.stat(file2_path).st_nlink, 3)
+    self.assertEqual(self.os.stat(file3_path).st_nlink, 3)
+    # Counts should be decremented when links are removed
+    self.os.unlink(file3_path)
+    self.assertEqual(self.os.stat(file1_path).st_nlink, 2)
+    self.assertEqual(self.os.stat(file2_path).st_nlink, 2)
+    # check that it gets decremented correctly again
+    self.os.unlink(file1_path)
+    self.assertEqual(self.os.stat(file2_path).st_nlink, 1)
+
   def testUMask(self):
     umask = os.umask(0o22)
     os.umask(umask)

pyfakefs/fake_filesystem.py

@@ -210,10 +210,10 @@ def __init__(self, name, st_mode=stat.S_IFREG | PERM_DEF_FILE,
       self.st_size = len(contents)
     else:
       self.st_size = 0
+    self.st_nlink = 1
     # Non faked features, write setter methods for fakeing them
     self.st_ino = None
     self.st_dev = None
-    self.st_nlink = None
     self.st_uid = None
     self.st_gid = None
     # shall be set on creating the file from the file system to get access to fs available space
@@ -436,6 +436,11 @@ def RemoveEntry(self, pathname_name):
     """
     if pathname_name in self.contents and self.filesystem:
       self.filesystem.ChangeDiskUsage(-self.contents[pathname_name].GetSize())
+
+    entry = self.contents[pathname_name]
+    entry.st_nlink -= 1
+    assert entry.st_nlink >= 0
+
     del self.contents[pathname_name]
 
   def GetSize(self):
@@ -1245,6 +1250,49 @@ def CreateLink(self, file_path, link_target):
     return self.CreateFile(resolved_file_path, st_mode=stat.S_IFLNK | PERM_DEF,
                            contents=link_target)
 
+  def CreateHardLink(self, old_path, new_path):
+    """Create a hard link at new_path, pointing at old_path.
+
+    Args:
+      old_path: an existing link to the target file
+      new_path: the destination path to create a new link at
+
+    Returns:
+      the FakeFile object referred to by old_path
+
+    Raises:
+      OSError:  if something already exists at new_path
+      OSError:  if the parent directory doesn't exist
+    """
+    new_path_normalized = self.NormalizePath(new_path)
+    if self.Exists(new_path_normalized):
+      raise IOError(errno.EEXIST,
+                    'File already exists in fake filesystem',
+                    new_path)
+
+    new_parent_directory, new_basename = self.SplitPath(new_path_normalized)
+    if not new_parent_directory:
+      new_parent_directory = self.cwd
+
+    if not self.Exists(new_parent_directory):
+      raise OSError(errno.ENOENT, 'No such fake directory',
+                    new_parent_directory)
+
+    # Retrieve the target file
+    try:
+      old_file = self.GetObject(old_path)
+    except:
+      raise OSError(errno.ENOENT,
+                    'No such file or directory in fake filesystem',
+                    old_path)
+
+    old_file.st_nlink += 1
+
+    # abuse the name field to control the filename of the newly created link
+    old_file.name = new_basename
+    self.AddObject(new_parent_directory, old_file)
+    return old_file
+
   def __str__(self):
     return str(self.root)
 
@@ -2169,9 +2217,8 @@ def symlink(self, link_target, path):
     """
     self.filesystem.CreateLink(path, link_target)
 
-  # pylint: disable-msg=C6002
-  # TODO: Link doesn't behave like os.link, this needs to be fixed properly.
-  link = symlink
+  def link(self, oldpath, newpath):
+      self.filesystem.CreateHardLink(oldpath, newpath)
 
   def fsync(self, file_des):
     """Perform fsync for a fake file (in other words, do nothing).