← Back to branch summary

~justin-fathomdb/nova/justinsb-openstack-api-volumes

« back to all changes in this revision

Viewing changes to vendor/Twisted-10.0.0/twisted/web/test/test_tap.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/web/test/test_tap.py
 
1
# Copyright (c) 2008 Twisted Matrix Laboratories.
 
2
# See LICENSE for details.
 
3
 
 
4
"""
 
5
Tests for L{twisted.web.tap}.
 
6
"""
 
7
 
 
8
import os, stat, pickle
 
9
 
 
10
from twisted.python.usage import UsageError
 
11
from twisted.python.filepath import FilePath
 
12
from twisted.internet.interfaces import IReactorUNIX
 
13
from twisted.internet import reactor
 
14
from twisted.python.threadpool import ThreadPool
 
15
from twisted.trial.unittest import TestCase
 
16
from twisted.application import strports
 
17
 
 
18
from twisted.web.server import Site
 
19
from twisted.web.static import Data, File
 
20
from twisted.web.distrib import ResourcePublisher, UserDirectory
 
21
from twisted.web.wsgi import WSGIResource
 
22
from twisted.web.tap import Options, makePersonalServerFactory, makeService
 
23
from twisted.web.twcgi import CGIScript, PHP3Script, PHPScript
 
24
from twisted.web.script import PythonScript
 
25
 
 
26
 
 
27
from twisted.spread.pb import PBServerFactory
 
28
 
 
29
application = object()
 
30
 
 
31
class ServiceTests(TestCase):
 
32
    """
 
33
    Tests for the service creation APIs in L{twisted.web.tap}.
 
34
    """
 
35
    def _pathOption(self):
 
36
        """
 
37
        Helper for the I{--path} tests which creates a directory and creates
 
38
        an L{Options} object which uses that directory as its static
 
39
        filesystem root.
 
40
 
 
41
        @return: A two-tuple of a L{FilePath} referring to the directory and
 
42
            the value associated with the C{'root'} key in the L{Options}
 
43
            instance after parsing a I{--path} option.
 
44
        """
 
45
        path = FilePath(self.mktemp())
 
46
        path.makedirs()
 
47
        options = Options()
 
48
        options.parseOptions(['--path', path.path])
 
49
        root = options['root']
 
50
        return path, root
 
51
 
 
52
 
 
53
    def test_path(self):
 
54
        """
 
55
        The I{--path} option causes L{Options} to create a root resource
 
56
        which serves responses from the specified path.
 
57
        """
 
58
        path, root = self._pathOption()
 
59
        self.assertIsInstance(root, File)
 
60
        self.assertEqual(root.path, path.path)
 
61
 
 
62
 
 
63
    def test_cgiProcessor(self):
 
64
        """
 
65
        The I{--path} option creates a root resource which serves a
 
66
        L{CGIScript} instance for any child with the C{".cgi"} extension.
 
67
        """
 
68
        path, root = self._pathOption()
 
69
        path.child("foo.cgi").setContent("")
 
70
        self.assertIsInstance(root.getChild("foo.cgi", None), CGIScript)
 
71
 
 
72
 
 
73
    def test_php3Processor(self):
 
74
        """
 
75
        The I{--path} option creates a root resource which serves a
 
76
        L{PHP3Script} instance for any child with the C{".php3"} extension.
 
77
        """
 
78
        path, root = self._pathOption()
 
79
        path.child("foo.php3").setContent("")
 
80
        self.assertIsInstance(root.getChild("foo.php3", None), PHP3Script)
 
81
 
 
82
 
 
83
    def test_phpProcessor(self):
 
84
        """
 
85
        The I{--path} option creates a root resource which serves a
 
86
        L{PHPScript} instance for any child with the C{".php"} extension.
 
87
        """
 
88
        path, root = self._pathOption()
 
89
        path.child("foo.php").setContent("")
 
90
        self.assertIsInstance(root.getChild("foo.php", None), PHPScript)
 
91
 
 
92
 
 
93
    def test_epyProcessor(self):
 
94
        """
 
95
        The I{--path} option creates a root resource which serves a
 
96
        L{PythonScript} instance for any child with the C{".epy"} extension.
 
97
        """
 
98
        path, root = self._pathOption()
 
99
        path.child("foo.epy").setContent("")
 
100
        self.assertIsInstance(root.getChild("foo.epy", None), PythonScript)
 
101
 
 
102
 
 
103
    def test_rpyProcessor(self):
 
104
        """
 
105
        The I{--path} option creates a root resource which serves the
 
106
        C{resource} global defined by the Python source in any child with
 
107
        the C{".rpy"} extension.
 
108
        """
 
109
        path, root = self._pathOption()
 
110
        path.child("foo.rpy").setContent(
 
111
            "from twisted.web.static import Data\n"
 
112
            "resource = Data('content', 'major/minor')\n")
 
113
        child = root.getChild("foo.rpy", None)
 
114
        self.assertIsInstance(child, Data)
 
115
        self.assertEqual(child.data, 'content')
 
116
        self.assertEqual(child.type, 'major/minor')
 
117
 
 
118
 
 
119
    def test_trpProcessor(self):
 
120
        """
 
121
        The I{--path} option creates a root resource which serves the
 
122
        pickled resource out of any child with the C{".rpy"} extension.
 
123
        """
 
124
        path, root = self._pathOption()
 
125
        path.child("foo.trp").setContent(pickle.dumps(Data("foo", "bar")))
 
126
        child = root.getChild("foo.trp", None)
 
127
        self.assertIsInstance(child, Data)
 
128
        self.assertEqual(child.data, 'foo')
 
129
        self.assertEqual(child.type, 'bar')
 
130
 
 
131
        warnings = self.flushWarnings()
 
132
 
 
133
        # If the trp module hadn't been imported before this test ran, there
 
134
        # will be two deprecation warnings; one for the module, one for the
 
135
        # function.  If the module has already been imported, the
 
136
        # module-scope deprecation won't be emitted again.
 
137
        if len(warnings) == 2:
 
138
            self.assertEqual(warnings[0]['category'], DeprecationWarning)
 
139
            self.assertEqual(
 
140
                warnings[0]['message'],
 
141
                "twisted.web.trp is deprecated as of Twisted 9.0.  Resource "
 
142
                "persistence is beyond the scope of Twisted Web.")
 
143
            warning = warnings[1]
 
144
        else:
 
145
            warning = warnings[0]
 
146
 
 
147
        self.assertEqual(warning['category'], DeprecationWarning)
 
148
        self.assertEqual(
 
149
            warning['message'],
 
150
            "twisted.web.trp.ResourceUnpickler is deprecated as of Twisted "
 
151
            "9.0.  Resource persistence is beyond the scope of Twisted Web.")
 
152
 
 
153
 
 
154
    def test_makePersonalServerFactory(self):
 
155
        """
 
156
        L{makePersonalServerFactory} returns a PB server factory which has
 
157
        as its root object a L{ResourcePublisher}.
 
158
        """
 
159
        # The fact that this pile of objects can actually be used somehow is
 
160
        # verified by twisted.web.test.test_distrib.
 
161
        site = Site(Data("foo bar", "text/plain"))
 
162
        serverFactory = makePersonalServerFactory(site)
 
163
        self.assertIsInstance(serverFactory, PBServerFactory)
 
164
        self.assertIsInstance(serverFactory.root, ResourcePublisher)
 
165
        self.assertIdentical(serverFactory.root.site, site)
 
166
 
 
167
 
 
168
    def test_personalServer(self):
 
169
        """
 
170
        The I{--personal} option to L{makeService} causes it to return a
 
171
        service which will listen on the server address given by the I{--port}
 
172
        option.
 
173
        """
 
174
        port = self.mktemp()
 
175
        options = Options()
 
176
        options.parseOptions(['--port', 'unix:' + port, '--personal'])
 
177
        service = makeService(options)
 
178
        service.startService()
 
179
        self.addCleanup(service.stopService)
 
180
        self.assertTrue(os.path.exists(port))
 
181
        self.assertTrue(stat.S_ISSOCK(os.stat(port).st_mode))
 
182
 
 
183
    if not IReactorUNIX.providedBy(reactor):
 
184
        test_personalServer.skip = (
 
185
            "The reactor does not support UNIX domain sockets")
 
186
 
 
187
 
 
188
    def test_defaultPersonalPath(self):
 
189
        """
 
190
        If the I{--port} option not specified but the I{--personal} option is,
 
191
        L{Options} defaults the port to C{UserDirectory.userSocketName} in the
 
192
        user's home directory.
 
193
        """
 
194
        options = Options()
 
195
        options.parseOptions(['--personal'])
 
196
        path = os.path.expanduser(
 
197
            os.path.join('~', UserDirectory.userSocketName))
 
198
        self.assertEqual(
 
199
            strports.parse(options['port'], None)[:2],
 
200
            ('UNIX', (path, None)))
 
201
 
 
202
    if not IReactorUNIX.providedBy(reactor):
 
203
        test_defaultPersonalPath.skip = (
 
204
            "The reactor does not support UNIX domain sockets")
 
205
 
 
206
 
 
207
    def test_defaultPort(self):
 
208
        """
 
209
        If the I{--port} option is not specified, L{Options} defaults the port
 
210
        to C{8080}.
 
211
        """
 
212
        options = Options()
 
213
        options.parseOptions([])
 
214
        self.assertEqual(
 
215
            strports.parse(options['port'], None)[:2],
 
216
            ('TCP', (8080, None)))
 
217
 
 
218
 
 
219
    def test_wsgi(self):
 
220
        """
 
221
        The I{--wsgi} option takes the fully-qualifed Python name of a WSGI
 
222
        application object and creates a L{WSGIResource} at the root which
 
223
        serves that application.
 
224
        """
 
225
        options = Options()
 
226
        options.parseOptions(['--wsgi', __name__ + '.application'])
 
227
        root = options['root']
 
228
        self.assertTrue(root, WSGIResource)
 
229
        self.assertIdentical(root._reactor, reactor)
 
230
        self.assertTrue(isinstance(root._threadpool, ThreadPool))
 
231
        self.assertIdentical(root._application, application)
 
232
 
 
233
        # The threadpool should start and stop with the reactor.
 
234
        self.assertFalse(root._threadpool.started)
 
235
        reactor.fireSystemEvent('startup')
 
236
        self.assertTrue(root._threadpool.started)
 
237
        self.assertFalse(root._threadpool.joined)
 
238
        reactor.fireSystemEvent('shutdown')
 
239
        self.assertTrue(root._threadpool.joined)
 
240
 
 
241
 
 
242
    def test_invalidApplication(self):
 
243
        """
 
244
        If I{--wsgi} is given an invalid name, L{Options.parseOptions}
 
245
        raises L{UsageError}.
 
246
        """
 
247
        options = Options()
 
248
        for name in [__name__ + '.nosuchthing', 'foo.']:
 
249
            exc = self.assertRaises(
 
250
                UsageError, options.parseOptions, ['--wsgi', name])
 
251
            self.assertEqual(str(exc), "No such WSGI application: %r" % (name,))