Add a test for uncovered remove-not-present-share case

Also fix the implementation to not blow up if the bucket for such a share
never existed.

src/allmydata/storage/server.py

@@ -482,8 +482,10 @@ class StorageServer(service.MultiService, Referenceable):
                 remaining_shares[sharenum] = shares[sharenum]
 
             if new_length == 0:
-                # delete empty bucket directories
-                if not os.listdir(bucketdir):
+                # delete bucket directories that exist but are empty.  They
+                # might not exist if a client showed up and asked us to
+                # truncate a share we weren't even holding.
+                if os.path.exists(bucketdir) and [] == os.listdir(bucketdir):
                     os.rmdir(bucketdir)
         return remaining_shares
 

src/allmydata/test/test_storage.py

@@ -1369,6 +1369,41 @@ class MutableServer(unittest.TestCase):
         self.failUnless(os.path.exists(prefixdir), prefixdir)
         self.failIf(os.path.exists(bucketdir), bucketdir)
 
+    def test_remove_non_present(self):
+        """
+        A write vector which would remove a share completely can be applied on a
+        server which does not have the share as a no-op.
+        """
+        ss = self.create("test_remove_non_present")
+
+        storage_index = "si1"
+        secrets = (
+            self.write_enabler(storage_index),
+            self.renew_secret(storage_index),
+            self.cancel_secret(storage_index),
+        )
+
+        sharenum = 3
+        testv = []
+        datav = []
+        new_length = 0
+        read_vector = []
+
+        # We don't even need to create any shares to exercise this
+        # functionality.  Just go straight to sending a truncate-to-zero
+        # write.
+        testv_is_good, read_data = ss.remote_slot_testv_and_readv_and_writev(
+            storage_index=storage_index,
+            secrets=secrets,
+            test_and_write_vectors={
+                sharenum: (testv, datav, new_length),
+            },
+            read_vector=read_vector,
+        )
+
+        self.assertTrue(testv_is_good)
+        self.assertEqual({}, read_data)
+
 
 class MDMFProxies(unittest.TestCase, ShouldFailMixin):
     def setUp(self):