/usr/lib/python2.7/dist-packages/ZODB/tests/util.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 | ##############################################################################
#
# Copyright (c) 2003 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.
#
##############################################################################
"""Conventience function for creating test databases
"""
from __future__ import with_statement
from ZODB.MappingStorage import DB
import atexit
import os
import persistent
import sys
import tempfile
import time
import transaction
import unittest
import warnings
import ZODB.utils
import zope.testing.setupstack
def setUp(test, name='test'):
transaction.abort()
d = tempfile.mkdtemp(prefix=name)
zope.testing.setupstack.register(test, zope.testing.setupstack.rmtree, d)
zope.testing.setupstack.register(
test, setattr, tempfile, 'tempdir', tempfile.tempdir)
tempfile.tempdir = d
zope.testing.setupstack.register(test, os.chdir, os.getcwd())
os.chdir(d)
zope.testing.setupstack.register(test, transaction.abort)
tearDown = zope.testing.setupstack.tearDown
class TestCase(unittest.TestCase):
def setUp(self):
self.globs = {}
name = self.__class__.__name__
mname = getattr(self, '_TestCase__testMethodName', '')
if mname:
name += '-' + mname
setUp(self, name)
tearDown = tearDown
def pack(db):
db.pack(time.time()+1)
class P(persistent.Persistent):
def __init__(self, name=None):
self.name = name
def __repr__(self):
return 'P(%s)' % self.name
class MininalTestLayer:
__bases__ = ()
__module__ = ''
def __init__(self, name):
self.__name__ = name
def setUp(self):
self.here = os.getcwd()
self.tmp = tempfile.mkdtemp(self.__name__, dir=os.getcwd())
os.chdir(self.tmp)
# sigh. tearDown isn't called when a layer is run in a sub-process.
atexit.register(clean, self.tmp)
def tearDown(self):
os.chdir(self.here)
zope.testing.setupstack.rmtree(self.tmp)
testSetUp = testTearDown = lambda self: None
def clean(tmp):
if os.path.isdir(tmp):
zope.testing.setupstack.rmtree(tmp)
class AAAA_Test_Runner_Hack(unittest.TestCase):
"""Hack to work around a bug in the test runner.
The first later (lex sorted) is run first in the foreground
"""
layer = MininalTestLayer('!no tests here!')
def testNothing(self):
pass
def assert_warning(category, func, warning_text=''):
if sys.version_info < (2, 6):
return func() # Can't use catch_warnings :(
with warnings.catch_warnings(record=True) as w:
warnings.simplefilter('default')
result = func()
for warning in w:
if ((warning.category is category)
and (warning_text in str(warning.message))):
return result
raise AssertionError(w)
def assert_deprecated(func, warning_text=''):
return assert_warning(DeprecationWarning, func, warning_text)
def wait(func=None, timeout=30):
if func is None:
return lambda f: wait(f, timeout)
for i in xrange(int(timeout*100)):
if func():
return
time.sleep(.01)
raise AssertionError
def store(storage, oid, value='x', serial=ZODB.utils.z64):
if not isinstance(oid, str):
oid = ZODB.utils.p64(oid)
if not isinstance(serial, str):
serial = ZODB.utils.p64(serial)
t = transaction.get()
storage.tpc_begin(t)
storage.store(oid, serial, value, '', t)
storage.tpc_vote(t)
storage.tpc_finish(t)
def mess_with_time(test=None, globs=None, now=1278864701.5):
now = [now]
def faux_time():
now[0] += 1
return now[0]
if test is None and globs is not None:
# sigh
faux_time.globs = globs
test = faux_time
import time
zope.testing.setupstack.register(test, setattr, time, 'time', time.time)
time.time = faux_time
|