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
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
|
# Copyright 2012 Canonical Ltd.
#
# This program is free software: you can redistribute it and/or modify it
# under the terms of the GNU General Public License version 3, as published
# by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful, but
# WITHOUT ANY WARRANTY; without even the implied warranties of
# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
# PURPOSE. See the GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License along
# with this program. If not, see <http://www.gnu.org/licenses/>.
"""Tests for the Ubuntu One proxy support."""
from os import path
from StringIO import StringIO
from twisted.application import internet, service
from twisted.internet import defer, ssl
from twisted.web import http, resource, server
SAMPLE_CONTENT = "hello world!"
SIMPLERESOURCE = "simpleresource"
DUMMY_KEY_FILENAME = "dummy.key"
DUMMY_CERT_FILENAME = "dummy.cert"
class SaveHTTPChannel(http.HTTPChannel):
"""A save protocol to be used in tests."""
protocolInstance = None
# pylint: disable=C0103
def connectionMade(self):
"""Keep track of the given protocol."""
SaveHTTPChannel.protocolInstance = self
http.HTTPChannel.connectionMade(self)
class SaveSite(server.Site):
"""A site that let us know when it's closed."""
protocol = SaveHTTPChannel
def __init__(self, *args, **kwargs):
"""Create a new instance."""
server.Site.__init__(self, *args, **kwargs)
# we disable the timeout in the tests, we will deal with it manually.
# pylint: disable=C0103
self.timeOut = None
class BaseMockWebServer(object):
"""A mock webserver for testing"""
def __init__(self):
"""Start up this instance."""
self.root = self.get_root_resource()
self.site = SaveSite(self.root)
application = service.Application('web')
self.service_collection = service.IServiceCollection(application)
#pylint: disable=E1101
self.tcpserver = internet.TCPServer(0, self.site)
self.tcpserver.setServiceParent(self.service_collection)
self.sslserver = internet.SSLServer(0, self.site, self.get_context())
self.sslserver.setServiceParent(self.service_collection)
self.service_collection.startService()
def get_dummy_path(self, filename):
"""Path pointing at the dummy certificate files."""
base_path = path.dirname(__file__)
return path.join(base_path, "ssl", filename)
def get_context(self):
"""Return an ssl context."""
key_path = self.get_dummy_path(DUMMY_KEY_FILENAME)
cert_path = self.get_dummy_path(DUMMY_CERT_FILENAME)
return ssl.DefaultOpenSSLContextFactory(key_path, cert_path)
def get_root_resource(self):
"""Get the root resource with all the children."""
raise NotImplementedError
def get_iri(self):
"""Build the iri for this mock server."""
#pylint: disable=W0212
port_num = self.tcpserver._port.getHost().port
return u"http://0.0.0.0:%d/" % port_num
def get_ssl_iri(self):
"""Build the iri for the ssl mock server."""
#pylint: disable=W0212
port_num = self.sslserver._port.getHost().port
return u"https://0.0.0.0:%d/" % port_num
def stop(self):
"""Shut it down."""
#pylint: disable=E1101
if self.site.protocol.protocolInstance:
self.site.protocol.protocolInstance.timeoutConnection()
return self.service_collection.stopService()
class SimpleResource(resource.Resource):
"""A simple web resource."""
def __init__(self):
"""Initialize this mock resource."""
resource.Resource.__init__(self)
self.rendered = defer.Deferred()
def render_GET(self, request):
"""Make a bit of html out of the resource's content."""
if not self.rendered.called:
self.rendered.callback(None)
return SAMPLE_CONTENT
class MockWebServer(BaseMockWebServer):
"""A mock webserver."""
def __init__(self):
"""Initialize this mock server."""
self.simple_resource = SimpleResource()
super(MockWebServer, self).__init__()
def get_root_resource(self):
"""Get the root resource with all the children."""
root = resource.Resource()
root.putChild(SIMPLERESOURCE, self.simple_resource)
return root
class FakeTransport(StringIO):
"""A fake transport that stores everything written to it."""
connected = True
disconnecting = False
def loseConnection(self):
"""Mark the connection as lost."""
self.connected = False
self.disconnecting = True
def getPeer(self):
"""Return the peer IAddress."""
return None
|