← Back to branch summary

~ntt-pf-lab/nova/monkey_patch_notification

« back to all changes in this revision

Viewing changes to vendor/Twisted-10.0.0/twisted/test/generator_failure_tests.py

  • Committer: Jesse Andrews
  • Date: 2010-05-28 06:05:26 UTC
  • Revision ID: git-v1:bf6e6e718cdc7488e2da87b21e258ccc065fe499
initial commit

Show diffs side-by-side

added added

removed removed

Lines of Context:
vendor/Twisted-10.0.0/twisted/test/generator_failure_tests.py
 
1
# Copyright (c) 2001-2007 Twisted Matrix Laboratories.
 
2
# See LICENSE for details.
 
3
 
 
4
 
 
5
"""
 
6
Python 2.5 test cases for failures thrown into generators.
 
7
"""
 
8
 
 
9
import sys
 
10
import traceback
 
11
 
 
12
from twisted.trial.unittest import TestCase
 
13
 
 
14
from twisted.python.failure import Failure
 
15
from twisted.test.test_failure import getDivisionFailure
 
16
from twisted.internet import defer
 
17
 
 
18
 
 
19
class TwoPointFiveFailureTests(TestCase):
 
20
 
 
21
    def test_inlineCallbacksTracebacks(self):
 
22
        """
 
23
        inlineCallbacks that re-raise tracebacks into their deferred
 
24
        should not lose their tracebacsk.
 
25
        """
 
26
        f = getDivisionFailure()
 
27
        d = defer.Deferred()
 
28
        try:
 
29
            f.raiseException()
 
30
        except:
 
31
            d.errback()
 
32
 
 
33
        failures = []
 
34
        def collect_error(result):
 
35
            failures.append(result)
 
36
 
 
37
        def ic(d):
 
38
            yield d
 
39
        ic = defer.inlineCallbacks(ic)
 
40
        ic(d).addErrback(collect_error)
 
41
 
 
42
        newFailure, = failures
 
43
        self.assertEquals(
 
44
            traceback.extract_tb(newFailure.getTracebackObject())[-1][-1],
 
45
            "1/0"
 
46
        )
 
47
 
 
48
 
 
49
    def _throwIntoGenerator(self, f, g):
 
50
        try:
 
51
            f.throwExceptionIntoGenerator(g)
 
52
        except StopIteration:
 
53
            pass
 
54
        else:
 
55
            self.fail("throwExceptionIntoGenerator should have raised "
 
56
                      "StopIteration")
 
57
 
 
58
    def test_throwExceptionIntoGenerator(self):
 
59
        """
 
60
        It should be possible to throw the exception that a Failure
 
61
        represents into a generator.
 
62
        """
 
63
        stuff = []
 
64
        def generator():
 
65
            try:
 
66
                yield
 
67
            except:
 
68
                stuff.append(sys.exc_info())
 
69
            else:
 
70
                self.fail("Yield should have yielded exception.")
 
71
        g = generator()
 
72
        f = getDivisionFailure()
 
73
        g.next()
 
74
        self._throwIntoGenerator(f, g)
 
75
 
 
76
        self.assertEquals(stuff[0][0], ZeroDivisionError)
 
77
        self.assertTrue(isinstance(stuff[0][1], ZeroDivisionError))
 
78
 
 
79
        self.assertEquals(traceback.extract_tb(stuff[0][2])[-1][-1], "1/0")
 
80
 
 
81
 
 
82
    def test_findFailureInGenerator(self):
 
83
        """
 
84
        Within an exception handler, it should be possible to find the
 
85
        original Failure that caused the current exception (if it was
 
86
        caused by throwExceptionIntoGenerator).
 
87
        """
 
88
        f = getDivisionFailure()
 
89
        f.cleanFailure()
 
90
 
 
91
        foundFailures = []
 
92
        def generator():
 
93
            try:
 
94
                yield
 
95
            except:
 
96
                foundFailures.append(Failure._findFailure())
 
97
            else:
 
98
                self.fail("No exception sent to generator")
 
99
 
 
100
        g = generator()
 
101
        g.next()
 
102
        self._throwIntoGenerator(f, g)
 
103
 
 
104
        self.assertEqual(foundFailures, [f])
 
105
 
 
106
 
 
107
    def test_failureConstructionFindsOriginalFailure(self):
 
108
        """
 
109
        When a Failure is constructed in the context of an exception
 
110
        handler that is handling an exception raised by
 
111
        throwExceptionIntoGenerator, the new Failure should be chained to that
 
112
        original Failure.
 
113
        """
 
114
        f = getDivisionFailure()
 
115
        f.cleanFailure()
 
116
 
 
117
        newFailures = []
 
118
 
 
119
        def generator():
 
120
            try:
 
121
                yield
 
122
            except:
 
123
                newFailures.append(Failure())
 
124
            else:
 
125
                self.fail("No exception sent to generator")
 
126
        g = generator()
 
127
        g.next()
 
128
        self._throwIntoGenerator(f, g)
 
129
 
 
130
        self.assertEqual(len(newFailures), 1)
 
131
        self.assertEqual(newFailures[0].getTraceback(), f.getTraceback())
 
132
 
 
133
    def test_ambiguousFailureInGenerator(self):
 
134
        """
 
135
        When a generator reraises a different exception,
 
136
        L{Failure._findFailure} inside the generator should find the reraised
 
137
        exception rather than original one.
 
138
        """
 
139
        def generator():
 
140
            try:
 
141
                try:
 
142
                    yield
 
143
                except:
 
144
                    [][1]
 
145
            except:
 
146
                self.assertIsInstance(Failure().value, IndexError)
 
147
        g = generator()
 
148
        g.next()
 
149
        f = getDivisionFailure()
 
150
        self._throwIntoGenerator(f, g)
 
151
 
 
152
    def test_ambiguousFailureFromGenerator(self):
 
153
        """
 
154
        When a generator reraises a different exception,
 
155
        L{Failure._findFailure} above the generator should find the reraised
 
156
        exception rather than original one.
 
157
        """
 
158
        def generator():
 
159
            try:
 
160
                yield
 
161
            except:
 
162
                [][1]
 
163
        g = generator()
 
164
        g.next()
 
165
        f = getDivisionFailure()
 
166
        try:
 
167
            self._throwIntoGenerator(f, g)
 
168
        except:
 
169
            self.assertIsInstance(Failure().value, IndexError)