python-repoze.tm2 2.0-1 / usr / lib / python2.7 / dist-packages / repoze / tm / tests.py

This file is indexed.

/usr/lib/python2.7/dist-packages/repoze/tm/tests.py is in python-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']

APT Browse - Built by Thomas Orozco.