text: ##############################################################################
  #
  # Copyright (c) 2001, 2002 Zope Foundation and Contributors.
  # All Rights Reserved.
  #
  # This software is subject to the provisions of the Zope Public License,
  # Version 2.1 (ZPL).  A copy of the ZPL should accompany this distribution.
  # THIS SOFTWARE IS PROVIDED "AS IS" AND ANY AND ALL EXPRESS OR IMPLIED
  # WARRANTIES ARE DISCLAIMED, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
  # WARRANTIES OF TITLE, MERCHANTABILITY, AGAINST INFRINGEMENT, AND FITNESS
  # FOR A PARTICULAR PURPOSE.
  #
  ##############################################################################
  """Check loadSerial() on storages that support historical revisions."""
  
  from ZODB.Connection import TransactionMetaData
  from ZODB.tests.MinPO import MinPO
  from ZODB.tests.StorageTestBase import zodb_unpickle, zodb_pickle, snooze
  from ZODB.tests.StorageTestBase import ZERO
  from ZODB.utils import p64, u64, load_current
  from ZODB.tests.util import time_monotonically_increases
  
  
  class RevisionStorage(object):
  
      def checkLoadSerial(self):
          oid = self._storage.new_oid()
          revid = ZERO
          revisions = {}
          for i in range(31, 38):
              revid = self._dostore(oid, revid=revid, data=MinPO(i))
              revisions[revid] = MinPO(i)
          # Now make sure all the revisions have the correct value
          for revid, value in revisions.items():
              data = self._storage.loadSerial(oid, revid)
              self.assertEqual(zodb_unpickle(data), value)
  
      @time_monotonically_increases
      def checkLoadBefore(self):
          # Store 10 revisions of one object and then make sure that we
          # can get all the non-current revisions back.
          oid = self._storage.new_oid()
          revs = []
          revid = None
          for i in range(10):
              # We need to ensure that successive timestamps are at least
              # two apart, so that a timestamp exists that's unambiguously
              # between successive timestamps.  Each call to snooze()
              # guarantees that the next timestamp will be at least one
              # larger (and probably much more than that) than the previous
              # one.
              snooze()
              snooze()
              revid = self._dostore(oid, revid, data=MinPO(i))
              revs.append(load_current(self._storage, oid))
  
          prev = u64(revs[0][1])
          for i in range(1, 10):
              tid = revs[i][1]
              cur = u64(tid)
              middle = prev + (cur - prev) // 2
              assert prev < middle < cur  # else the snooze() trick failed
              prev = cur
              t = self._storage.loadBefore(oid, p64(middle))
              self.assertTrue(t is not None)
              data, start, end = t
              self.assertEqual(revs[i-1][0], data)
              self.assertEqual(tid, end)
  
      def checkLoadBeforeEdges(self):
          # Check the edges cases for a non-current load.
          oid = self._storage.new_oid()
  
          self.assertRaises(KeyError, self._storage.loadBefore,
                            oid, p64(0))
  
          revid1 = self._dostore(oid, data=MinPO(1))
  
          self.assertEqual(self._storage.loadBefore(oid, p64(0)), None)
          self.assertEqual(self._storage.loadBefore(oid, revid1), None)
  
          cur = p64(u64(revid1) + 1)
          data, start, end = self._storage.loadBefore(oid, cur)
          self.assertEqual(zodb_unpickle(data), MinPO(1))
          self.assertEqual(start, revid1)
          self.assertEqual(end, None)
  
          revid2 = self._dostore(oid, revid=revid1, data=MinPO(2))
          data, start, end = self._storage.loadBefore(oid, cur)
          self.assertEqual(zodb_unpickle(data), MinPO(1))
          self.assertEqual(start, revid1)
          self.assertEqual(end, revid2)
  
      @time_monotonically_increases
      def checkLoadBeforeOld(self):
          # Look for a very old revision.  With the BaseStorage implementation
          # this should require multple history() calls.
          oid = self._storage.new_oid()
          revs = []
          revid = None
          for i in range(50):
              revid = self._dostore(oid, revid, data=MinPO(i))
              revs.append(revid)
  
          data, start, end = self._storage.loadBefore(oid, revs[12])
          self.assertEqual(zodb_unpickle(data), MinPO(11))
          self.assertEqual(start, revs[11])
          self.assertEqual(end, revs[12])
  
      # Unsure:  Is it okay to assume everyone testing against RevisionStorage
      # implements undo?
  
      def checkLoadBeforeUndo(self):
          # Do several transactions then undo them.
          oid = self._storage.new_oid()
          revid = None
          for i in range(5):
              revid = self._dostore(oid, revid, data=MinPO(i))
          revs = []
          for i in range(4):
              info = self._storage.undoInfo()
              tid = info[0]["id"]
              # Always undo the most recent txn, so the value will
              # alternate between 3 and 4.
              self._undo(tid, note="undo %d" % i)
              revs.append(load_current(self._storage, oid))
  
          prev_tid = None
          for i, (data, tid) in enumerate(revs):
              t = self._storage.loadBefore(oid, p64(u64(tid) + 1))
              self.assertEqual(data, t[0])
              self.assertEqual(tid, t[1])
              if prev_tid:
                  self.assertTrue(prev_tid < t[1])
              prev_tid = t[1]
              if i < 3:
                  self.assertEqual(revs[i+1][1], t[2])
              else:
                  self.assertEqual(None, t[2])
  
      def checkLoadBeforeConsecutiveTids(self):
          eq = self.assertEqual
          oid = self._storage.new_oid()
  
          def helper(tid, revid, x):
              data = zodb_pickle(MinPO(x))
              t = TransactionMetaData()
              try:
                  self._storage.tpc_begin(t, p64(tid))
                  self._storage.store(oid, revid, data, '', t)
                  # Finish the transaction
                  self._storage.tpc_vote(t)
                  newrevid = self._storage.tpc_finish(t)
              except:  # noqa: E722 do not use bare 'except'
                  self._storage.tpc_abort(t)
                  raise
              return newrevid
          revid1 = helper(1, None, 1)
          revid2 = helper(2, revid1, 2)
          helper(3, revid2, 3)
          data, start_tid, end_tid = self._storage.loadBefore(oid, p64(2))
          eq(zodb_unpickle(data), MinPO(1))
          eq(u64(start_tid), 1)
          eq(u64(end_tid), 2)
  
      def checkLoadBeforeCreation(self):
          eq = self.assertEqual
          oid1 = self._storage.new_oid()
          oid2 = self._storage.new_oid()
          self._dostore(oid1)
          revid2 = self._dostore(oid2)
          results = self._storage.loadBefore(oid2, revid2)
          eq(results, None)
  
      # TODO:  There are other edge cases to handle, including pack.
  