Mar 2 12

Liceo frances – the little ones

by mandel

Here you have or little guys playing rugby:

Is in Spanish, sorry, but is worth watching :)

Feb 28 12

Generating a self-signed SSL certificate

by mandel

I have been writing some integration tests lately between Ubuntu One and proxies which use SSL certificates. The idea behind this tests was to be able to test that we deal correctly with those certificates that are not correct (notify the user, remember exceptions, etc..) For that I wrote this small function that I used to generate the certificates.

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
import os
 
from socket import gethostname
 
from OpenSSL import crypto
 
 
def generate_self_signed_cert(cert_dir, is_valid=True):
    """Generate a SSL certificate.
 
    If the cert_path and the key_path are present they will be overwritten.
    """
    if not os.path.exists(cert_dir):
        os.makedirs(cert_dir)
    cert_path = os.path.join(cert_dir, 'squid.crt')
    key_path = os.path.join(cert_dir, 'squid.key')
 
    if os.path.exists(cert_path):
        os.unlink(cert_path)
    if os.path.exists(key_path):
        os.unlink(key_path)
 
    # create a key pair
    key = crypto.PKey()
    key.generate_key(crypto.TYPE_RSA, 1024)
 
    # create a self-signed cert
    cert = crypto.X509()
    cert.get_subject().C = 'UK'
    cert.get_subject().ST = 'London'
    cert.get_subject().L = 'London'
    cert.get_subject().O = 'Canonical'
    cert.get_subject().OU = 'Ubuntu One'
    cert.get_subject().CN = gethostname() if is_valid else gethostname()[::-1]
    cert.set_serial_number(1000)
    cert.gmtime_adj_notBefore(0)
    cert.gmtime_adj_notAfter(10 * 365 * 24 * 60 * 60) 
    cert.set_issuer(cert.get_subject())
    cert.set_pubkey(key)
    cert.sign(key, 'sha1')
 
    with open(cert_path, 'wt') as fd: 
        fd.write(crypto.dump_certificate(crypto.FILETYPE_PEM, cert))
 
    with open(key_path, 'wt') as fd: 
        fd.write(crypto.dump_privatekey(crypto.FILETYPE_PEM, key))
 
    return cert_path, key_path

I leave to the reader to modify the function to match their needs.

Feb 17 12

QArrow a widget similar to GtkArrow in PyQt

by mandel

Recently in Ubuntu One we have been working or using PyQt for the UI of our application so that we could keep a smaller code base. While doing the move I noticed that we needed to have a widget similar to GtkArrow and to my surprise there is not one.

The following is a small implementation fo such a widget that might help other people that are in need of it. Even maybe someone that cares enough will write it in C++ and will propose it to Qt, sorry but I don’t have the time.

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
 
from PyQt4.QtGui import QPainter, QStyle, QStyleOption, QWidget
 
 
class QArrow(QWidget):
    """Qt implementation similar to GtkArrow."""
 
   UP = 0
   DOWN = 1
   LEFT = 2
   RIGHT = 3
 
   def __init__(self, direction, parent=None):
       """Create a new instance."""
       QWidget.__init__(self, parent)
       if not direction in (QArrow.UP, QArrow.DOWN,
                 QArrow.LEFT, QArrow.RIGHT):
           raise ValueError('Wrong arrow direction.')
       self._direction = direction
 
    def paintEvent(self, event):
        """Paint a nice primitive arrow."""
        opt = QStyleOption()
        opt.initFrom(self)
        p = QPainter(self)
        if self._direction == QArrow.UP:
            primitive = QStyle.PE_IndicatorArrowUp
        elif self._direction == QArrow.DOWN:
            primitive = QStyle.PE_IndicatorArrowDown
        elif self._direction == QArrow.LEFT:
            primitive = QStyle.PE_IndicatorArrowLeft
        else:
            primitive = QStyle.PE_IndicatorArrowRight
        self.style().drawPrimitive(primitive, opt, p, self)

Took longer to think the correct way to do it than the actual coding, at the end is very simple.

Feb 16 12

JQuery accordion with check box.

by mandel

I have been playing around with jquery a little lately and I found that it is not very well documented how to add an input field like a checkbox inside the header of an accordion so that you can click on the header and the accordeon works as expected and the checkbox can be clicked. I’m not an expert in this things but I found the way to work around the issue:

The HTML:

1
2
3
4
5
6
7
8
9
10
11
12
<div id="accordion">
    <div>
        <h3 id='example'>
            <a href='#'>
                <label>Title<input class='header-checkbox' type='checkbox' /></label>
            </a>
        </h3>
        <div>
            <p></p>
        </div>
    </div>
</div>

The Javascript:

$('#example').find('input').click(
    function(e){
        e.stopPropagation();
    }
);

I’m sure this is very easy for any advance js guy, but since I googled and did not find it I though I would be a nice guy and post it here for the next guy.

Feb 13 12

Add a custom context menu in a WebView

by mandel

Yesterday I was working on a small UI that uses a WebKit.WebView to show certain information to the user. My idea yesterday was to try and show a native context menu on the view that does not have the default menu items that the WebView has, the only way I’ve found to this is a as follows:

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
from gi.repository import GObject, Gdk, Gtk, WebKit
from utils import get_data_file
 
 
class CustomView(Gtk.EventBox):
    """Main window used to show the ui and generate signals."""
 
    def __init__(self):
        """Create a new instance."""
        super(CustomView, self).__init__()
        self.set_events(Gdk.EventMask.BUTTON_PRESS_MASK)
        self._create_context_menu()
        self.scroll = Gtk.ScrolledWindow()
        self.scroll.show()
        self.view = WebKit.WebView()
        self.view.show()
        self.scroll.add(self.view)
        # load the html used for the main window, from that point
        # on we will let the html and js do the work and raise
        # signals via the title
        self.view.open(get_data_file(MAIN_WINDOW_HMTL))
        # lets no use the webkit context menu
        settings = self.view.get_settings()
        settings.set_property('enable-default-context-menu',
                              False)
        self.view.set_settings(settings)
        self.add(self.scroll)
        # connect the on button pressed event of the event box
        # so that we can add a context menu
        self.connect('button_press_event', self._on_button_press_event)
 
    def _create_context_menu(self):
        """Create the context menu."""
        self.menu = Gtk.Menu()
        delete_menu = Gtk.MenuItem("Delete Task")
        self.menu.append(delete_menu)
 
    def _on_button_press_event(self, widget, event):
        """Deal with the button press event."""
        if event.button == 3:
            self.menu.popup(None, None, None, None, event.button, event.time)
            self.menu.show_all()

Does any one out there know a better way to do this. I know that there is a “populate-popup” signal I can listen to but I cannot get my head around on how to do exactly what I want, which is remove all defaul menu items and add my own. And, does it make sense to have to do that for every popup signal?

Feb 7 12

Python at codemotion.es

by mandel

As the codemotion.es page states, codemotion is:

Codemotion es el evento que reunirá en España a técnicos, desarrolladores y estudiantes de todas las comunidades y lenguajes. Por primera vez se celebrará en España después de 5 años de éxito en Italia.

Which badly translats to:

Codemotion is the event that will gather technicians, developers and students of different communities and languages. For the first time the event will take place in Spain after 5 years in Italy

Python is within the languages that will take part of the events thanks to the python madrid group. The group will go over the different talk proposals in the next meeting and will try to get you the best talks possible. So if you wanna take part you can do the following

Lets try and make python a better know language in Spain!.

Feb 7 12

Get/Set proxy settings in Gnome with GObject instrospection

by mandel

With GObject introspection is very simple to set the settings of your system trough python. Fist, lets use the command line to find out our current settings:

gsettings list-recursively org.gnome.system.proxy

The following script allows you to retrieve the http proxy settings that you are currently using:

from gi.repository import Gio
 
def get_settings():
    """Get proxy settings."""
    http_settings = Gio.Settings.new('org.gnome.system.proxy.http')
    host = http_settings.get_string('host')
    port = http_settings.get_int('port')
    if http_settings.get_boolean('use-authentication'):
        username = http_settings.get_string('authentication_user')
        password = http_settings.get_string('authentication_password')
    else:
        username = password = None
    return host, port, username, password

Setting them is as easy as getting them:

from gi.repository import Gio
 
def set_settings(host, port, username=None, password=None):
     """Set proxy settings."""
     http_settings = Gio.Settings.new('org.gnome.system.proxy.http')
     http_settings.set_string('host', host)
     http_settings.set_int('port', port)
     if username is not None:
         http_settings.set_boolean('use-authentication', True)
         http_settings.set_string('authentication_user', username)
         http_settings.set_string('authentication_password', password)

This is not utterly complicated but I’m notice that there are not many examples out there, so there you go. There is no code there that can be considered hard but I’d like to point out that if you use the get_value method from the Settings object you will have to call the appropriate get_* method from the returned GVariant, that is:

host = http_settings.get_string('host')

is equal to the following:

host = http_settings.get_value('host').get_string()
Jan 17 12

ReadDirectoryChangesW, if you want it asynch use the correct flag

by mandel

I have had a nice riddle given to me by Tim-Erwin within the comments of Bitten in the ass by ReadDirectoryChangesW and multithreading which I really appreciate. His problem was with the following code:

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
import win32file
import win32con
import win32event
import pywintypes
 
FILE_LIST_DIRECTORY = 0x0001
folder = r"C:some folder to watch"
 
handle = win32file.CreateFile (folder, FILE_LIST_DIRECTORY,
              win32con.FILE_SHARE_READ | win32con.FILE_SHARE_WRITE,
              None, win32con.OPEN_EXISTING,
              win32con.FILE_FLAG_BACKUP_SEMANTICS, None)
buffer = win32file.AllocateReadBuffer(1024)
overlapped = pywintypes.OVERLAPPED()
overlapped.hEvent = win32event.CreateEvent(None, 0, 0, None)
flags = win32con.FILE_NOTIFY_CHANGE_FILE_NAME |
                  win32con.FILE_NOTIFY_CHANGE_DIR_NAME |
                  win32con.FILE_NOTIFY_CHANGE_ATTRIBUTES |
                  win32con.FILE_NOTIFY_CHANGE_SIZE |
                  win32con.FILE_NOTIFY_CHANGE_LAST_WRITE |
                  win32con.FILE_NOTIFY_CHANGE_SECURITY
 
win32file.ReadDirectoryChangesW(handle, buffer, False, flags, overlapped)
print "never reached until a change is done in the folder"
win32event.WaitForSingleObject(overlapped.hEvent, 5000)

As Tim-Erwin said in his comments, the above code was not being blocked in the WaitForSingleObject call but in the ReadDirecotryChangesW one. This is clearly my fault because in the blog post I did not give a very important detail that is needed to get the code working async. In order for the above to work correctly it is imperative that you pass to the CreateFile function the FILE_FLAG_OVERLAPPED flag. That means, that in order to fix the code and be blocked in WaitForSingleObject we have to modify the CreateFile call in the following way:

9
10
11
12
handle = win32file.CreateFile (folder, FILE_LIST_DIRECTORY,
              win32con.FILE_SHARE_READ | win32con.FILE_SHARE_WRITE,
              None, win32con.OPEN_EXISTING,
              win32con.FILE_FLAG_BACKUP_SEMANTICS | win32file.FILE_FLAG_OVERLAPPED , None)

That fixes Tim-Erwin problem and should help any other person with the same issue. Sorry for not being explicit on such an important detail Tim-Erwin, please accept my apologies :) .

Jan 10 12

Adding an alternative storage pool

by mandel

This is here for me to remember the next time I need to do this task:

  1. Copy the default pool definition:

    virsh pool-dumpxml default > pool.xml
  2. edit pool.xml changing the following vars:

    <pool type='dir'>
      <name>{$name}</name>
      <uuid>{$id}</uuid>
      <capacity>43544694784</capacity>
      <allocation>30412328960</allocation>
      <available>13132365824</available>
      <source>
      </source>
      <target>
        <path>{$path}</path>
        <permissions>
          <mode>0700</mode>
          <owner>-1</owner>
          <group>-1</group>
        </permissions>
      </target>
    </pool>
  3. virsh pool-create pool.xml
  4. virsh pool-refresh name

Doing the above you can add a new pool, for example one that is not in you ssd.

Jan 4 12

Cleaning and closing once again

by mandel

In the past days I have been working on implementing a python TestCase that can be used to perform integration tests to the future implementation of proxy support that will be landing in Ubuntu One. The idea of the TestCase is the following:

  • Start a proxy so that connections go throw it. The proxy has to be listening to two different ports, one in which auth is not required and a second one in which auth is required. At the moment the only supported proxy is Squid using base auth.
  • The test case should provide a way to access to the proxy details for subclasses to use.
  • The test case should integrate with the ubuntuone-dev-tools.

Initially, one of the major problems I had was to start squid in two different ports so that:

  • Port A accepts non-auth requests.
  • Port A rejects auth requests.
  • Port B accepts auth requests.

The idea is simple, if you use port A you should never auth while you must in port B, and example configuration of the ACLs and ports is the following:

auth_param basic casesensitive on
# Use a default auth using ncsa and the passed generated file.
auth_param basic program ${auth_process} ${auth_file}
#Recommended minimum configuration:
acl all src all
acl manager proto cache_object
acl localhost src 127.0.0.1/32
acl to_localhost dst 127.0.0.0/8 0.0.0.0/32
#
# Example rule allowing access from your local networks.
# Adapt to list your (internal) IP networks from where browsing
# should be allowed
acl localnet src 10.0.0.0/8	# RFC1918 possible internal network
acl localnet src 172.16.0.0/12	# RFC1918 possible internal network
acl localnet src 192.168.0.0/16	# RFC1918 possible internal network
#
acl SSL_ports port 443		# https
acl SSL_ports port 563		# snews
acl SSL_ports port 873		# rsync
acl Safe_ports port 80		# http
acl Safe_ports port 21		# ftp
acl Safe_ports port 443		# https
acl Safe_ports port 70		# gopher
acl Safe_ports port 210		# wais
acl Safe_ports port 1025-65535	# unregistered ports
acl Safe_ports port 280		# http-mgmt
acl Safe_ports port 488		# gss-http
acl Safe_ports port 591		# filemaker
acl Safe_ports port 777		# multiling http
acl Safe_ports port 631		# cups
acl Safe_ports port 873		# rsync
acl Safe_ports port 901		# SWAT
acl purge method PURGE
acl CONNECT method CONNECT

# make an acl for users that have auth
acl password proxy_auth REQUIRED myportname ${auth_port_number}
acl auth_port_connected myportname ${auth_port_number}
acl nonauth_port_connected myportname ${noauth_port_number}


# Settings used for the tests:
# Allow users connected to the nonauth port
# Allow users authenticated AND connected to the auth port
http_access allow nonauth_port_connected
http_access allow password

#Recommended minimum configuration:
#
# Only allow cachemgr access from localhost
http_access allow manager localhost
http_access deny manager
# Only allow purge requests from localhost
http_access allow purge localhost
http_access deny purge
# Deny requests to unknown ports
http_access deny !Safe_ports
# Deny CONNECT to other than SSL ports
http_access deny CONNECT !SSL_ports
# Example rule allowing access from your local networks.
# Adapt localnet in the ACL section to list your (internal) IP networks
# from where browsing should be allowed
#http_access allow localnet
http_access allow localhost

# And finally deny all other access to this proxy
http_access deny all

#Allow ICP queries from local networks only
icp_access allow localnet
icp_access deny all

# Squid normally listens to port 3128 but we are going to listento two
# different ports, one for auth one for nonauth.
http_port ${noauth_port_number}
http_port ${auth_port_number}

#We recommend you to use at least the following line.
hierarchy_stoplist cgi-bin ?

# Default cache settings.
cache_dir ufs ${spool_temp} 100 16 256

# access log settings
access_log ${squid_temp}/access.log squid

#Default cache stroe log
cache_store_log ${squid_temp}/store.log

#Default pid file name
pid_filename ${squid_temp}/squid.pid

#Default netdb file name:
netdb_filename ${spool_temp}/logs/netdb.state

#Suggested default:
refresh_pattern ^ftp:		1440	20%	10080
refresh_pattern ^gopher:	1440	0%	1440
refresh_pattern -i (/cgi-bin/|?) 0	0%	0
refresh_pattern (Release|Packages(.gz)*)$	0	20%	2880
# example line deb packages
#refresh_pattern (.deb|.udeb)$   129600 100% 129600
refresh_pattern .		0	20%	4320

# Don't upgrade ShoutCast responses to HTTP
acl shoutcast rep_header X-HTTP09-First-Line ^ICY.[0-9]
upgrade_http0.9 deny shoutcast

# Apache mod_gzip and mod_deflate known to be broken so don't trust
# Apache to signal ETag correctly on such responses
acl apache rep_header Server ^Apache
broken_vary_encoding allow apache

extension_methods REPORT MERGE MKACTIVITY CHECKOUT

hosts_file /etc/hosts

# Leave coredumps in the first cache dir
coredump_dir ${spool_temp}

Once the above was achieved the code of the test case was quite simple for Ubuntu O, unfortunatly, it was not that issues in Ubuntu P because there we have squid3 which supports http 1.1 and keeps the proxy keeps the connection alive. The fact that the connection is kept alive means that the reactor has a selectable running because the proxy keep it there. In order to solve the issue I wrote the code so that the server could say that the connection timedout. Here is the code that does it:

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
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
# -*- coding: utf-8 -*-
#
# Copyright 2011 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/>.
"""Test the squid test case."""
import base64
 
from twisted.application import internet, service
from twisted.internet import defer, reactor
from twisted.web import client, error, http, resource, server
 
from ubuntuone.devtools.testcases.squid import SquidTestCase
 
 
SAMPLE_RESOURCE = "<p>Hello World!</p>"
SIMPLERESOURCE = "simpleresource"
THROWERROR = "throwerror"
UNAUTHORIZED = "unauthorized"
 
# ignore common twisted lint errors
# pylint: disable=C0103, W0212
 
 
class ProxyClientFactory(client.HTTPClientFactory):
    """Factory that supports proxy."""
 
    def __init__(self, proxy_url, proxy_port, url, headers=None):
        # we set the proxy details before the init because the parent __init__
        # calls setURL
        self.proxy_url = proxy_url
        self.proxy_port = proxy_port
        self.disconnected_d = defer.Deferred()
        client.HTTPClientFactory.__init__(self, url, headers=headers)
 
    def setURL(self, url):
        self.host = self.proxy_url
        self.port = self.proxy_port
        self.url = url
        self.path = url
 
    def clientConnectionLost(self, connector, reason, reconnecting=0):
        """Connection lost."""
        self.disconnected_d.callback(self)
 
 
class ProxyWebClient(object):
    """Provide useful web methods with proxy."""
 
    def __init__(self, proxy_url=None, proxy_port=None, username=None,
            password=None):
        """Create a new instance with the proxy settings."""
        self.proxy_url = proxy_url
        self.proxy_port = proxy_port
        self.username = username
        self.password = password
        self.factory = None
        self.connectors = []
 
    def _connect(self, url, contextFactory):
        """Perform the connection."""
        scheme, _, _, _ = client._parse(url)
        # pylint: disable=E1101
        if scheme == 'https':
            from twisted.internet import ssl
            if contextFactory is None:
                contextFactory = ssl.ClientContextFactory()
            self.connectors.append(reactor.connectSSL(self.proxy_url,
                                                      self.proxy_port,
                                                      self.factory,
                                                      contextFactory))
        else:
            self.connectors.append(reactor.connectTCP(self.proxy_url,
                                                      self.proxy_port,
                                                      self.factory))
            # pylint: enable=E1101
 
    def _process_auth_error(self, failure, url, contextFactory):
        """Process an auth failure."""
        failure.trap(error.Error)
        if failure.value.status == str(http.PROXY_AUTH_REQUIRED):
            # we try to get the page using the basic auth
            auth = base64.b64encode('%s:%s' % (self.username, self.password))
            auth_header = 'Basic ' + auth.strip()
            self.factory = ProxyClientFactory(self.proxy_url, self.proxy_port,
                            url, headers={'Proxy-Authorization': auth_header})
            self._connect(url, contextFactory)
            return self.factory.deferred
        else:
            return failure
 
    def get_page(self, url, contextFactory=None, *args, **kwargs):
        """Download a webpage as a string.
 
        This method relies on the twisted.web.client.getPage but adds and extra
        step. If there is an auth error the method will perform a second try
        so that the username and password are used.
        """
        self.factory = ProxyClientFactory(self.proxy_url, self.proxy_port, url,
                                          headers={'Connection': 'close'})
        self._connect(url, contextFactory)
        self.factory.deferred.addErrback(self._process_auth_error, url,
                                    contextFactory)
        return self.factory.deferred
 
    @defer.inlineCallbacks
    def shutdown(self):
        """Clean all connectors."""
        for connector in self.connectors:
            yield connector.disconnect()
        defer.returnValue(True)
 
 
class SimpleResource(resource.Resource):
    """A simple web resource."""
 
    def render_GET(self, request):
        """Make a bit of html out of these resource's
        content."""
        return SAMPLE_RESOURCE
 
 
class SaveHTTPChannel(http.HTTPChannel):
    """A save protocol to be used in tests."""
 
    protocolInstance = None
 
    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 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.
        self.timeOut = None
 
 
class MockWebServer(object):
    """A mock webserver for testing"""
 
    def __init__(self):
        """Start up this instance."""
        root = resource.Resource()
        root.putChild(SIMPLERESOURCE, SimpleResource())
 
        root.putChild(THROWERROR, resource.NoResource())
 
        unauthorized_resource = resource.ErrorPage(resource.http.UNAUTHORIZED,
                                                "Unauthorized", "Unauthorized")
        root.putChild(UNAUTHORIZED, unauthorized_resource)
 
        self.site = SaveSite(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.service_collection.startService()
 
    def get_url(self):
        """Build the url for this mock server."""
        #pylint: disable=W0212
        port_num = self.tcpserver._port.getHost().port
        return "http://localhost:%d/" % port_num
 
    @defer.inlineCallbacks
    def stop(self):
        """Shut it down."""
        #pylint: disable=E1101
        # make the connection time out so that is works with squid3 when
        # the connection is kept alive.
        if self.site.protocol.protocolInstance:
            self.site.protocol.protocolInstance.timeoutConnection()
        yield self.service_collection.stopService()
 
 
class ProxyTestCase(SquidTestCase):
    """A squid test with no auth proxy."""
 
    @defer.inlineCallbacks
    def setUp(self):
        """Set the tests."""
        yield super(ProxyTestCase, self).setUp()
        self.ws = MockWebServer()
        self.proxy_client = None
        self.addCleanup(self.teardown_client_server)
        self.url = self.ws.get_url() + SIMPLERESOURCE
 
    def teardown_client_server(self):
        """Clean resources."""
        if self.proxy_client is not None:
            self.proxy_client.shutdown()
            return defer.gatherResults([self.ws.stop(),
                               self.proxy_client.shutdown(),
                               self.proxy_client.factory.disconnected_d])
        else:
            return self.ws.stop()
 
    def access_noauth_url(self, address, port):
        """Access a url throught the proxy."""
        self.proxy_client = ProxyWebClient(proxy_url=address, proxy_port=port)
        return self.proxy_client.get_page(self.url)
 
    def access_auth_url(self, address, port, username, password):
        """Access a url throught the proxy."""
        self.proxy_client = ProxyWebClient(proxy_url=address, proxy_port=port,
                                         username=username, password=password)
        return self.proxy_client.get_page(self.url)
 
    @defer.inlineCallbacks
    def test_noauth_url_access(self):
        """Test accessing to the url."""
        settings = self.get_nonauth_proxy_settings()
        # if there is an exception we fail.
        data = yield self.access_noauth_url(settings['host'],
                                            settings['port'])
        self.assertEqual(SAMPLE_RESOURCE, data)
 
    @defer.inlineCallbacks
    def test_auth_url_access(self):
        """Test accessing to the url."""
        settings = self.get_auth_proxy_settings()
        # if there is an exception we fail.
        data = yield self.access_auth_url(settings['host'],
                                          settings['port'],
                                          settings['username'],
                                          settings['password'])
        self.assertEqual(SAMPLE_RESOURCE, data)
 
    def test_auth_url_401(self):
        """Test failing accessing the url."""
        settings = self.get_auth_proxy_settings()
        # swap password for username to fail
        d = self.failUnlessFailure(self.access_auth_url(settings['host'],
                                        settings['port'], settings['password'],
                                        settings['username']), error.Error)
        return d
 
    def test_auth_url_407(self):
        """Test failing accessing the url."""
        settings = self.get_auth_proxy_settings()
        d = self.failUnlessFailure(self.access_noauth_url(settings['host'],
                                   settings['port']), error.Error)
        return d

The above code is the tests for the test case and the important bits are:

133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
class SaveHTTPChannel(http.HTTPChannel):
    """A save protocol to be used in tests."""
 
    protocolInstance = None
 
    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 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.
        self.timeOut = None

The above defines a protocol that will know the instance that it was used so that we can trigger the time out in a clean up function.

190
191
        if self.site.protocol.protocolInstance:
            self.site.protocol.protocolInstance.timeoutConnection()

This tells the server to time out.

207
208
209
210
211
212
213
214
215
    def teardown_client_server(self):
        """Clean resources."""
        if self.proxy_client is not None:
            self.proxy_client.shutdown()
            return defer.gatherResults([self.ws.stop(),
                               self.proxy_client.shutdown(),
                               self.proxy_client.factory.disconnected_d])
        else:
            return self.ws.stop()

And the clean up function. That is all, now I guess I’ll move to add proxy support or ensure that the test case works on Windows, which is certainly going to be a diff issue.