/usr/share/pyshared/ZODB/tests/Synchronization.py is in python-zodb 1:3.9.7-2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 | ##############################################################################
#
# 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.
#
##############################################################################
"""Test the storage's implemenetation of the storage synchronization spec.
The Synchronization spec
http://www.zope.org/Documentation/Developer/Models/ZODB/
ZODB_Architecture_Storage_Interface_State_Synchronization_Diag.html
It specifies two states committing and non-committing. A storage
starts in the non-committing state. tpc_begin() transfers to the
committting state; tpc_abort() and tpc_finish() transfer back to
non-committing.
Several other methods are only allowed in one state or another. Many
methods allowed only in the committing state require that they apply
to the currently committing transaction.
The spec is silent on a variety of methods that don't appear to modify
the state, e.g. load(), undoLog(), pack(). It's unclear whether there
is a separate set of synchronization rules that apply to these methods
or if the synchronization is implementation dependent, i.e. only what
is need to guarantee a corrected implementation.
The synchronization spec is also silent on whether there is any
contract implied with the caller. If the storage can assume that a
single client is single-threaded and that it will not call, e.g., store()
until after it calls tpc_begin(), the implementation can be
substantially simplified.
New and/or unspecified methods:
tpc_vote(): handled like tpc_abort
undo(): how's that handled?
Methods that have nothing to do with committing/non-committing:
load(), loadSerial(), getName(), getSize(), __len__(), history(),
undoLog(), pack().
Specific questions:
The spec & docs say that undo() takes three arguments, the second
being a transaction. If the specified arg isn't the current
transaction, the undo() should raise StorageTransactionError. This
isn't implemented anywhere. It looks like undo can be called at
anytime.
FileStorage does not allow undo() during a pack. How should this be
tested? Is it a general restriction?
"""
from transaction import Transaction
from ZODB.POSException import StorageTransactionError
OID = "\000" * 8
SERIALNO = "\000" * 8
TID = "\000" * 8
class SynchronizedStorage:
def verifyNotCommitting(self, callable, *args):
self.assertRaises(StorageTransactionError, callable, *args)
def verifyWrongTrans(self, callable, *args):
t = Transaction()
self._storage.tpc_begin(t)
self.assertRaises(StorageTransactionError, callable, *args)
self._storage.tpc_abort(t)
def checkStoreNotCommitting(self):
self.verifyNotCommitting(self._storage.store,
OID, SERIALNO, "", "", Transaction())
def checkStoreWrongTrans(self):
self.verifyWrongTrans(self._storage.store,
OID, SERIALNO, "", "", Transaction())
def checkAbortNotCommitting(self):
self._storage.tpc_abort(Transaction())
def checkAbortWrongTrans(self):
t = Transaction()
self._storage.tpc_begin(t)
self._storage.tpc_abort(Transaction())
self._storage.tpc_abort(t)
def checkFinishNotCommitting(self):
t = Transaction()
self._storage.tpc_finish(t)
self._storage.tpc_abort(t)
def checkFinishWrongTrans(self):
t = Transaction()
self._storage.tpc_begin(t)
self._storage.tpc_finish(Transaction())
self._storage.tpc_abort(t)
def checkBeginCommitting(self):
t = Transaction()
self._storage.tpc_begin(t)
self._storage.tpc_begin(t)
self._storage.tpc_abort(t)
# TODO: how to check undo?
|