Logo Search packages:      
Sourcecode: ubuntuone-dev-tools version File versions  Download package


# -*- coding: utf-8 -*-

# Author: Guillermo Gonzalez <guillermo.gonzalez@canonical.com>
# Copyright 2009-2010 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
# 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/>.

"""Base tests cases and test utilities."""

from __future__ import with_statement

import contextlib
import os
import shutil
import sys

from functools import wraps

from twisted.internet import defer
from twisted.trial.unittest import TestCase, SkipTest

# DBusRunner for DBusTestCase using tests
from ubuntuone.devtools.services.dbus import DBusRunner

# pylint: disable=F0401,C0103
    import dbus
except ImportError:
    dbus = None

    import dbus.service as service
except ImportError:
    service = None

    from dbus.mainloop.glib import DBusGMainLoop
except ImportError:
    DBusGMainLoop = None

# pylint: enable=F0401,C0103
def environ(env_var, new_value):
    """context manager to replace/add an environ value"""
    old_value = os.environ.get(env_var, None)
    os.environ[env_var] = new_value
    if old_value is None:
        os.environ[env_var] = old_value

def _id(obj):
    """Return the obj calling the funct."""
    return obj

# pylint: disable=C0103
def skipTest(reason):
    """Unconditionally skip a test."""

    def decorator(test_item):
        """Decorate the test so that it is skipped."""
        if not (isinstance(test_item, type) and\
            issubclass(test_item, TestCase)):

            def skip_wrapper(*args, **kwargs):
                """Skip a test method raising an exception."""
                raise SkipTest(reason)
            test_item = skip_wrapper

        # tell twisted.trial.unittest to skip the test, pylint will complain
        # since it thinks we are redefining a name out of the scope
        # pylint: disable=W0621,W0612
        test_item.skip = reason
        # pylint: enable=W0621,W0612
        # because the item was skipped, we will make sure that no
        # services are started for it
        if hasattr(test_item, "required_services"):
            # pylint: disable=W0612
            test_item.required_services = lambda *args, **kwargs: []
            # pylint: enable=W0612
        return test_item
    return decorator

def skipIf(condition, reason):
    """Skip a test if the condition is true."""
    if condition:
        return skipTest(reason)
    return _id

def skipIfOS(current_os, reason):
    """Skip test for a particular os or lists of them."""
    if os:
        if sys.platform in current_os or sys.platform == current_os:
            return skipTest(reason)
        return _id
    return _id

def skipIfNotOS(current_os, reason):
    """Skip test we are not in a particular os."""
    if os:
        if sys.platform not in current_os or\
            sys.platform != current_os:
            return skipTest(reason)
        return _id
    return _id

# pylint: enable=C0103

00131 class InvalidSessionBus(Exception):
    """Error when we are connected to the wrong session bus in tests."""

00135 class FakeDBusInterface(object):
    """A fake DBusInterface..."""

00138     def shutdown(self, with_restart=False):
        """...that only knows how to go away"""

00142 class BaseTestCase(TestCase):
    """Base TestCase with helper methods to handle temp dir.

    This class provides:
        mktemp(name): helper to create temporary dirs
        rmtree(path): support read-only shares
        makedirs(path): support read-only shares


00152     def required_services(self):
        """Return the list of required services for DBusTestCase."""
        return []

00156     def mktemp(self, name='temp'):
        """Customized mktemp that accepts an optional name argument."""
        tempdir = os.path.join(self.tmpdir, name)
        if os.path.exists(tempdir):
        return tempdir

00165     def tmpdir(self):
        """Default tmpdir: module/class/test_method."""
        # check if we already generated the root path
        root_dir = getattr(self, '__root', None)
        if root_dir:
            return root_dir
        max_filename = 32  # some platforms limit lengths of filenames
        base = os.path.join(self.__class__.__module__[:max_filename],
        # use _trial_temp dir, it should be os.gwtcwd()
        # define the root temp dir of the testcase, pylint: disable=W0201
        self.__root = os.path.join(os.getcwd(), base)
        return self.__root

00180     def rmtree(self, path):
        """Custom rmtree that handle ro parent(s) and childs."""
        if not os.path.exists(path):
        # change perms to rw, so we can delete the temp dir
        if path != getattr(self, '__root', None):
            os.chmod(os.path.dirname(path), 0755)
        if not os.access(path, os.W_OK):
            os.chmod(path, 0755)
        # pylint: disable=W0612
        for dirpath, dirs, files in os.walk(path):
            for dirname in dirs:
                if not os.access(os.path.join(dirpath, dirname), os.W_OK):
                    os.chmod(os.path.join(dirpath, dirname), 0777)

00196     def makedirs(self, path):
        """Custom makedirs that handle ro parent."""
        parent = os.path.dirname(path)
        if os.path.exists(parent):
            os.chmod(parent, 0755)

@skipIf(dbus is None or service is None or DBusGMainLoop is None,
    "The test requires dbus.")
00206 class DBusTestCase(BaseTestCase):
    """Test the DBus event handling."""

00209     def required_services(self):
        """Return the list of required services for DBusTestCase."""
        services = super(DBusTestCase, self).required_services()
        return services

00216     def setUp(self):
        """Setup the infrastructure fo the test (dbus service)."""
        # Class 'BaseTestCase' has no 'setUp' member
        # pylint: disable=E1101
        # dbus modules will be imported by the decorator
        # pylint: disable=E0602
        yield super(DBusTestCase, self).setUp()

        # We need to ensure DBUS_SESSION_BUS_ADDRESS is private here
        from urllib import unquote
        bus_address = os.environ.get('DBUS_SESSION_BUS_ADDRESS', None)
        if os.path.dirname(unquote(bus_address.split(',')[0].split('=')[1])) \
                != os.path.dirname(os.getcwd()):
            raise InvalidSessionBus('DBUS_SESSION_BUS_ADDRES is wrong.')

        # Set up the main loop and bus connection
        self.loop = DBusGMainLoop(set_as_default=True)
        self.bus = dbus.bus.BusConnection(address_or_type=bus_address,

        # Monkeypatch the dbus.SessionBus/SystemBus methods, to ensure we
        # always point at our own private bus instance.
        self.patch(dbus, 'SessionBus', lambda: self.bus)
        self.patch(dbus, 'SystemBus', lambda: self.bus)

        # Check that we are on the correct bus for real
# Disable this for now, because our tests are extremely broken :(
#        bus_names = self.bus.list_names()
#        if len(bus_names) > 2:
#            raise InvalidSessionBus('Too many bus connections: %s (%r)' %
#                                    (len(bus_names), bus_names))

        # monkeypatch busName.__del__ to avoid errors on gc
        # we take care of releasing the name in shutdown
        service.BusName.__del__ = lambda _: None
        yield self.bus.set_exit_on_disconnect(False)
        self.signal_receivers = set()

00255     def tearDown(self):
        """Cleanup the test."""
        yield self.bus.flush()
        yield self.bus.close()
        yield super(DBusTestCase, self).tearDown()

Generated by  Doxygen 1.6.0   Back to index