refs: Clean-up parent directories when removing references

dulwich/refs.py

@@ -772,18 +772,41 @@ class DiskRefsContainer(RefsContainer):
                     orig_ref = self.get_packed_refs().get(name, ZERO_SHA)
                 if orig_ref != old_ref:
                     return False
-            # may only be packed
+
+            # remove the reference file itself
             try:
                 os.remove(filename)
             except OSError as e:
-                if e.errno != errno.ENOENT:
+                if e.errno != errno.ENOENT:  # may only be packed
                     raise
+
             self._remove_packed_ref(name)
             self._log(name, old_ref, None, committer=committer,
                       timestamp=timestamp, timezone=timezone, message=message)
         finally:
             # never write, we just wanted the lock
             f.abort()
+
+        # outside of the lock, clean-up any parent directory that might now
+        # be empty. this ensures that re-creating a reference of the same
+        # name of what was previously a directory works as expected
+        parent = name
+        while True:
+            try:
+                parent, _ = parent.rsplit(b'/', 1)
+            except ValueError:
+                break
+
+            parent_filename = self.refpath(parent)
+            try:
+                os.rmdir(parent_filename)
+            except OSError as e:
+                # this can be caused by the parent directory being
+                # removed by another process, being not empty, etc.
+                # in any case, this is non fatal because we already
+                # removed the reference, just ignore it
+                break
+
         return True
 
 

dulwich/tests/test_refs.py

@@ -462,6 +462,23 @@ class DiskRefsContainerTests(RefsContainerTests, TestCase):
                 b'df6800012397fb85c56e7418dd4eb9405dee075c'))
         self.assertRaises(KeyError, lambda: self._refs[b'refs/tags/refs-0.1'])
 
+    def test_remove_parent(self):
+        self._refs[b'refs/heads/foo/bar'] = (
+            b'df6800012397fb85c56e7418dd4eb9405dee075c'
+        )
+        del self._refs[b'refs/heads/foo/bar']
+        ref_file = os.path.join(
+            self._refs.path, b'refs', b'heads', b'foo', b'bar',
+        )
+        self.assertFalse(os.path.exists(ref_file))
+        ref_file = os.path.join(self._refs.path, b'refs', b'heads', b'foo')
+        self.assertFalse(os.path.exists(ref_file))
+        ref_file = os.path.join(self._refs.path, b'refs', b'heads')
+        self.assertTrue(os.path.exists(ref_file))
+        self._refs[b'refs/heads/foo'] = (
+            b'df6800012397fb85c56e7418dd4eb9405dee075c'
+        )
+
     def test_read_ref(self):
         self.assertEqual(b'ref: refs/heads/master',
                          self._refs.read_ref(b'HEAD'))