[Zodb-checkins] CVS: StandaloneZODB/Tools/tests - testfstest.py:1.1

[Jeremy Hylton]

Update of /cvs-repository/StandaloneZODB/Tools/tests
In directory cvs.zope.org:/tmp/cvs-serv4299

Added Files:
	testfstest.py 
Log Message:
Initial version of tests for FileStorage consistency checker 




=== Added File StandaloneZODB/Tools/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()