/usr/lib/python2.7/dist-packages/ZODB/FileStorage/tests.py is in python-zodb 1:3.10.7-1.
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 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 | ##############################################################################
#
# Copyright (c) Zope Corporation 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.
#
##############################################################################
import doctest
import os
import time
import transaction
import unittest
import ZODB.blob
import ZODB.FileStorage
import ZODB.tests.util
def pack_keep_old():
"""Should a copy of the database be kept?
The pack_keep_old constructor argument controls whether a .old file (and .old directory for blobs is kept.)
>>> fs = ZODB.FileStorage.FileStorage('data.fs', blob_dir='blobs')
>>> db = ZODB.DB(fs)
>>> conn = db.open()
>>> import ZODB.blob
>>> conn.root()[1] = ZODB.blob.Blob()
>>> conn.root()[1].open('w').write('some data')
>>> conn.root()[2] = ZODB.blob.Blob()
>>> conn.root()[2].open('w').write('some data')
>>> transaction.commit()
>>> conn.root()[1].open('w').write('some other data')
>>> del conn.root()[2]
>>> transaction.commit()
>>> old_size = os.stat('data.fs').st_size
>>> def get_blob_size(d):
... result = 0
... for path, dirs, file_names in os.walk(d):
... for file_name in file_names:
... result += os.stat(os.path.join(path, file_name)).st_size
... return result
>>> blob_size = get_blob_size('blobs')
>>> db.pack(time.time()+1)
>>> packed_size = os.stat('data.fs').st_size
>>> packed_size < old_size
True
>>> os.stat('data.fs.old').st_size == old_size
True
>>> packed_blob_size = get_blob_size('blobs')
>>> packed_blob_size < blob_size
True
>>> get_blob_size('blobs.old') == blob_size
True
>>> db.close()
>>> fs = ZODB.FileStorage.FileStorage('data.fs', blob_dir='blobs',
... create=True, pack_keep_old=False)
>>> db = ZODB.DB(fs)
>>> conn = db.open()
>>> conn.root()[1] = ZODB.blob.Blob()
>>> conn.root()[1].open('w').write('some data')
>>> conn.root()[2] = ZODB.blob.Blob()
>>> conn.root()[2].open('w').write('some data')
>>> transaction.commit()
>>> conn.root()[1].open('w').write('some other data')
>>> del conn.root()[2]
>>> transaction.commit()
>>> db.pack(time.time()+1)
>>> os.stat('data.fs').st_size == packed_size
True
>>> os.path.exists('data.fs.old')
False
>>> get_blob_size('blobs') == packed_blob_size
True
>>> os.path.exists('blobs.old')
False
>>> db.close()
"""
def pack_with_repeated_blob_records():
"""
There is a bug in ZEO that causes duplicate bloc database records
to be written in a blob store operation. (Maybe this has been
fixed by the time you read this, but there might still be
transactions in the wild that have duplicate records.
>>> fs = ZODB.FileStorage.FileStorage('t', blob_dir='bobs')
>>> db = ZODB.DB(fs)
>>> conn = db.open()
>>> conn.root()[1] = ZODB.blob.Blob()
>>> transaction.commit()
>>> tm = transaction.TransactionManager()
>>> oid = conn.root()[1]._p_oid
>>> blob_record, oldserial = fs.load(oid)
Now, create a transaction with multiple saves:
>>> trans = tm.begin()
>>> fs.tpc_begin(trans)
>>> open('ablob', 'w').write('some data')
>>> _ = fs.store(oid, oldserial, blob_record, '', trans)
>>> _ = fs.storeBlob(oid, oldserial, blob_record, 'ablob', '', trans)
>>> fs.tpc_vote(trans)
>>> fs.tpc_finish(trans)
>>> time.sleep(.01)
>>> db.pack()
>>> conn.sync()
>>> conn.root()[1].open().read()
'some data'
>>> db.close()
"""
def _save_index():
"""
_save_index can fail for large indexes.
>>> import ZODB.utils
>>> fs = ZODB.FileStorage.FileStorage('data.fs')
>>> t = transaction.begin()
>>> fs.tpc_begin(t)
>>> oid = 0
>>> for i in range(5000):
... oid += (1<<16)
... _ = fs.store(ZODB.utils.p64(oid), ZODB.utils.z64, 'x', '', t)
>>> fs.tpc_vote(t)
>>> fs.tpc_finish(t)
>>> import sys
>>> old_limit = sys.getrecursionlimit()
>>> sys.setrecursionlimit(50)
>>> fs._save_index()
Make sure we can restore:
>>> import logging
>>> handler = logging.StreamHandler(sys.stdout)
>>> logger = logging.getLogger('ZODB.FileStorage')
>>> logger.setLevel(logging.DEBUG)
>>> logger.addHandler(handler)
>>> index, pos, tid = fs._restore_index()
>>> index.items() == fs._index.items()
True
>>> pos, tid = fs._pos, fs._tid
cleanup
>>> fs.close()
>>> logger.setLevel(logging.NOTSET)
>>> logger.removeHandler(handler)
>>> sys.setrecursionlimit(old_limit)
"""
def test_suite():
return unittest.TestSuite((
doctest.DocFileSuite(
'zconfig.txt', 'iterator.test',
setUp=ZODB.tests.util.setUp, tearDown=ZODB.tests.util.tearDown,
),
doctest.DocTestSuite(
setUp=ZODB.tests.util.setUp, tearDown=ZODB.tests.util.tearDown,
),
))
|