Initial version of tests for FileStorage consistency checker

src/scripts/tests/testfstest.py

+"""Verify that fstest.py can find errors.
+
+XXX To run this test script fstest.py must be on your PYTHONPATH.
+"""
+
+from cStringIO import StringIO
+import os
+import re
+import struct
+import tempfile
+import unittest
+
+import fstest
+from fstest import FormatError, U64
+
+class TestCorruptedFS(unittest.TestCase):
+
+    # XXX path?
+    f = open('test-checker.fs', 'rb')
+    datafs = f.read()
+    f.close()
+    del f
+
+    def setUp(self):
+        self._temp = tempfile.mktemp()
+        self._file = open(self._temp, 'wb')
+
+    def tearDown(self):
+        if not self._file.closed:
+            self._file.close()
+        if os.path.exists(self._temp):
+            try:
+                os.remove(self._temp)
+            except os.error:
+                pass
+
+    def noError(self):
+        if not self._file.closed:
+            self._file.close()
+        fstest.check(self._temp)
+
+    def detectsError(self, rx):
+        if not self._file.closed:
+            self._file.close()
+        try:
+            fstest.check(self._temp)
+        except FormatError, msg:
+            mo = re.search(rx, str(msg))
+            self.failIf(mo is None, "unexpected error: %s" % msg)
+        else:
+            self.fail("fstest did not detect corruption")
+
+    def getHeader(self):
+        buf = self._datafs.read(16)
+        if not buf:
+            return 0, ''
+        tl = U64(buf[8:])
+        return tl, buf
+
+    def copyTransactions(self, n):
+        """Copy at most n transactions from the good data"""
+        f = self._datafs = StringIO(self.datafs)
+        self._file.write(f.read(4))
+        for i in range(n):
+            tl, data = self.getHeader()
+            if not tl:
+                return
+            self._file.write(data)
+            rec = f.read(tl - 8)
+            self._file.write(rec)
+
+    def testGood(self):
+        self._file.write(self.datafs)
+        self.noError()
+
+    def testTwoTransactions(self):
+        self.copyTransactions(2)
+        self.noError()
+
+    def testEmptyFile(self):
+        self.detectsError("empty file")
+
+    def testInvalidHeader(self):
+        self._file.write('SF12')
+        self.detectsError("invalid file header")
+
+    def testTruncatedTransaction(self):
+        self._file.write(self.datafs[:4+22])
+        self.detectsError("truncated")
+
+    def testCheckpointFlag(self):
+        self.copyTransactions(2)
+        tl, data = self.getHeader()
+        assert tl > 0, "ran out of good transaction data"
+        self._file.write(data)
+        self._file.write('c')
+        self._file.write(self._datafs.read(tl - 9))
+        self.detectsError("checkpoint flag")
+
+    def testInvalidStatus(self):
+        self.copyTransactions(2)
+        tl, data = self.getHeader()
+        assert tl > 0, "ran out of good transaction data"
+        self._file.write(data)
+        self._file.write('Z')
+        self._file.write(self._datafs.read(tl - 9))
+        self.detectsError("invalid status")
+
+    def testTruncatedRecord(self):
+        self.copyTransactions(3)
+        tl, data = self.getHeader()
+        assert tl > 0, "ran out of good transaction data"
+        self._file.write(data)
+        buf = self._datafs.read(tl / 2)
+        self._file.write(buf)
+        self.detectsError("truncated possibly")
+
+    def testBadLength(self):
+        self.copyTransactions(2)
+        tl, data = self.getHeader()
+        assert tl > 0, "ran out of good transaction data"
+        self._file.write(data)
+        buf = self._datafs.read(tl - 8)
+        self._file.write(buf[0])
+        assert tl <= 1<<16, "can't use this transaction for this test"
+        self._file.write("\777\777")
+        self._file.write(buf[3:])
+        self.detectsError("invalid transaction header")
+
+    def testDecreasingTimestamps(self):
+        self.copyTransactions(0)
+        tl, data = self.getHeader()
+        buf = self._datafs.read(tl - 8)
+        t1 = data + buf
+
+        tl, data = self.getHeader()
+        buf = self._datafs.read(tl - 8)
+        t2 = data + buf
+
+        self._file.write(t2[:8] + t1[8:])
+        self._file.write(t1[:8] + t2[8:])
+        self.detectsError("time-stamp")
+
+    def testTruncatedData(self):
+        # This test must re-write the transaction header length in
+        # order to trigger the error in check_drec().  If it doesn't,
+        # the truncated data record would also caught a truncated
+        # transaction record.
+        self.copyTransactions(1)
+        tl, data = self.getHeader()
+        pos = self._file.tell()
+        self._file.write(data)
+        buf = self._datafs.read(tl - 8)
+        hdr = buf[:15]
+        ul, dl, el = struct.unpack(">HHH", hdr[-6:])
+        self._file.write(buf[:15 + ul + dl + el])
+        data = buf[15 + ul + dl + el:]
+        self._file.write(data[:24])
+        self._file.seek(pos + 8, 0)
+        newlen = struct.pack(">II", 0, tl - (len(data) - 24))
+        self._file.write(newlen)
+        self.detectsError("truncated at")
+        
+    def testBadDataLength(self):
+        self.copyTransactions(1)
+        tl, data = self.getHeader()
+        self._file.write(data)
+        buf = self._datafs.read(tl - 8)
+        hdr = buf[:7]
+        # write the transaction meta data
+        ul, dl, el = struct.unpack(">HHH", hdr[-6:])
+        self._file.write(buf[:7 + ul + dl + el])
+
+        # write the first part of the data header
+        data = buf[7 + ul + dl + el:]
+        self._file.write(data[:24])
+        self._file.write("\000" * 4 + "\077" + "\000" * 3)
+        self._file.write(data[32:])
+        self.detectsError("record exceeds transaction")
+
+if __name__ == "__main__":
+    unittest.main()
+