← Back to branch summary

~certify-web-dev/twisted/certify-trunk

« back to all changes in this revision

Viewing changes to twisted/trial/test/test_util.py

  • Committer: Bazaar Package Importer
  • Author(s): Matthias Klose
  • Date: 2006-02-22 22:52:47 UTC
  • Revision ID: james.westby@ubuntu.com-20060222225247-0mjb8ij9473m5zse
Tags: 2.2.0-1ubuntu1
Synchronize with Debian unstable.

Show diffs side-by-side

added added

removed removed

Lines of Context:
twisted/trial/test/test_util.py
1
 
# -*- test-case-name: twisted.trial.test.test_util -*-
2
 
from twisted.python import failure, log
3
 
from twisted.python.runtime import platformType
 
1
from twisted.python import log
 
2
from twisted.internet import utils
4
3
from twisted.internet import defer, reactor, threads, interfaces
5
 
from twisted.trial import unittest, util, runner
6
 
 
7
 
import os, time, signal
8
 
 
9
 
class UserError(Exception):
10
 
    pass
11
 
 
12
 
class TestUserMethod(unittest.TestCase):
13
 
    def setUp(self):
14
 
        self.janitor = util._Janitor()
15
 
 
16
 
    def errorfulMethod(self):
17
 
        raise UserError, 'i am a user error'
18
 
 
19
 
    def errorfulDeferred(self):
20
 
        f = None
21
 
        try:
22
 
            self.errorfulMethod()
23
 
        except:
24
 
            f = failure.Failure()
25
 
        return defer.fail(f)
26
 
 
27
 
    def testErrorHandling(self):
28
 
        """wrapper around user code"""
29
 
        umw = util.UserMethodWrapper(self.errorfulMethod, self.janitor)
30
 
        self.failUnlessRaises(util.UserMethodError, umw)
31
 
        self.failUnless(umw.errors[0].check(UserError))
32
 
        self.failUnless(umw.endTime >= umw.startTime)
33
 
 
34
 
    def testDeferredError(self):
35
 
        umw = util.UserMethodWrapper(self.errorfulDeferred, self.janitor)
36
 
        self.failUnlessRaises(util.UserMethodError, umw)
37
 
        self.failUnless(umw.errors[0].check(UserError))
38
 
        self.failUnless(umw.endTime >= umw.startTime)
39
 
 
 
4
from twisted.trial import unittest, util
 
5
from twisted.trial.test import packages
 
6
 
 
7
import sys, os, time
 
8
 
 
9
suppress = [(['ignore', 'Do NOT use wait.*'], {})]
40
10
 
41
11
class WaitReentrancyTest(unittest.TestCase):
42
12
 
73
43
 
74
44
    def testReentrantWait(self):
75
45
        self.assertRaises(util.WaitIsNotReentrantError, self._reentrantWait)
76
 
 
77
 
 
78
 
class TestWait2(unittest.TestCase):
79
 
    NUM_FAILURES = 3
80
 
 
81
 
    def _generateFailure(self):
82
 
        try:
83
 
            raise RuntimeError, "i am a complete and utter failure"
84
 
        except RuntimeError:
85
 
            return failure.Failure()
86
 
 
87
 
    def _errorfulMethod(self):
88
 
        L = [self._generateFailure() for x in xrange(self.NUM_FAILURES)]
89
 
        raise util.MultiError(L)
90
 
 
91
 
    def testMultiError(self):
92
 
        self.assertRaises(util.MultiError, self._errorfulMethod)
93
 
        try:
94
 
            self._errorfulMethod()
95
 
        except util.MultiError, e:
96
 
            self.assert_(hasattr(e, 'failures'))
97
 
            self.assertEqual(len(e.failures), self.NUM_FAILURES)
98
 
            for f in e.failures:
99
 
                self.assert_(f.check(RuntimeError))
100
 
 
101
 
    def testMultiErrorAsFailure(self):
102
 
        self.assertRaises(util.MultiError, self._errorfulMethod)
103
 
        try:
104
 
            self._errorfulMethod()
105
 
        except util.MultiError:
106
 
            f = failure.Failure()
107
 
            self.assert_(hasattr(f, 'value'))
108
 
            self.assert_(hasattr(f.value, 'failures'))
109
 
            self.assertEqual(len(f.value.failures), self.NUM_FAILURES)
110
 
            for f in f.value.failures:
111
 
                self.assert_(f.check(RuntimeError))
 
46
        
 
47
    def test_twoWaitImplementations(self):
 
48
        # If this test times out, then wait is being re-entered.
 
49
        tc = TestMktemp('test_name')
 
50
        tc._timedOut = False # whitebox
 
51
        d = defer.Deferred()
 
52
        def _runsInsideWait(r):
 
53
            d = defer.Deferred()
 
54
            self.assertRaises(util.WaitIsNotReentrantError, util.wait, d)
 
55
        d.addCallback(utils.suppressWarnings(_runsInsideWait, *suppress))
 
56
        reactor.callLater(0, d.callback, 'yo')
 
57
        tc._wait(d)
 
58
    test_twoWaitImplementations.timeout = 4
112
59
 
113
60
 
114
61
class TestMktemp(unittest.TestCase):
115
 
    def testMktmp(self):
116
 
        tmp = self.mktemp()
117
 
        tmp1 = self.mktemp()
118
 
        exp = os.path.join('twisted.trial.test.test_trial', 'UtilityTestCase', 'testMktmp')
119
 
        self.failIfEqual(tmp, tmp1)
120
 
        self.failIf(os.path.exists(exp))
 
62
    def test_name(self):
 
63
        name = self.mktemp()
 
64
        dirs = os.path.dirname(name).split(os.sep)[:-1]
 
65
        self.failUnlessEqual(
 
66
            dirs, ['twisted.trial.test.test_util', 'TestMktemp', 'test_name'])
 
67
 
 
68
    def test_unique(self):
 
69
        name = self.mktemp()
 
70
        self.failIfEqual(name, self.mktemp())
 
71
 
 
72
    def test_created(self):
 
73
        name = self.mktemp()
 
74
        dirname = os.path.dirname(name)
 
75
        self.failUnless(os.path.exists(dirname))
 
76
        self.failIf(os.path.exists(name))
 
77
 
 
78
    def test_location(self):
 
79
        path = os.path.abspath(self.mktemp())
 
80
        self.failUnless(path.startswith(os.getcwd()))
121
81
 
122
82
 
123
83
class TestWaitInterrupt(unittest.TestCase):
199
159
        for a, b in zip(parents, expected):
200
160
            self.failUnlessEqual(a, b)
201
161
                     
 
162
 
 
163
class TestFindObject(packages.PackageTest):
 
164
    def setUp(self):
 
165
        packages.PackageTest.setUp(self, '_TestFindObject')
 
166
        self.oldPath = sys.path[:]
 
167
        sys.path.append('_TestFindObject')
 
168
 
 
169
    def tearDown(self):
 
170
        sys.path = self.oldPath
 
171
        packages.PackageTest.tearDown(self, '_TestFindObject')
 
172
 
 
173
    def test_importPackage(self):
 
174
        package1 = util.findObject('package')
 
175
        import package as package2
 
176
        self.failUnlessEqual(package1, (True, package2))
 
177
 
 
178
    def test_importModule(self):
 
179
        test_sample2 = util.findObject('goodpackage.test_sample')
 
180
        from goodpackage import test_sample
 
181
        self.failUnlessEqual((True, test_sample), test_sample2)
 
182
 
 
183
    def test_importError(self):
 
184
        self.failUnlessRaises(ZeroDivisionError,
 
185
                              util.findObject, 'package.test_bad_module')
 
186
 
 
187
    def test_sophisticatedImportError(self):
 
188
        self.failUnlessRaises(ImportError,
 
189
                              util.findObject, 'package2.test_module')
 
190
 
 
191
    def test_importNonexistentPackage(self):
 
192
        self.failUnlessEqual(util.findObject('doesntexist')[0], False)
 
193
 
 
194
    def test_findNonexistentModule(self):
 
195
        self.failUnlessEqual(util.findObject('package.doesntexist')[0], False)
 
196
 
 
197
    def test_findNonexistentObject(self):
 
198
        self.failUnlessEqual(util.findObject(
 
199
            'goodpackage.test_sample.doesnt')[0], False)
 
200
        self.failUnlessEqual(util.findObject(
 
201
            'goodpackage.test_sample.AlphabetTest.doesntexist')[0], False)
 
202
 
 
203
    def test_findObjectExist(self):
 
204
        alpha1 = util.findObject('goodpackage.test_sample.AlphabetTest')
 
205
        from goodpackage import test_sample
 
206
        self.failUnlessEqual(alpha1, (True, test_sample.AlphabetTest))
 
207