test_repairer: add a deterministic test of share data corruption that always flips...

diff --git a/src/allmydata/test/common.py b/src/allmydata/test/common.py
index f61e5ddef8253bd8181cb07054b1f7fd5e2a0fc5..b10509199fb5365cbf100d1f8c7db0906301f0d1 100644 (file)
--- a/src/allmydata/test/common.py
+++ b/src/allmydata/test/common.py
@@ -1200,6 +1200,22 @@ def _corrupt_share_data(data, debug=False):
 
         return corrupt_field(data, 0x0c+0x44, sharedatasize)
 
+def _corrupt_share_data_last_byte(data, debug=False):
+    """Scramble the file data -- flip all bits of the last byte."""
+    sharevernum = struct.unpack(">L", data[0x0c:0x0c+4])[0]
+    assert sharevernum in (1, 2), "This test is designed to corrupt immutable shares of v1 or v2 in specific ways, not v%d." % sharevernum
+    if sharevernum == 1:
+        sharedatasize = struct.unpack(">L", data[0x0c+0x08:0x0c+0x08+4])[0]
+        offset = 0x0c+0x24+sharedatasize-1
+    else:
+        sharedatasize = struct.unpack(">Q", data[0x0c+0x08:0x0c+0x0c+8])[0]
+        offset = 0x0c+0x44+sharedatasize-1
+
+    newdata = data[:offset] + chr(ord(data[offset])^0xFF) + data[offset+1:]
+    if debug:
+        log.msg("testing: flipping all bits of byte at offset %d: %r, newdata: %r" % (offset, data[offset], newdata[offset]))
+    return newdata
+
 def _corrupt_crypttext_hash_tree(data, debug=False):
     """Scramble the file data -- the field containing the crypttext hash tree
     will have one bit flipped or else will be changed to a random value.

diff --git a/src/allmydata/test/test_repairer.py b/src/allmydata/test/test_repairer.py
index 7efb4a82a41ea34c04b409d53b6fdcc179a706e9..94cd9ee4b39c1e752dbaa0f3cff9b35a3063f7ed 100644 (file)
--- a/src/allmydata/test/test_repairer.py
+++ b/src/allmydata/test/test_repairer.py
@@ -239,6 +239,11 @@ class Verifier(GridTestMixin, unittest.TestCase, RepairTestMixin):
         return self._help_test_verify(common._corrupt_share_data,
                                       self.judge_invisible_corruption)
 
+    def test_corrupt_sharedata_last_byte(self):
+        self.basedir = "repairer/Verifier/corrupt_sharedata_last_byte"
+        return self._help_test_verify(common._corrupt_share_data_last_byte,
+                                      self.judge_invisible_corruption)
+
     def test_corrupt_ueb_length(self):
         self.basedir = "repairer/Verifier/corrupt_ueb_length"
         return self._help_test_verify(common._corrupt_length_of_uri_extension,