1
from twisted.trial import unittest
3
from twisted.internet import defer
4
from twisted.internet.address import IPv4Address
6
from twisted.web2.test.test_http import LoopbackRelay, TestConnection
7
from twisted.web2.test.test_http import TestClient, HTTPTests, TestRequest
9
from twisted.web2.test.test_client import TestServer, ClientTests
10
from twisted.web2.test.test_server import SimpleRequest
12
from twisted.web2 import server
13
from twisted.web2 import http_headers
14
from twisted.web2 import stream
15
from twisted.web2 import twscgi
17
def parseSCGIHeaders(headers):
18
return zip(*[iter(headers.split(':', 1)[1].split('\x00'))]*2)
20
class SCGITests(HTTPTests):
21
def connect(self, logFile=None):
22
cxn = TestConnection()
23
cxn.client = self.clientProtocol
24
cxn.server = self.serverProtocol
26
cxn.serverToClient = LoopbackRelay(cxn.client, logFile)
27
cxn.clientToServer = LoopbackRelay(cxn.server, logFile)
28
cxn.server.makeConnection(cxn.serverToClient)
29
cxn.client.makeConnection(cxn.clientToServer)
34
class SCGIClientTests(SCGITests, ClientTests):
36
self.serverProtocol = TestServer()
38
def doTestSCGI(self, request):
39
if request.stream.length is None:
40
return http.Response(responsecode.LENGTH_REQUIRED)
41
factory = twscgi.SCGIClientProtocolFactory(request)
42
self.clientProtocol = factory.buildProtocol(None)
43
self.cxn = self.connect()
44
return factory.deferred
46
def testSimpleRequest(self):
47
def gotResponse(resp):
48
self.assertEquals(resp.code, 200)
49
self.assertEquals(resp.headers.getHeader('Content-Type'),
50
http_headers.MimeType.fromString('text/plain'))
52
return defer.maybeDeferred(resp.stream.read
53
).addCallback(self.assertEquals, '42')
55
req = SimpleRequest(None, 'GET', '/')
57
d = self.doTestSCGI(req)
58
d.addCallback(gotResponse)
60
self.iterate(self.cxn)
62
headers = parseSCGIHeaders(self.cxn.server.data)
64
self.assertEquals(headers[0], ('CONTENT_LENGTH', '0'))
66
self.failUnlessIn(('SCGI', '1'), headers)
68
self.writeLines(self.cxn, ['Status: 200 OK',
69
'Content-Type: text/plain',
76
def testOperatesOnStreamDirectly(self):
77
def gotResponse(resp):
78
self.assertEquals(resp.code, 200)
79
self.assertEquals(resp.headers.getHeader('Content-Type'),
80
http_headers.MimeType.fromString('text/plain'))
85
return defer.maybeDeferred(stream.read
86
).addCallback(self.assertEquals, '42')
88
req = SimpleRequest(None, 'GET', '/')
90
d = self.doTestSCGI(req)
91
d.addCallback(gotResponse)
93
self.iterate(self.cxn)
95
headers = parseSCGIHeaders(self.cxn.server.data)
97
self.assertEquals(headers[0], ('CONTENT_LENGTH', '0'))
99
self.failUnlessIn(('SCGI', '1'), headers)
101
self.writeLines(self.cxn, ['Status: 200 OK',
102
'Content-Type: text/plain',