← Back to branch summary

~alecu/ubuntuone-storage-protocol/timestamp-server

« back to all changes in this revision

Viewing changes to tests/test_proxy_tunnel.py

  • Committer: Tarmac
  • Author(s): Rodney Dawes
  • Date: 2010-11-16 15:10:48 UTC
  • mfrom: (121.1.4 use-packages)
  • Revision ID: tarmac-20101116151048-b0e20j7lorb4yhe1
Switch to using packaged mocker and ubuntuone-dev-tools
Use pyflakes with u1lint and also run pep8
Fix a lot of pylint/pyflakes/pep8 errors

Show diffs side-by-side

added added

removed removed

Lines of Context:
tests/test_proxy_tunnel.py
2
2
#
3
3
# Copyright 2009 Canonical Ltd.
4
4
#
5
 
# This program is free software: you can redistribute it and/or modify it 
 
5
# This program is free software: you can redistribute it and/or modify it
6
6
# under the terms of the GNU Affero General Public License version 3,
7
7
# as published by the Free Software Foundation.
8
8
#
9
 
# This program is distributed in the hope that it will be useful, but 
10
 
# WITHOUT ANY WARRANTY; without even the implied warranties of 
11
 
# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR 
 
9
# This program is distributed in the hope that it will be useful, but
 
10
# WITHOUT ANY WARRANTY; without even the implied warranties of
 
11
# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
12
12
# PURPOSE.  See the GNU Affero General Public License for more details.
13
13
#
14
14
# You should have received a copy of the GNU Affero General Public License
24
24
from twisted.test.proto_helpers import StringTransport
25
25
from ubuntuone.storageprotocol.proxy_tunnel import ProxyTunnelFactory
26
26
 
 
27
 
27
28
class FakeTransport(StringTransport):
28
29
    """Fake transport"""
29
30
    def __init__(self, reply_func, remote_func=None):
53
54
        """Fake starttls"""
54
55
        self.clear()
55
56
 
 
57
 
56
58
class FakeConnectHTTPS(object):
57
59
    """Fake HTTPS Connection"""
58
60
    def __init__(self, host, port, factory, proxy_callback,
77
79
        """Peer callback"""
78
80
        self.peer_callback_f(data, self.protocol, self.factory)
79
81
 
 
82
 
80
83
def make_server(proxy_callback, peer_callback, auth):
81
84
    """Make a server"""
82
85
    d = defer.Deferred()
111
114
                     user, passwd, peer_callback)
112
115
    return d
113
116
 
 
117
 
114
118
def test_response(response_string, auth=None):
115
119
    """Test the response"""
116
120
    def response(data, proto, fact):
126
130
            proto.dataReceived(response_string)
127
131
    return make_server(response, lambda x, y, z: None, auth)
128
132
 
 
133
 
129
134
class TunnelTests(TwistedTestCase):
130
135
    """Test the proxy tunnel"""
131
136
    def test_connect(self):
144
149
    def test_auth_required(self):
145
150
        """Test auth requriement"""
146
151
        d = defer.Deferred()
147
 
        d2 = test_response("HTTP/1.0 407 Proxy Authentication Required\r\n\r\n")
 
152
        d2 = test_response(
 
153
            "HTTP/1.0 407 Proxy Authentication Required\r\n\r\n")
148
154
        d2.addCallbacks(
149
155
            lambda r: d.errback(Exception("error: connection made")),
150
156
            lambda r: d.callback("ok"))
168
174
    def test_echo(self):
169
175
        """Test echo"""
170
176
        d = defer.Deferred()
 
177
 
171
178
        def response(data, proto, fact):
172
179
            """Response callback"""
173
180
            proto.dataReceived("HTTP/1.0 200 Connection Made\r\n\r\n")
 
181
 
174
182
        def peer_callback(data, proto, fact):
175
183
            """Peer callback"""
176
184
            proto.dataReceived(data)
 
185
 
177
186
        class TestClient(Protocol):
178
187
            """Test echo Client"""
179
188
            __message = "hello world!"
191
200
                if data == self.__message:
192
201
                    d.callback("ok")
193
202
                else:
194
 
                    d.errback(Exception("wrong data: "+data))
 
203
                    d.errback(Exception("wrong data: " + data))
195
204
 
196
205
        class TestClientFactory(ClientFactory):
197
206
            """Factory for test echo client"""
211
220
    def test_connection_lost(self):
212
221
        """Test connection loss"""
213
222
        d = defer.Deferred()
 
223
 
214
224
        def response(data, proto, fact):
215
225
            """Response callback"""
216
226
            proto.dataReceived("HTTP/1.0 200 Connection Made\r\n\r\n")
 
227
 
217
228
        def peer_callback(data, proto, fact):
218
229
            """Peer callback"""
219
230
            proto.connectionLost("goodbye")
236
247
        class TestClientFactory(ClientFactory):
237
248
            """Factory for test client"""
238
249
            protocol = TestClient
 
250
 
239
251
            def __init__(self):
240
252
                """Initialize ourselves"""
241
253
                pass
249
261
        return d
250
262
 
251
263
 
252
 
 
253
264
def test_suite():
254
265
    """Test suite"""
255
266
    return unittest.TestLoader().loadTestsFromName(__name__)
256
267
 
 
268
 
257
269
if __name__ == "__main__":
258
270
    unittest.main()