← Back to branch summary

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

« back to all changes in this revision

Viewing changes to twisted/web2/test/test_scgi.py

  • Committer: Bazaar Package Importer
  • Author(s): Matthias Klose
  • Date: 2007-01-17 14:52:35 UTC
  • mfrom: (1.1.5 upstream) (2.1.2 etch)
  • Revision ID: james.westby@ubuntu.com-20070117145235-btmig6qfmqfen0om
Tags: 2.5.0-0ubuntu1
New upstream version, compatible with python2.5.

Show diffs side-by-side

added added

removed removed

Lines of Context:
twisted/web2/test/test_scgi.py
 
1
from twisted.trial import unittest
 
2
 
 
3
from twisted.internet import defer
 
4
from twisted.internet.address import IPv4Address
 
5
 
 
6
from twisted.web2.test.test_http import LoopbackRelay, TestConnection
 
7
from twisted.web2.test.test_http import TestClient, HTTPTests, TestRequest
 
8
 
 
9
from twisted.web2.test.test_client import TestServer, ClientTests
 
10
from twisted.web2.test.test_server import SimpleRequest
 
11
 
 
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
 
16
 
 
17
def parseSCGIHeaders(headers):
 
18
    return zip(*[iter(headers.split(':', 1)[1].split('\x00'))]*2)
 
19
 
 
20
class SCGITests(HTTPTests):
 
21
    def connect(self, logFile=None):
 
22
        cxn = TestConnection()
 
23
        cxn.client = self.clientProtocol
 
24
        cxn.server = self.serverProtocol
 
25
        
 
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)
 
30
 
 
31
        return cxn
 
32
 
 
33
        
 
34
class SCGIClientTests(SCGITests, ClientTests):
 
35
    def setUp(self):
 
36
        self.serverProtocol = TestServer()
 
37
 
 
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
 
45
 
 
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'))
 
51
 
 
52
            return defer.maybeDeferred(resp.stream.read
 
53
                                       ).addCallback(self.assertEquals, '42')
 
54
            
 
55
        req = SimpleRequest(None, 'GET', '/')
 
56
 
 
57
        d = self.doTestSCGI(req)
 
58
        d.addCallback(gotResponse)
 
59
 
 
60
        self.iterate(self.cxn)
 
61
 
 
62
        headers = parseSCGIHeaders(self.cxn.server.data)
 
63
 
 
64
        self.assertEquals(headers[0], ('CONTENT_LENGTH', '0'))
 
65
 
 
66
        self.failUnlessIn(('SCGI', '1'), headers)
 
67
        
 
68
        self.writeLines(self.cxn, ['Status: 200 OK',
 
69
                                   'Content-Type: text/plain',
 
70
                                   'Content-Length: 2',
 
71
                                   '',
 
72
                                   '42'])
 
73
 
 
74
        return d
 
75
 
 
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'))
 
81
            
 
82
            stream = resp.stream
 
83
            resp.stream = None
 
84
 
 
85
            return defer.maybeDeferred(stream.read
 
86
                                       ).addCallback(self.assertEquals, '42')
 
87
            
 
88
        req = SimpleRequest(None, 'GET', '/')
 
89
 
 
90
        d = self.doTestSCGI(req)
 
91
        d.addCallback(gotResponse)
 
92
 
 
93
        self.iterate(self.cxn)
 
94
 
 
95
        headers = parseSCGIHeaders(self.cxn.server.data)
 
96
 
 
97
        self.assertEquals(headers[0], ('CONTENT_LENGTH', '0'))
 
98
 
 
99
        self.failUnlessIn(('SCGI', '1'), headers)
 
100
        
 
101
        self.writeLines(self.cxn, ['Status: 200 OK',
 
102
                                   'Content-Type: text/plain',
 
103
                                   'Content-Length: 2',
 
104
                                   '',
 
105
                                   '42'])
 
106
        
 
107
        return d