Logo Search packages:      
Sourcecode: ubuntuone-control-panel version File versions  Download package

test_replication_client.py

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

# Authors: Natalia B. Bidart <natalia.bidart@canonical.com>
#
# Copyright 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
# 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 DBus service when accessing desktopcouch replications."""

from twisted.internet.defer import inlineCallbacks

from ubuntuone.controlpanel import replication_client
from ubuntuone.controlpanel.tests import TestCase

EXCLUSIONS = set()


00029 class FakedReplication(object):
    """Faked a DC replication exclusion."""

    def __init__(self):
        self.all_exclusions = lambda: EXCLUSIONS
        self.replicate = EXCLUSIONS.remove
        self.exclude = EXCLUSIONS.add


00038 class FakedReplicationModule(object):
    """Faked a DC replication module."""

    ReplicationExclusion = FakedReplication


00044 class ReplicationsTestCase(TestCase):
    """Test for the replications client methods."""

    def setUp(self):
        super(ReplicationsTestCase, self).setUp()

        orig_get_proxy = replication_client.get_replication_proxy

        def get_proxy():
            """Fake the proxy getter."""
            return orig_get_proxy(replication_module=FakedReplicationModule())

        self.patch(replication_client, 'get_replication_proxy', get_proxy)

    def tearDown(self):
        EXCLUSIONS.clear()
        super(ReplicationsTestCase, self).tearDown()

    @inlineCallbacks
00063     def test_no_pairing_record(self):
        """Handle ValueError from replication layer."""

        def no_pairing_record(*args, **kwargs):
            """Fail with ValueError."""
            raise ValueError('No pairing record.')

        self.patch(FakedReplicationModule, 'ReplicationExclusion',
                   no_pairing_record)

        yield self.assertFailure(replication_client.get_replications(),
                                 replication_client.NoPairingRecord)

    @inlineCallbacks
00077     def test_get_replications(self):
        """Replications are correctly retrieved."""
        result = yield replication_client.get_replications()
        self.assertEqual(result, replication_client.REPLICATIONS)

    @inlineCallbacks
00083     def test_get_exclusions(self):
        """Exclusions are correctly retrieved."""
        replications = yield replication_client.get_replications()
        for rep in replications:
            yield replication_client.exclude(rep)

        result = yield replication_client.get_exclusions()
        self.assertEqual(result, replications)

    @inlineCallbacks
00093     def test_replicate(self):
        """Replicate a service is correct."""
        replications = yield replication_client.get_replications()
        rid = list(replications)[0]
        yield replication_client.exclude(rid)

        yield replication_client.replicate(rid)
        exclusions = yield replication_client.get_exclusions()
        self.assertNotIn(rid, exclusions)

    @inlineCallbacks
00104     def test_replicate_name_not_in_replications(self):
        """Replicate a service fails if not in replications."""
        replications = yield replication_client.get_replications()
        rid = 'not in replications'
        assert rid not in replications

        yield self.assertFailure(replication_client.replicate(rid),
                                 replication_client.InvalidIdError)

    @inlineCallbacks
00114     def test_replicate_name_not_in_exclusions(self):
        """Replicate a service fails if not in exclusions."""
        replications = yield replication_client.get_replications()
        rid = list(replications)[0]
        assert rid in replications

        exclusions = yield replication_client.get_exclusions()
        assert rid not in exclusions

        yield self.assertFailure(replication_client.replicate(rid),
                                 replication_client.NotExcludedError)

    @inlineCallbacks
00127     def test_exclude(self):
        """Excluding a service is correct."""
        replications = yield replication_client.get_replications()
        rid = list(replications)[0]
        yield replication_client.exclude(rid)
        yield replication_client.replicate(rid)

        yield replication_client.exclude(rid)
        exclusions = yield replication_client.get_exclusions()
        self.assertIn(rid, exclusions)

    @inlineCallbacks
00139     def test_exclude_name_not_in_replications(self):
        """Excluding a service fails if not in replications."""
        replications = yield replication_client.get_replications()
        rid = 'not in replications'
        assert rid not in replications

        yield self.assertFailure(replication_client.exclude(rid),
                                 replication_client.InvalidIdError)

    @inlineCallbacks
00149     def test_exclude_name_in_exclusions(self):
        """Excluding a service fails if already on exclusions."""
        replications = yield replication_client.get_replications()
        rid = list(replications)[0]
        assert rid in replications

        yield replication_client.exclude(rid)
        exclusions = yield replication_client.get_exclusions()
        assert rid in exclusions

        yield self.assertFailure(replication_client.exclude(rid),
                                 replication_client.AlreadyExcludedError)

Generated by  Doxygen 1.6.0   Back to index