« back to all changes in this revision
Viewing changes to tests/proxy/__init__.py
- browse files at revision 1202
- compare with another revision
- download diff
- download tarball
- view history from revision 1202
- Committer: Tarmac
- Author(s): Alejandro J. Cura
- Date: 2012-03-08 16:25:23 UTC
- mfrom: (1196.1.15 proxy-tunnel-server)
- Revision ID: tarmac-20120308162523-wk66msuv9f84xpls
- The infrastructure for a QtNetwork based process to tunnel syncdaemon traffic thru proxies. (LP: #929207)
- files added:
- tests/proxy
- tests/proxy/ssl
- ubuntuone/proxy
- files modified:
- Makefile.am
added
removed
tests/proxy/__init__.py
1
# Copyright 2012 Canonical Ltd.
2
#
3
# This program is free software: you can redistribute it and/or modify it
4
# under the terms of the GNU General Public License version 3, as published
5
# by the Free Software Foundation.
6
#
7
# This program is distributed in the hope that it will be useful, but
8
# WITHOUT ANY WARRANTY; without even the implied warranties of
9
# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
10
# PURPOSE. See the GNU General Public License for more details.
11
#
12
# You should have received a copy of the GNU General Public License along
13
# with this program. If not, see <http://www.gnu.org/licenses/>.
14
"""Tests for the Ubuntu One proxy support."""
15
16
from os import path
17
from StringIO import StringIO
18
19
from twisted.application import internet, service
20
from twisted.internet import defer, ssl
21
from twisted.web import http, resource, server
22
23
SAMPLE_CONTENT = "hello world!"
24
SIMPLERESOURCE = "simpleresource"
25
DUMMY_KEY_FILENAME = "dummy.key"
26
DUMMY_CERT_FILENAME = "dummy.cert"
27
28
29
class SaveHTTPChannel(http.HTTPChannel):
30
"""A save protocol to be used in tests."""
31
32
protocolInstance = None
33
34
# pylint: disable=C0103
35
def connectionMade(self):
36
"""Keep track of the given protocol."""
37
SaveHTTPChannel.protocolInstance = self
38
http.HTTPChannel.connectionMade(self)
39
40
41
class SaveSite(server.Site):
42
"""A site that let us know when it's closed."""
43
44
protocol = SaveHTTPChannel
45
46
def __init__(self, *args, **kwargs):
47
"""Create a new instance."""
48
server.Site.__init__(self, *args, **kwargs)
49
# we disable the timeout in the tests, we will deal with it manually.
50
# pylint: disable=C0103
51
self.timeOut = None
52
53
54
class BaseMockWebServer(object):
55
"""A mock webserver for testing"""
56
57
def __init__(self):
58
"""Start up this instance."""
59
self.root = self.get_root_resource()
60
self.site = SaveSite(self.root)
61
application = service.Application('web')
62
self.service_collection = service.IServiceCollection(application)
63
#pylint: disable=E1101
64
self.tcpserver = internet.TCPServer(0, self.site)
65
self.tcpserver.setServiceParent(self.service_collection)
66
self.sslserver = internet.SSLServer(0, self.site, self.get_context())
67
self.sslserver.setServiceParent(self.service_collection)
68
self.service_collection.startService()
69
70
def get_dummy_path(self, filename):
71
"""Path pointing at the dummy certificate files."""
72
base_path = path.dirname(__file__)
73
return path.join(base_path, "ssl", filename)
74
75
def get_context(self):
76
"""Return an ssl context."""
77
key_path = self.get_dummy_path(DUMMY_KEY_FILENAME)
78
cert_path = self.get_dummy_path(DUMMY_CERT_FILENAME)
79
return ssl.DefaultOpenSSLContextFactory(key_path, cert_path)
80
81
def get_root_resource(self):
82
"""Get the root resource with all the children."""
83
raise NotImplementedError
84
85
def get_iri(self):
86
"""Build the iri for this mock server."""
87
#pylint: disable=W0212
88
port_num = self.tcpserver._port.getHost().port
89
return u"http://0.0.0.0:%d/" % port_num
90
91
def get_ssl_iri(self):
92
"""Build the iri for the ssl mock server."""
93
#pylint: disable=W0212
94
port_num = self.sslserver._port.getHost().port
95
return u"https://0.0.0.0:%d/" % port_num
96
97
def stop(self):
98
"""Shut it down."""
99
#pylint: disable=E1101
100
if self.site.protocol.protocolInstance:
101
self.site.protocol.protocolInstance.timeoutConnection()
102
return self.service_collection.stopService()
103
104
105
class SimpleResource(resource.Resource):
106
"""A simple web resource."""
107
108
def __init__(self):
109
"""Initialize this mock resource."""
110
resource.Resource.__init__(self)
111
self.rendered = defer.Deferred()
112
113
def render_GET(self, request):
114
"""Make a bit of html out of the resource's content."""
115
if not self.rendered.called:
116
self.rendered.callback(None)
117
return SAMPLE_CONTENT
118
119
120
class MockWebServer(BaseMockWebServer):
121
"""A mock webserver."""
122
123
def __init__(self):
124
"""Initialize this mock server."""
125
self.simple_resource = SimpleResource()
126
super(MockWebServer, self).__init__()
127
128
def get_root_resource(self):
129
"""Get the root resource with all the children."""
130
root = resource.Resource()
131
root.putChild(SIMPLERESOURCE, self.simple_resource)
132
return root
133
134
135
class FakeTransport(StringIO):
136
"""A fake transport that stores everything written to it."""
137
138
connected = True
139
disconnecting = False
140
141
def loseConnection(self):
142
"""Mark the connection as lost."""
143
self.connected = False
144
self.disconnecting = True
145
146
def getPeer(self):
147
"""Return the peer IAddress."""
148
return None
Loggerhead is a web-based interface for Breezy
Version: 2.0.1