/usr/lib/python3/dist-packages/repoze/tm/tests.py is in python3-repoze.tm2 2.0-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 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 | import unittest
class TestTM(unittest.TestCase):
def _getTargetClass(self):
from repoze.tm import TM
return TM
def _start_response(self, status, headers, exc_info=None):
pass
def _makeOne(self, app, commit_veto=None):
return self._getTargetClass()(app, commit_veto)
def test_ekey_inserted(self):
app = DummyApplication()
tm = self._makeOne(app)
tm.transaction = DummyTransactionModule()
from repoze.tm import ekey
env = {}
result = [chunk for chunk in tm(env, self._start_response)]
self.assertEqual(result, ['hello'])
self.assertTrue(ekey in env)
def test_committed(self):
app = DummyApplication()
tm = self._makeOne(app)
transaction = DummyTransactionModule()
tm.transaction = transaction
result = [chunk for chunk in tm({}, self._start_response)]
self.assertEqual(result, ['hello'])
self.assertEqual(transaction.committed, True)
self.assertEqual(transaction.aborted, False)
def test_aborted_via_doom(self):
app = DummyApplication()
tm = self._makeOne(app)
transaction = DummyTransactionModule(doom=True)
tm.transaction = transaction
result = [chunk for chunk in tm({}, self._start_response)]
self.assertEqual(result, ['hello'])
self.assertEqual(transaction.committed, False)
self.assertEqual(transaction.aborted, True)
def test_aborted_via_exception(self):
app = DummyApplication(exception=True)
tm = self._makeOne(app)
transaction = DummyTransactionModule()
tm.transaction = transaction
def execute_request():
[chunk for chunk in tm({}, self._start_response)]
self.assertRaises(ValueError, execute_request)
self.assertEqual(transaction.committed, False)
self.assertEqual(transaction.aborted, True)
def test_aborted_via_exception_and_doom(self):
app = DummyApplication(exception=True)
tm = self._makeOne(app)
transaction = DummyTransactionModule(doom=True)
tm.transaction = transaction
def execute_request():
[chunk for chunk in tm({}, self._start_response)]
self.assertRaises(ValueError, execute_request)
self.assertEqual(transaction.committed, False)
self.assertEqual(transaction.aborted, True)
def test_aborted_via_commit_veto(self):
app = DummyApplication(status="403 Forbidden")
def commit_veto(environ, status, headers):
self.assertTrue(isinstance(environ, dict),
"environ is not passed properly")
self.assertTrue(isinstance(headers, list),
"headers are not passed properly")
self.assertTrue(isinstance(status, str),
"status is not passed properly")
return not (200 <= int(status.split()[0]) < 400)
tm = self._makeOne(app, commit_veto)
transaction = DummyTransactionModule()
tm.transaction = transaction
[chunk for chunk in tm({}, self._start_response)]
self.assertEqual(transaction.committed, False)
self.assertEqual(transaction.aborted, True) # ici
def test_committed_via_commit_veto_exception(self):
app = DummyApplication(status="403 Forbidden")
def commit_veto(environ, status, headers):
return None
tm = self._makeOne(app, commit_veto)
transaction = DummyTransactionModule()
tm.transaction = transaction
[chunk for chunk in tm({}, self._start_response)]
self.assertEqual(transaction.committed, True)
self.assertEqual(transaction.aborted, False)
def test_aborted_via_commit_veto_exception(self):
app = DummyApplication(status="403 Forbidden")
def commit_veto(environ, status, headers):
raise ValueError('foo')
tm = self._makeOne(app, commit_veto)
transaction = DummyTransactionModule()
tm.transaction = transaction
def execute_request():
[chunk for chunk in tm({}, self._start_response)]
self.assertRaises(ValueError, execute_request)
self.assertEqual(transaction.committed, False)
self.assertEqual(transaction.aborted, True)
def test_cleanup_on_commit(self):
from repoze.tm import after_end
dummycalled = []
def dummy():
dummycalled.append(True)
env = {}
app = DummyApplication()
tm = self._makeOne(app)
transaction = DummyTransactionModule()
setattr(transaction, after_end.key, [dummy])
tm.transaction = transaction
[chunk for chunk in tm(env, self._start_response)]
self.assertEqual(transaction.committed, True)
self.assertEqual(transaction.aborted, False)
self.assertEqual(dummycalled, [True])
def test_cleanup_on_abort(self):
from repoze.tm import after_end
dummycalled = []
def dummy():
dummycalled.append(True)
env = {}
app = DummyApplication(exception=True)
tm = self._makeOne(app)
transaction = DummyTransactionModule()
setattr(transaction, after_end.key, [dummy])
tm.transaction = transaction
def execute_request():
[chunk for chunk in tm(env, self._start_response)]
self.assertRaises(ValueError, execute_request)
self.assertEqual(transaction.committed, False)
self.assertEqual(transaction.aborted, True)
self.assertEqual(dummycalled, [True])
class TestAfterEnd(unittest.TestCase):
def _getTargetClass(self):
from repoze.tm import AfterEnd
return AfterEnd
def _makeOne(self):
return self._getTargetClass()()
def test_register(self):
registry = self._makeOne()
func = lambda *x: None
txn = Dummy()
registry.register(func, txn)
self.assertEqual(getattr(txn, registry.key), [func])
def test_unregister_exists(self):
registry = self._makeOne()
func = lambda *x: None
txn = Dummy()
registry.register(func, txn)
self.assertEqual(getattr(txn, registry.key), [func])
registry.unregister(func, txn)
self.assertFalse(hasattr(txn, registry.key))
def test_unregister_notexists(self):
registry = self._makeOne()
func = lambda *x: None
txn = Dummy()
setattr(txn, registry.key, [None])
registry.unregister(func, txn)
self.assertEqual(getattr(txn, registry.key), [None])
def test_unregister_funcs_is_None(self):
registry = self._makeOne()
func = lambda *x: None
txn = Dummy()
self.assertEqual(registry.unregister(func, txn), None)
class UtilityFunctionTests(unittest.TestCase):
def test_isActive(self):
from repoze.tm import ekey
from repoze.tm import isActive
self.assertEqual(isActive({ekey:True}), True)
self.assertEqual(isActive({}), False)
class TestMakeTM(unittest.TestCase):
def test_make_tm_withveto(self):
from repoze.tm import make_tm
tm = make_tm(DummyApplication(), {}, 'repoze.tm.tests:fakeveto')
self.assertEqual(tm.commit_veto, fakeveto)
def test_make_tm_noveto(self):
from repoze.tm import make_tm
tm = make_tm(DummyApplication(), {}, None)
self.assertEqual(tm.commit_veto, None)
class Test_default_commit_veto(unittest.TestCase):
def _callFUT(self, status, headers=()):
from repoze.tm import default_commit_veto
return default_commit_veto(None, status, headers)
def test_it_true_5XX(self):
self.assertTrue(self._callFUT('500 Server Error'))
self.assertTrue(self._callFUT('503 Service Unavailable'))
def test_it_true_4XX(self):
self.assertTrue(self._callFUT('400 Bad Request'))
self.assertTrue(self._callFUT('411 Length Required'))
def test_it_false_2XX(self):
self.assertFalse(self._callFUT('200 OK'))
self.assertFalse(self._callFUT('201 Created'))
def test_it_false_3XX(self):
self.assertFalse(self._callFUT('301 Moved Permanently'))
self.assertFalse(self._callFUT('302 Found'))
def test_it_true_x_tm_abort_specific(self):
self.assertTrue(self._callFUT('200 OK', [('X-Tm-Abort', True)]))
def test_it_false_x_tm_commit(self):
self.assertFalse(self._callFUT('200 OK', [('X-Tm', 'commit')]))
def test_it_true_x_tm_abort(self):
self.assertTrue(self._callFUT('200 OK', [('X-Tm', 'abort')]))
def test_it_true_x_tm_anythingelse(self):
self.assertTrue(self._callFUT('200 OK', [('X-Tm', '')]))
def test_x_tm_generic_precedes_x_tm_abort_specific(self):
self.assertFalse(self._callFUT('200 OK', [('X-Tm', 'commit'),
('X-Tm-Abort', True)]))
def fakeveto(environ, status, headers):
""" """
class DummyTransactionModule:
begun = False
committed = False
aborted = False
def __init__(self, doom=False):
self.doom = doom
def begin(self):
self.begun = True
def get(self):
return self
def commit(self):
self.committed = True
def abort(self):
self.aborted = True
def isDoomed(self):
return self.doom
class Dummy:
pass
class DummyApplication:
def __init__(self, exception=False, status="200 OK"):
self.exception = exception
self.status = status
def __call__(self, environ, start_response):
start_response(self.status, [], None)
if self.exception:
raise ValueError('raising')
return ['hello']
|