Extending Selenium with jQuery

Last week I wrote about combining Selenium and py.test and I promised to also talk about my function find_elements_by_jquery().

Selenium by default can find elements by id, CSS selector and XPath, but I often find I already know the query as a jQuery selector, and so frequently it’s easiest just to use that.

We start by overloading the Selenium webdriver. Since the webdriver is exposed through several classes (one per web browser), we do this in a particularly meta way.

from selenium.webdriver.remote.webdriver import WebElement
from selenium.common.exceptions import InvalidSelectorException

def MyWebDriver(base, **kwargs):
    return type('MyWebDriver', (_MyWebDriver, base), kwargs)

class _MyWebDriver(object):
    def create_web_element(self, element_id):
        return MyWebElement(self, element_id)

    def find_elements_by_jquery(self, jq):
        return self.execute_script('''return $('%s').get();''' % jq)

    def find_element_by_jquery(self, jq):
        elems = self.find_elements_by_jquery(jq)
        if len(elems) == 1:
            return elems[0]
        else:
            raise InvalidSelectorException(
                "jQuery selector returned %i elements, expected 1" % len(elems))

We then do a similar implementation for the webelement:

class MyWebElement(WebElement):
    def __repr__(self):
        """Return a pretty name for an element"""

        id = self.get_attribute('id')
        class_ = self.get_attribute('class')

        if len(id) > 0:
            return '#' + id
        elif len(class_) > 0:
            return '.'.join([self.tag_name] + class_.split(' '))
        else:
            return self.tag_name

    def find_elements_by_jquery(self, jq):
        return self.parent.execute_script(
            '''return $(arguments[0]).find('%s').get();''' % jq, self)

    def find_element_by_jquery(self, jq):
        elems = self.find_elements_by_jquery(jq)
        if len(elems) == 1:
            return elems[0]
        else:
            raise InvalidSelectorException(
                "jQuery selector returned %i elements, expected 1" % len(elems))

We can now pass in jQuery selectors for instance b.find_element_by_jquery('#region option:selected'). Or form.find_elements_by_jquery(':input'). It’s especially incredibly powerful when all of your DOM manipulation already works in terms of jQuery selectors.

As an added bonus, overloading the classes lets us add functionality like Firebug style element names (MyWebElement.__repr__) or wrap things like the Wait utility into the webdriver, e.g.

from selenium.webdriver.support.ui import WebDriverWait as Wait
from selenium.common.exceptions import TimeoutException

class FrontendError(Exception):
    pass

# class _MyWebDriver...
    def wait(self, event, timeout=10):
        try:
            Wait(self, timeout).until(event)
        except (TimeoutException, FrontendError) as e:
            # do we have an error dialog
            dialog = self.find_element_by_id('error-dialog')
            if dialog.is_displayed():
                content = dialog.find_element_by_id('error-dialog-content')
                raise FrontendError(content.text)
            else:
                raise e

Combining py.test and Selenium to test webapps

Recently I started adding unit and acceptance tests to a webapp using Selenium, integrated into the existing py.test framework that tests the backend code.

py.test fixtures make using Selenium, via its Python bindings, really straightforward. Here’s how I did it.

First I put all the Selenium related tests in a tests/selenium/ directory. I then created tests/selenium/conftest.py and wrote a fixture to allow tests to access a single instance of the webdriver for the entire session:

import pytest
from selenium import webdriver

browsers = {
    'firefox': webdriver.Firefox,
    'chrome': webdriver.Chrome,
}

@pytest.fixture(scope='session',
                params=browsers.keys())
def driver(request):
    if 'DISPLAY' not in os.environ:
        pytest.skip('Test requires display server (export DISPLAY)')

    b = browsers[request.param]()

    request.addfinalizer(lambda *args: b.quit())

    return b

Note that we’re able to parameterise the fixture so that it runs with multiple browsers. We then add a per-function fixture that sets up the session for an individual test:

@pytest.fixture
def b(driver, url):
    b = driver
    b.set_window_size(1200, 800)
    b.get(url)

    return b

A fixture can refer to other fixtures of more generic scope. So url is a fixture that accesses the optional --url property.

def pytest_addoption(parser):
    parser.addoption('--url', action='store',
                     default='http://localhost/portal/portal.html')

@pytest.fixture(scope='session')
def url(request):
    return request.config.option.url

These fixtures are available for all tests in that package. Tests have the form:

def test_badger(b):
    # test goes here

We can also create per-module fixtures, that optionally inherit our generic fixtures. Say for example we want to run a number of tests (e.g. for WCAG 2.0 compliance) on a number of parameterised instances of the set-up webapp. We might do this in test_wcag.py:

import pytest

@pytest.fixture(scope='module')
def wcag(driver, url):
    """
    Set up a single session for these tests.
    """

    b = driver
    b.set_window_size(1200, 800)
    b.get(url)

    # do stuff here with Selenium to set up webapp
    
    return b

We can now write tests ((find_elements_by_jquery() is a method I’ve added in an extension of Selenium’s webdriver, and is a topic for another post.)) in this module, e.g.

@pytest.mark.wcagF17
@pytest.mark.wcagF62
@pytest.mark.wcagF77
def test_unique_ids(wcag):
    """
    All ids in the document should be unique.
    """

    elems = wcag.find_elements_by_jquery('[id]')
    ids = map(lambda e: e.get_attribute('id'), elems)

    assert len(elems) >= 1 # sanity check
    assert util.unique(ids)

Again, we can parameterise this fixture to set up the webapp in a number of different ways. Note that we have to use driver as our fixture, not b. This is because we can only refer to fixtures more general in scope than the one we are writing.

let us not mourn telepathy-python

But let us instead celebrate gobject-introspection.

While there has been no official announcement, I think it’s probably time we declared telepathy-python to have passed away. Deprecated. In the great attic of the sky.

The python-telepathy client-side bindings have been in a state of disrepair for a long time now, not able to do the heavy lifting that other bindings such as tp-glib and tp-qt4 have been focusing on.

telepathy-butterfly is not really maintained, isn’t supported properly by empathy master, and is looking to be replaced in the future by the MSN Live XMPP support. In the meantime bug #663829 migrates existing butterfly accounts to telepathy-haze. I’m not sure about telepathy-sunshine, but I suspect it also won’t work with the latest Empathy.

Through work I started, and a bunch of other people have continued, telepathy-glib client-side bindings have been growing increasingly more introspectable as time passes. This is now used by gnome-shell, and it works pretty well.

Since someone asked today, here’s how you set your global presence in Python:

from gi.repository import TelepathyGLib as Tp
from gi.repository import GObject

loop = GObject.MainLoop()
am = Tp.AccountManager.dup()
am.prepare_async(None, lambda *args: loop.quit(), None)
loop.run()

am.set_all_requested_presences(Tp.ConnectionPresenceType.AVAILABLE,
    'available', 'Set by script')

Unfortunately we can’t introspect the service-side bindings, they do some things that are very creative and clever, but that gobject-introspection just doesn’t understand. Which means you can’t write a CM using PyGObject yet. The solution, I think, is likely to come with a future, GVariant-based telepathy-glib, but I’m still trying to figure out how we can make things (like mixins) work in a way that g-i can understand.

fixing button theming with GtkBuilder

This is a bit icky. It would be neater if the Python bindings exposed hildon_gtk_widget_set_theme_size(), but not much. So, to fix the button theming if you've created your interface with GtkBuilder, it looks something like this:

# these aren't exported anywhere, copied from Maemo GTK+
HILDON_HEIGHT_FINGER = 70
HILDON_HEIGHT_THUMB = 105

# fix theming
for widget in self.ui.get_objects():
    if not isinstance(widget, gtk.Button): continue
    # hildon_gtk_widget_set_theme_size is not bound into Python
    if widget.get_name().startswith('largebutton'):
        widget.set_size_request(-1, HILDON_HEIGHT_THUMB)
    elif widget.get_name().startswith('kpbutton'):
        widget.set_size_request(HILDON_HEIGHT_THUMB, HILDON_HEIGHT_THUMB)
    widget.set_name('HildonButton-thumb')

happy new year

Had a pleasant day off eating snacks, watching The Pretender with friends and hacking on my tram tracking app. I added geolocation, which meant needing to test on the device, so I had to package up python-suds for Maemo (git-buildpackage repo, Maemo package).

The app actually runs quite nicely on the device, although each SOAP query is a little slower than in Scratchbox. This makes the Update Database quite a bit slower (also possibly calling COMMIT after each INSERT is a little expensive, I'm not sure). Otherwise things are quite zippy, including searching by location.

I'm not entirely sure I'm using the location API correctly, I don't seem to receive any updates to the location. I think it does some caching to speed up lookups and cut down on signals, so I'll need to try it from another location, but I don't even seem to receive an initial signal when the GPS locks.

I started having a go at packaging the application itself, but ran into some error I don't understand (Debian always seems to throw obscure errors when I try to package things). Regardless, the branch is here. Would love some help here.

It seems like all the fundamentals are now in place; including favourites, status messages and geolocation. Still want to add support for tracking individual trams. Also need to tweak the interfaces, buttons don't look like they have the right texture. Was thinking of using my Google Maps/GtkWebKit experiments to add a “View On Map” option for location based searches.

a hacky way of monitoring messages in Telepathy

Sometimes people come up with some creative solutions to solve their problems. The correct way to monitor data, such as messages, coming from Telepathy is to write an Observer, however sometimes you just want to get a feed of all of the text messages (e.g. so you can feed it to your keyboard's LCD or something).

The following is a pure D-Bus solution (although it includes telepathy.interfaces to cut down on typing). It listens to all Channel.Type.Text.Received signals, looks up the connection they came from and resolves the sender handle to a name. However note: it makes a lot more D-Bus calls than is required with Telepathy. Really you should cache the results for these handles and listen to the signals that tell you when that information has updated. If you were doing things properly, that's what you'd do.

import dbus, gobject
from dbus.mainloop.glib import DBusGMainLoop
from telepathy.interfaces import *
from datetime import datetime

dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
bus = dbus.SessionBus()

def message(id, timestamp, sender, type, flags, text, path=None):

    # path is the object path of the channel, from this we can derive the
    # object path of the connection, and acquire a proxy to it
    service = '.'.join(path.split('/')[1:8])
    conn_path = '/' + service.replace('.', '/')
    conn = bus.get_object(service, conn_path)

    # request the alias and id of the sender handle
    d = dbus.Interface(conn, CONNECTION_INTERFACE_CONTACTS).GetContactAttributes([sender],
        [CONNECTION, CONNECTION_INTERFACE_ALIASING], False)
    alias = d[sender].get(CONNECTION_INTERFACE_ALIASING + '/alias',
        d[sender].get(CONNECTION + '/contact-id', "Unknown"))

    dt = datetime.fromtimestamp(timestamp)

    print "%s <%s> %s" % (dt.strftime('%H:%M'), alias, text)

# listen to all Channel.Type.Text.Received signals
bus.add_signal_receiver(message,
    dbus_interface=CHANNEL_TYPE_TEXT,
    signal_name='Received',
    path_keyword='path')

loop = gobject.MainLoop()
loop.run()

Like I said, this is not efficient use of Telepathy. If it eats your D-Bus, don't blame me.

On the other hand, this has actually started a conversation about possible new convenience classes for telepathy-python.

bowl and chopsticks

Melbourne Tram Tracker for the N900

So Collabora's robotic and non-robotic overlords very graciously bought everyone on staff an N900 for Christmas. In my opinion, it's actually a very nice phone (although possibly a little on the large side); but the let down is there just isn't the same host of applications for it. Still, possessing both the tools and the skills, I figured I should do something about this, rather than complain.

One of the most useful iPhone applications in Melbourne is the real-time tram tracker. For stops without a display board, you can type in the stop ID and get the upcoming arrivals at that stop. You can also find nearby stops via GPS and a bunch of other things. It turns out that Yarra Trams offer a SOAP WSDL web service that is reasonably well documented, so I've spent a few days putting together a basic tram tracker for Maemo 5 (even if only two people will ever use it).


It currently can show upcoming trams for a stop by ID or by searching for stops by road names. Could possibly also do things like search for stop by route. There is a lot of information available. It doesn't yet do searching by location; the information is in the database, I've just not yet looked at how the location APIs work yet. Also need to add support for storing favourites.

I also want to add support for tracking a tram by tram ID. I'm wondering if it's possible to use the GPS to detect periods of immobility and check the upcoming tram stop after the tram starts moving again. I habitually miss stops; so what I think would be neat is to dial in a stop number or cross road you're looking for, and have your phone notify you when you're approaching it.

The web service uses python-suds, which is unfortunately not packaged for Debian, so I can't just rebuild it for Maemo (if anyone wants to package this up for me, that would be really awesome). Then I'll find out how well my app actually runs on the device.

In case anyone cares, the source code is here.

a threaded processing queue in PyGTK

I'm currently writing a PyGTK client that needs to make network requests using a library that doesn't integrate with the GLib mainloop (python-suds), so I found myself wanting to be able to make network requests without blocking the mainloop, and getting callbacks in my main thread when operations were done. The pattern to use is clearly having a dedicated network thread. In C I might have used GAsyncQueue, however I've found myself quite liking queue.Queue.

The following is a fairly generic class for queuing asynchronous requests. Calling the add_request() method from the main thread queues a function to be run in the worker thread. If the callback or error keywords are provided, these will then be called from the GLib mainloop in the main thread (queued via g_idle_add).

from threading import Thread
from Queue import Queue

import gobject

class ThreadQueue(object):
    def __init__(self):
        self.q = Queue()

        t = Thread(target=self._thread_worker)
        t.setDaemon(True)
        t.start()

    def add_request(self, func, *args, **kwargs):
        """Add a request to the queue. Pass callback= and/or error= as
           keyword arguments to receive return from functions or exceptions.
        """

        self.q.put((func, args, kwargs))

    def _thread_worker(self):
        while True:
            request = self.q.get()
            self.do_request(request)
            self.q.task_done()

    def do_request(self, (func, args, kwargs)):
        if 'callback' in kwargs:
            callback = kwargs['callback']
            del kwargs['callback']
        else:
            callback = None

        if 'error' in kwargs:
            error = kwargs['error']
            del kwargs['error']
        else:
            error = None

        try:
            r = func(*args, **kwargs)
            if not isinstance(r, tuple): r = (r,)
            if callback: self.do_callback(callback, *r)
        except Exception, e:
            if error: self.do_callback(error, e)
            else: print "Unhandled error:", e

    def do_callback(self, callback, *args):
        def _callback(callback, args):
            callback(*args)
            return False

        gobject.idle_add(_callback, callback, args)

We can then inherit this class to provide setup for our specific application:

class WebService(ThreadQueue):
    def __init__(self, guid=None, **kwargs):
        """Initialise the service. If guid is not provided, one will be
           requested (returned in the callback). Pass callback= or error=
           to receive notification of readiness."""
        ThreadQueue.__init__(self)

        self.guid = guid
        self.add_request(self._setup_client, **kwargs)

    def _setup_client(self):
        print "Setting up client"

        ...

        return self.guid

Which we call from our program like this:

class Client(object):
    def __init__(self):
        self.w = WebService(guid=guid, callback=self.client_ready)

    def client_ready(self, guid):
        print "client ready:", guid

gobject.threads_init()

Client()

gtk.main()

What's really cool though is adding methods to the API that are called asynchronously for you. Python makes this possible through the power of decorators. Add the following decorator to a method, and it instead of it being called directly, it will be added to the processing queue.

def async_method(func):
    """Makes the given method asynchronous, meaning when it is called it
       will be queued with add_request.
    """

    def bound_func(obj, *args, **kwargs):
        obj.add_request(func, obj, *args, **kwargs)

    return bound_func

class WebService(ThreadQueue):

    @async_method
    def GetStopInformation(self, stopNo):
        print "Requesting information for stop", stopNo

        ...

And that's it! If you can't follow it, don't worry too much. This is possibly the most Pythonesque bit of code I've ever written, but I've tried to make it generic enough that other people can use it for whatever they need. It's currently part of my app that's beginning to take shape, but the source is here.

Incidently, Maemo people: are there Glade definition files allowing me to use Hildon widgets, GtkBuild and Glade 3? That would be super awesome if there were.

Useful python trick of the day: dict.get()

How often in Python have you written something like:

d[k] if k in d else "Default"

Admittedly this has gotten a lot shorter since the inclusion of ternary operators in Python, but did you know that Python provides a get method for dictionaries that allows you to provide a default?

d.get(k, "Default")

I only found out today when reviewing code. Exciting!

decorator factory for dbus-python methods

This is a crazy idea I had; that I want to share with people.

When you're implementing an object in dbus-python, you decorate your published method calls like this:

class ExampleObserver(dbus.service.Object):
    ...

    @dbus.service.method(dbus_interface=telepathy.interfaces.CLIENT_OBSERVER,
                         in_signature='ooa(oa{sv})oaoa{sv}',
                         out_signature='')
    def ObserveChannels(self, account, connection, channels, dispatch_operation,
                        requests_satisfied, observer_info):
        ...

The input and output signatures are incredibly easy to get wrong. The thing is, most D-Bus APIs (e.g. Telepathy) have a specification that contains these arguments. Some APIs (e.g. Telepathy-Python) provide generated code including interface names and constants. So why can't we do something more like?

class ExampleObserver(dbus.service.Object):
    ...

    @telepathy.decorators.Telepathy.Client.Observer.ObserveChannels
    def ObserveChannels(self, account, connection, channels, dispatch_operation,
                        requests_satisfied, observer_info):
        ...

With a decorator factory that looks up the parameters and then wraps the dbus.service.method factory.

Well, I just wrote a proof-of-concept. It looks something like this:

class Decorators(object):
    methods = {
        'org.freedesktop.DBus.Properties.GetAll': [ 's', 'a{sv}' ],
        'org.freedesktop.DBus.Properties.Get': [ 'ss', 'v' ],
        'org.freedesktop.Telepathy.Client.Observer.ObserveChannels': [ 'ooa(oa{sv})oaoa{sv}', '' ],
    }

    def __init__(self, namespace):
        self._namespace = namespace

    def __getattr__(self, key):
        return Decorators('%s.%s' % (self._namespace, key))

    def __call__(self, func):
        iface = self._namespace.rsplit('.', 1)[0]
        in_sig, out_sig = self.methods[self._namespace]
        return dbus.service.method(dbus_interface=iface,
                                   in_signature=in_sig,
                                   out_signature=out_sig)(func)

    def __str__(self):
        return self._namespace

decorators = Decorators('org.freedesktop')

Obviously in the real version, it would have a generated map of functions, or map of interfaces each with a map of functions, and a way to handle signals, but neat huh?