twisted/web/test/test_tap.py

   1 # Copyright (c) Twisted Matrix Laboratories.
   2 # See LICENSE for details.
   3
   4 """
   5 Tests for L{twisted.web.tap}.
   6 """
   7
   8 import os, stat
   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
  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_epyProcessor(self):
  74         """
  75         The I{--path} option creates a root resource which serves a
  76         L{PythonScript} instance for any child with the C{".epy"} extension.
  77         """
  78         path, root = self._pathOption()
  79         path.child("foo.epy").setContent("")
  80         self.assertIsInstance(root.getChild("foo.epy", None), PythonScript)
  81
  82
  83     def test_rpyProcessor(self):
  84         """
  85         The I{--path} option creates a root resource which serves the
  86         C{resource} global defined by the Python source in any child with
  87         the C{".rpy"} extension.
  88         """
  89         path, root = self._pathOption()
  90         path.child("foo.rpy").setContent(
  91             "from twisted.web.static import Data\n"
  92             "resource = Data('content', 'major/minor')\n")
  93         child = root.getChild("foo.rpy", None)
  94         self.assertIsInstance(child, Data)
  95         self.assertEqual(child.data, 'content')
  96         self.assertEqual(child.type, 'major/minor')
  97
  98
  99     def test_makePersonalServerFactory(self):
 100         """
 101         L{makePersonalServerFactory} returns a PB server factory which has
 102         as its root object a L{ResourcePublisher}.
 103         """
 104         # The fact that this pile of objects can actually be used somehow is
 105         # verified by twisted.web.test.test_distrib.
 106         site = Site(Data("foo bar", "text/plain"))
 107         serverFactory = makePersonalServerFactory(site)
 108         self.assertIsInstance(serverFactory, PBServerFactory)
 109         self.assertIsInstance(serverFactory.root, ResourcePublisher)
 110         self.assertIdentical(serverFactory.root.site, site)
 111
 112
 113     def test_personalServer(self):
 114         """
 115         The I{--personal} option to L{makeService} causes it to return a
 116         service which will listen on the server address given by the I{--port}
 117         option.
 118         """
 119         port = self.mktemp()
 120         options = Options()
 121         options.parseOptions(['--port', 'unix:' + port, '--personal'])
 122         service = makeService(options)
 123         service.startService()
 124         self.addCleanup(service.stopService)
 125         self.assertTrue(os.path.exists(port))
 126         self.assertTrue(stat.S_ISSOCK(os.stat(port).st_mode))
 127
 128     if not IReactorUNIX.providedBy(reactor):
 129         test_personalServer.skip = (
 130             "The reactor does not support UNIX domain sockets")
 131
 132
 133     def test_defaultPersonalPath(self):
 134         """
 135         If the I{--port} option not specified but the I{--personal} option is,
 136         L{Options} defaults the port to C{UserDirectory.userSocketName} in the
 137         user's home directory.
 138         """
 139         options = Options()
 140         options.parseOptions(['--personal'])
 141         path = os.path.expanduser(
 142             os.path.join('~', UserDirectory.userSocketName))
 143         self.assertEqual(
 144             strports.parse(options['port'], None)[:2],
 145             ('UNIX', (path, None)))
 146
 147     if not IReactorUNIX.providedBy(reactor):
 148         test_defaultPersonalPath.skip = (
 149             "The reactor does not support UNIX domain sockets")
 150
 151
 152     def test_defaultPort(self):
 153         """
 154         If the I{--port} option is not specified, L{Options} defaults the port
 155         to C{8080}.
 156         """
 157         options = Options()
 158         options.parseOptions([])
 159         self.assertEqual(
 160             strports.parse(options['port'], None)[:2],
 161             ('TCP', (8080, None)))
 162
 163
 164     def test_wsgi(self):
 165         """
 166         The I{--wsgi} option takes the fully-qualifed Python name of a WSGI
 167         application object and creates a L{WSGIResource} at the root which
 168         serves that application.
 169         """
 170         options = Options()
 171         options.parseOptions(['--wsgi', __name__ + '.application'])
 172         root = options['root']
 173         self.assertTrue(root, WSGIResource)
 174         self.assertIdentical(root._reactor, reactor)
 175         self.assertTrue(isinstance(root._threadpool, ThreadPool))
 176         self.assertIdentical(root._application, application)
 177
 178         # The threadpool should start and stop with the reactor.
 179         self.assertFalse(root._threadpool.started)
 180         reactor.fireSystemEvent('startup')
 181         self.assertTrue(root._threadpool.started)
 182         self.assertFalse(root._threadpool.joined)
 183         reactor.fireSystemEvent('shutdown')
 184         self.assertTrue(root._threadpool.joined)
 185
 186
 187     def test_invalidApplication(self):
 188         """
 189         If I{--wsgi} is given an invalid name, L{Options.parseOptions}
 190         raises L{UsageError}.
 191         """
 192         options = Options()
 193         for name in [__name__ + '.nosuchthing', 'foo.']:
 194             exc = self.assertRaises(
 195                 UsageError, options.parseOptions, ['--wsgi', name])
 196             self.assertEqual(str(exc), "No such WSGI application: %r" % (name,))