/usr/share/pyshared/twisted/web2/test/test_scgi.py is in python-twisted-web2 8.1.0-3.
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 | from twisted.trial import unittest
from twisted.internet import defer
from twisted.internet.address import IPv4Address
from twisted.web2.test.test_http import LoopbackRelay, TestConnection
from twisted.web2.test.test_http import TestClient, HTTPTests, TestRequest
from twisted.web2.test.test_client import TestServer, ClientTests
from twisted.web2.test.test_server import SimpleRequest
from twisted.web2 import server
from twisted.web2 import http_headers
from twisted.web2 import stream
from twisted.web2 import twscgi
def parseSCGIHeaders(headers):
return zip(*[iter(headers.split(':', 1)[1].split('\x00'))]*2)
class SCGITests(HTTPTests):
def connect(self, logFile=None):
cxn = TestConnection()
cxn.client = self.clientProtocol
cxn.server = self.serverProtocol
cxn.serverToClient = LoopbackRelay(cxn.client, logFile)
cxn.clientToServer = LoopbackRelay(cxn.server, logFile)
cxn.server.makeConnection(cxn.serverToClient)
cxn.client.makeConnection(cxn.clientToServer)
return cxn
class SCGIClientTests(SCGITests, ClientTests):
def setUp(self):
self.serverProtocol = TestServer()
def doTestSCGI(self, request):
if request.stream.length is None:
return http.Response(responsecode.LENGTH_REQUIRED)
factory = twscgi.SCGIClientProtocolFactory(request)
self.clientProtocol = factory.buildProtocol(None)
self.cxn = self.connect()
return factory.deferred
def testSimpleRequest(self):
def gotResponse(resp):
self.assertEquals(resp.code, 200)
self.assertEquals(resp.headers.getHeader('Content-Type'),
http_headers.MimeType.fromString('text/plain'))
return defer.maybeDeferred(resp.stream.read
).addCallback(self.assertEquals, '42')
req = SimpleRequest(None, 'GET', '/')
d = self.doTestSCGI(req)
d.addCallback(gotResponse)
self.iterate(self.cxn)
headers = parseSCGIHeaders(self.cxn.server.data)
self.assertEquals(headers[0], ('CONTENT_LENGTH', '0'))
self.failUnlessIn(('SCGI', '1'), headers)
self.writeLines(self.cxn, ['Status: 200 OK',
'Content-Type: text/plain',
'Content-Length: 2',
'',
'42'])
return d
def testOperatesOnStreamDirectly(self):
def gotResponse(resp):
self.assertEquals(resp.code, 200)
self.assertEquals(resp.headers.getHeader('Content-Type'),
http_headers.MimeType.fromString('text/plain'))
stream = resp.stream
resp.stream = None
return defer.maybeDeferred(stream.read
).addCallback(self.assertEquals, '42')
req = SimpleRequest(None, 'GET', '/')
d = self.doTestSCGI(req)
d.addCallback(gotResponse)
self.iterate(self.cxn)
headers = parseSCGIHeaders(self.cxn.server.data)
self.assertEquals(headers[0], ('CONTENT_LENGTH', '0'))
self.failUnlessIn(('SCGI', '1'), headers)
self.writeLines(self.cxn, ['Status: 200 OK',
'Content-Type: text/plain',
'Content-Length: 2',
'',
'42'])
return d
|