/usr/share/pyshared/axiom/test/util.py is in python-axiom 0.7.1-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 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 | """
Helpers for writing Axiom tests.
"""
from twisted.python.filepath import FilePath
from twisted.trial.unittest import SkipTest
from axiom.store import Store
_theBaseStorePaths = {}
def _getBaseStorePath(testCase, creator):
if creator not in _theBaseStorePaths:
s = creator(testCase)
_theBaseStorePaths[creator] = s.dbdir
s.close()
return _theBaseStorePaths[creator]
def getPristineStore(testCase, creator):
"""
Get an Axiom Store which has been created and initialized by C{creator} but
which has been otherwise untouched. If necessary, C{creator} will be
called to make one.
@type testCase: L{twisted.trial.unittest.TestCase}
@type creator: one-argument callable
@param creator: A factory for the Store configuration desired. Will be
invoked with the testCase instance if necessary.
@rtype: L{axiom.store.Store}
"""
dbdir = FilePath(testCase.mktemp())
basePath = _getBaseStorePath(testCase, creator)
basePath.copyTo(dbdir)
return Store(dbdir)
class CommandStubMixin:
"""
Pretend to be the parent command for a subcommand.
"""
def getStore(self):
# fake out "parent" implementation for stuff.
return self.store
def getSynopsis(self):
return '<CommandStubMixin>'
subCommand = property(lambda self: self.__class__.__name__)
class CommandStub(object):
"""
Mock for L{axiom.scripts.axiomatic.Options} which is always set as the
C{parent} attribute of an I{axiomatic} subcommand.
@ivar _store: The L{Store} associated which will be supplied to the
subcommand.
"""
def __init__(self, store, subCommand):
self._store = store
self.subCommand = subCommand
def getSynopsis(self):
return "Usage: axiomatic [options]"
def getStore(self):
return self._store
class QueryCounter:
"""
This is a counter object which measures the number of VDBE instructions
SQLite will execute to fulfill a particular query.
The count of VDBE instructions is very useful as a proxy for CPU time and
disk usage, because it (as opposed to CPU time and disk usage) is
deterministic between runs of a given query regardless of various accidents
of operating-system latency.
When creating data for a query involving a limit, start with B{more} Items
than will be returned by the limited query, not exactly the right number.
SQLite will do a little bit more work in the case where the limit restricts
the number of Items returned, and this will cause a test to fail even
though the performance characteristics being demonstrated are actually
correct.
Put another way, if you are testing::
s.query(MyItem, limit=5)
You should create six instances of C{MyItem} before the first C{measure}
call and then create one or more additional instances of C{MyItem} before
the second C{measure} call.
"""
def __init__(self, store):
"""
Create a new query counter and install it on the provided store.
@param store: an axiom L{Store}.
"""
self.reset()
self.store = store
c = self.store.connection._connection
# XXX: this only works with the pysqlite backend, even _with_ the hack
# detection; if we ever care about the apsw backend again, we should
# probably do something about adding the hack to it, adding this as a
# public Axiom API, or something.
sph = getattr(c, "set_progress_handler", None)
if sph is None:
raise SkipTest(
"QueryCounter requires PySQLite 2.4 or newer, or a patch "
"(see <http://initd.org/tracker/pysqlite/ticket/182>) to "
"expose the set_progress_handler API.")
sph(self.progressHandler, 1)
def progressHandler(self):
"""
This method will be called internally by SQLite for each bytecode executed.
It increments a counter.
@return: 0, aka SQLITE_OK, so that this does not abort the current
query.
"""
self.counter += 1
return 0
def reset(self):
"""Reset the internal counter to 0.
"""
self.counter = 0
def measure(self, f, *a, **k):
"""
The primary public API of this class, which runs a given function and
counts the number of bytecodes between its start and finish.
@return: an integer, the number of VDBE instructions executed.
"""
save = self.counter
self.reset()
try:
f(*a, **k)
finally:
result = self.counter
self.counter = save
return result
|