repro

iterator_undo/repro_create_storage.py

-import ZODB
+from contextlib import closing
+from ZODB.FileStorage import FileStorage
+from ZODB import DB, TimeStamp
+import persistent
+import time
+import transaction
+from ZODB.utils import readable_tid_repr
+
+import logging
+
+logging.basicConfig(level=logging.INFO)
+
+
+class P(persistent.Persistent):
+    def __init__(self, value):
+        self.value = value
+
+
+def faketime():
+    # fake time to have a reproductible database
+    _xtime0 = time.mktime(time.strptime("04 Jan 1979", "%d %b %Y"))
+
+    def xtime_reset():
+        global _xtime
+        _xtime = _xtime0
+
+    xtime_reset()
+
+    def xtime():
+        global _xtime
+        _xtime += 1.1
+        return _xtime
+
+    time.time = xtime
+
+
+def change_object(value):
+    with closing(FileStorage(filename)) as stor, closing(DB(stor)) as db, closing(
+        db.open()
+    ) as conn:
+        transaction.get().note(value)
+        obj = conn.root()["obj"]
+        obj.value = value
+        transaction.commit()
+
+
+filename = "repro_initial.fs"
+filename_copy = "repro_copy.fs"
+
+
+with closing(FileStorage(filename, create=True)) as stor, closing(
+    DB(stor)
+) as db, closing(db.open()) as conn:
+    root = conn.root()
+    obj = P("__version_0__")
+    root["obj"] = obj
+    transaction.get().note("version 0")
+    transaction.commit()
+
+
+# make a first version of object
+with closing(FileStorage(filename)) as stor, closing(DB(stor)) as db, closing(
+    db.open()
+) as conn:
+    transaction.get().note("version 1")
+    obj = conn.root()["obj"]
+    obj.value = "__version_1__"
+    transaction.commit()
+
+
+# make a second version of object
+with closing(FileStorage(filename)) as stor, closing(DB(stor)) as db, closing(
+    db.open()
+) as conn:
+    transaction.get().note("version 2")
+    obj = conn.root()["obj"]
+    obj.value = "__version_2__"
+    transaction.commit()
+
+
+# undo these two changes ( go back to __version_0__)
+with closing(FileStorage(filename)) as stor, closing(DB(stor)) as db, closing(
+    db.open()
+) as conn:
+    undo_log = db.undoInfo(0, 2)
+    assert [u["description"] for u in undo_log] == ["version 2", "version 1"]
+
+    transaction.get().note("undo 2 and 1 (back to 0)")
+    db.undoMultiple([u["id"] for u in undo_log])
+    transaction.commit()
+
+
+# make a third version of object
+with closing(FileStorage(filename)) as stor, closing(DB(stor)) as db, closing(
+    db.open()
+) as conn:
+    transaction.get().note("version 3")
+    obj = conn.root()["obj"]
+    assert obj.value == "__version_0__", obj.value  # check undo was effective
+    obj.value = "__version_3__"
+    transaction.commit()
+
+
+# undo version 3 ( go back to __version_0__)
+with closing(FileStorage(filename)) as stor, closing(DB(stor)) as db, closing(
+    db.open()
+) as conn:
+    undo_log = db.undoInfo(0, 1)
+    assert [u["description"] for u in undo_log] == [
+        "version 3",
+    ]
+
+    transaction.get().note("undo 2 and 1 (back to 0)")
+    db.undoMultiple([u["id"] for u in undo_log])
+    transaction.commit()
+
+
+# check the state is OK
+with closing(FileStorage(filename)) as stor, closing(DB(stor)) as db, closing(
+    db.open()
+) as conn:
+    obj = conn.root()["obj"]
+    assert obj.value == "__version_0__", obj.value
+
+
+# copy this to a new storage
+with closing(FileStorage(filename)) as stor, closing(
+    FileStorage(filename_copy, create=True)
+) as stor_copy:
+    stor_copy.copyTransactionsFrom(stor)
+
+
+with closing(FileStorage(filename_copy)) as stor, closing(DB(stor)) as db, closing(
+    db.open()
+) as conn:
+    obj = conn.root()["obj"]
+    # AssertionError: __version_1__
+    assert obj.value == "__version_0__", obj.value