~diegosarmentero/ubuntuone-control-panel/tab-shares-functions

# -*- coding: utf-8 -*-
#
# Copyright 2010-2012 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 control panel backend."""

import operator
import os

from collections import defaultdict

import simplejson

from twisted.internet import defer
from twisted.internet.defer import inlineCallbacks, returnValue
from ubuntuone.devtools.handlers import MementoHandler

from ubuntuone.controlpanel import backend, replication_client
from ubuntuone.controlpanel.backend import (
    ACCOUNT_API, DEVICES_API, DEVICE_REMOVE_API, QUOTA_API,
    DEVICE_TYPE_COMPUTER,
    FILE_SYNC_DISABLED,
    FILE_SYNC_DISCONNECTED,
    FILE_SYNC_ERROR,
    FILE_SYNC_IDLE,
    FILE_SYNC_STARTING,
    FILE_SYNC_STOPPED,
    FILE_SYNC_SYNCING,
    FILE_SYNC_UNKNOWN,
    MSG_KEY, STATUS_KEY,
    UBUNTUONE_FROM_OAUTH,
    UBUNTUONE_LINK,
)
from ubuntuone.controlpanel.tests import (TestCase,
    EMPTY_DESCRIPTION_JSON,
    EXPECTED_ACCOUNT_INFO,
    EXPECTED_ACCOUNT_INFO_WITH_CURRENT_PLAN,
    EXPECTED_DEVICE_NAMES_INFO,
    EXPECTED_DEVICES_INFO,
    LOCAL_DEVICE,
    ROOT_PATH,
    SAMPLE_ACCOUNT_NO_CURRENT_PLAN,
    SAMPLE_ACCOUNT_WITH_CURRENT_PLAN,
    SAMPLE_DEVICES_JSON,
    SAMPLE_FOLDERS,
    SAMPLE_QUOTA_JSON,
    SAMPLE_SHARED,
    SAMPLE_SHARES,
    SHARES_PATH,
    SHARES_PATH_LINK,
    TOKEN,
    USER_HOME,
)

SAMPLE_SIGNED = UBUNTUONE_FROM_OAUTH + '?oauth_nonce=' \
    '36886134&oauth_timestamp=1310671062&oauth_consumer_key=consumer_key&' \
    'oauth_signature_method=HMAC-SHA1&next=%2Fblah&oauth_version=1.0&' \
    'oauth_token=GkInOfSMGwTXAUoVQwLUoPxElEEUdhsLVNTPhxHJDUIeHCPNEo&' \
    'oauth_signature=s6h0LRBiWchTADrTJWaJUSuaGpo%3D'


# pylint: disable=E1101, W0201, W0212


class CallRecorder(object):
    """A class that records every call clients made to it."""

    def __init__(self):
        self._called = defaultdict(int)

    def __getattribute__(self, attr_name):
        """Override to we can record calls to members."""
        result = super(CallRecorder, self).__getattribute__(attr_name)
        if attr_name != '_called':
            called = super(CallRecorder, self).__getattribute__('_called')
            called[attr_name] += 1
        return result


class MockWebClient(CallRecorder):
    """A mock webclient."""

    def __init__(self, get_credentials):
        """Initialize this mock instance."""
        super(MockWebClient, self).__init__()
        self.get_credentials = get_credentials
        self.failure = False
        self.results = {}

    def call_api(self, method):
        """Get a given url from the webservice."""
        if self.failure == 401:
            return defer.fail(backend.UnauthorizedError(self.failure))
        elif self.failure:
            return defer.fail(backend.WebClientError(self.failure))
        else:
            result = simplejson.loads(self.results[method])
            return defer.succeed(result)

    @defer.inlineCallbacks
    def build_signed_iri(self, iri, params):
        """Fake the IRI signing."""
        creds = yield self.get_credentials()
        result = u'%s-%s-%s' % (iri, unicode(creds), unicode(params))
        defer.returnValue(result)


class MockLoginClient(CallRecorder):
    """A mock login_client module."""

    def __init__(self):
        """Initialize this mock instance."""
        super(MockLoginClient, self).__init__()
        self.kwargs = None
        self.creds = TOKEN

    def find_credentials(self):
        """Return the mock credentials."""
        return defer.succeed(self.creds)

    def clear_credentials(self):
        """Clear the mock credentials."""
        self.creds = {}
        return defer.succeed(None)

    def login_email_password(self, **kwargs):
        """Simulate a login."""
        self.kwargs = kwargs
        self.creds = TOKEN
        return defer.succeed(self.creds)


class MockSDClient(CallRecorder):
    """A mock sd_client module."""

    def __init__(self):
        """Initialize this mock instance."""
        super(MockSDClient, self).__init__()
        self.throttling = False
        self.show_all_notifications = True
        self.autoconnect = True
        self.udf_autosubscribe = False
        self.share_autosubscribe = False
        self.limits = {"download": -1, "upload": -1}
        self.file_sync = True
        self.status = {
            'name': 'TEST', 'queues': 'GORGEOUS', 'connection': '',
            'description': 'Some test state, nothing else.',
            'is_error': '', 'is_connected': 'True', 'is_online': '',
        }
        self.status_changed_handler = None
        self.public_files_list_handler = None
        self.public_access_changed_handler = None
        self.subscribed_folders = []
        self.subscribed_shares = []
        self.actions = []
        self.shares = []
        self.folders = []
        self.volumes_refreshed = False
        self.menu_data = {'recent-transfers': (), 'uploading': ()}
        self.publicfiles = [
            {'path': '/home/file1', 'public_url': 'http:ubuntuone.com/asd123'},
            {'path': '/home/file2', 'public_url': 'http:ubuntuone.com/qwe456'},
        ]
        self.public_access_info = {
            'public_url': 'http:ubuntuone.com/zxc789'
        }

    def get_throttling_limits(self):
        """Return the sample speed limits."""
        return self.limits

    def set_throttling_limits(self, limits):
        """Set the sample speed limits."""
        self.limits["download"] = int(limits["download"])
        self.limits["upload"] = int(limits["upload"])

    def bandwidth_throttling_enabled(self):
        """Return the state of throttling."""
        return self.throttling

    def enable_bandwidth_throttling(self):
        """Enable bw throttling."""
        self.throttling = True

    def disable_bandwidth_throttling(self):
        """Disable bw throttling."""
        self.throttling = False

    def show_all_notifications_enabled(self):
        """Return the state of show_all_notifications."""
        return self.show_all_notifications

    def enable_show_all_notifications(self):
        """Enable show_all_notifications."""
        self.show_all_notifications = True

    def disable_show_all_notifications(self):
        """Disable show_all_notifications."""
        self.show_all_notifications = False

    def autoconnect_enabled(self):
        """Return the state of autoconnect."""
        return self.autoconnect

    def enable_autoconnect(self):
        """Enable autoconnect."""
        self.autoconnect = True

    def disable_autoconnect(self):
        """Disable autoconnect."""
        self.autoconnect = False

    def udf_autosubscribe_enabled(self):
        """Return the state of udf_autosubscribe."""
        return self.udf_autosubscribe

    def enable_udf_autosubscribe(self):
        """Enable udf_autosubscribe."""
        self.udf_autosubscribe = True

    def disable_udf_autosubscribe(self):
        """Disable udf_autosubscribe."""
        self.udf_autosubscribe = False

    def share_autosubscribe_enabled(self):
        """Return the state of share_autosubscribe."""
        return self.share_autosubscribe

    def enable_share_autosubscribe(self):
        """Enable share_autosubscribe."""
        self.share_autosubscribe = True

    def disable_share_autosubscribe(self):
        """Disable share_autosubscribe."""
        self.share_autosubscribe = False

    def files_sync_enabled(self):
        """Get if file sync service is enabled."""
        return self.file_sync

    def set_files_sync_enabled(self, enabled):
        """Set the file sync service to be 'enabled'."""
        self.file_sync = enabled

    def connect_file_sync(self):
        """Connect files service."""
        self.actions.append('connect')

    def disconnect_file_sync(self):
        """Disconnect file_sync service."""
        self.actions.append('disconnect')

    def start_file_sync(self):
        """Start the file_sync service."""
        self.actions.append('start')

    def stop_file_sync(self):
        """Stop the file_sync service."""
        self.actions.append('stop')

    def get_home_dir(self):
        """Grab the home dir."""
        return USER_HOME

    def get_root_dir(self):
        """Grab the root dir."""
        return ROOT_PATH

    def get_shares_dir(self):
        """Grab the shares dir."""
        return SHARES_PATH

    def get_shares_dir_link(self):
        """Grab the shares dir."""
        return SHARES_PATH_LINK

    def get_folders(self):
        """Grab list of folders."""
        return self.folders

    def subscribe_folder(self, volume_id):
        """Subcribe to 'volume_id'."""
        self.subscribed_folders.append(volume_id)

    def unsubscribe_folder(self, volume_id):
        """Unsubcribe from 'volume_id'."""
        self.subscribed_folders.remove(volume_id)

    def validate_path(self, path):
        """Validate a path for folder creation."""
        return path != USER_HOME

    def create_folder(self, path):
        """Grab list of folders."""
        self.folders.append(path)

    def get_shares(self):
        """Grab list of shares."""
        return self.shares

    def subscribe_share(self, volume_id):
        """Subcribe to 'volume_id'."""
        self.subscribed_shares.append(volume_id)

    def unsubscribe_share(self, volume_id):
        """Unsubcribe from 'volume_id'."""
        self.subscribed_shares.remove(volume_id)

    def get_current_status(self):
        """Grab syncdaemon status."""
        return self.status

    def set_status_changed_handler(self, handler):
        """Connect a handler for tracking syncdaemon status changes."""
        self.status_changed_handler = handler

    def get_shared(self):
        """Grab list of shared (shares from the user to others)."""
        return SAMPLE_SHARED

    def refresh_volumes(self):
        """Refresh the volume list."""
        self.volumes_refreshed = True
        return defer.succeed(None)

    def sync_menu(self):
        """Return the sync menu data."""
        return self.menu_data

    def get_public_files(self):
        """Trigger the action to get the public files."""
        if self.public_files_list_handler is not None:
            self.public_files_list_handler(self.publicfiles)

    def set_public_files_list_handler(self, handler):
        """Return the handler to be called for the public files list."""
        self.public_files_list_handler = handler

    def change_public_access(self, path, is_public):
        """Change the type access of a file."""
        if self.public_access_changed_handler is not None and is_public:
            self.public_access_changed_handler(self.public_access_info)

    def set_public_access_changed_handler(self, handler):
        """Return the handler to be called when a access type change."""
        self.public_access_changed_handler = handler


class MockReplicationClient(CallRecorder):
    """A mock replication_client module."""

    CONTACTS = 'legendary'

    replications = set([CONTACTS, 'other'])
    exclusions = set([CONTACTS])

    def get_replications(self):
        """Grab the list of replications in this machine."""
        return MockReplicationClient.replications

    def get_exclusions(self):
        """Grab the list of exclusions in this machine."""
        return MockReplicationClient.exclusions

    def replicate(self, replication_id):
        """Remove replication_id from the exclusions list."""
        if replication_id not in MockReplicationClient.replications:
            raise replication_client.ReplicationError(replication_id)
        MockReplicationClient.exclusions.remove(replication_id)

    def exclude(self, replication_id):
        """Add replication_id to the exclusions list."""
        if replication_id not in MockReplicationClient.replications:
            raise replication_client.ReplicationError(replication_id)
        MockReplicationClient.exclusions.add(replication_id)


def fail(*args, **kwargs):
    """Helper to raise any error."""
    raise ValueError(args)


class BackendBasicTestCase(TestCase):
    """Simple tests for the backend."""

    timeout = 3

    @defer.inlineCallbacks
    def setUp(self):
        yield super(BackendBasicTestCase, self).setUp()
        self.patch(backend, "WebClient", MockWebClient)
        self.patch(backend, "CredentialsManagementTool", MockLoginClient)
        self.patch(backend.sd_client, "SyncDaemonClient", MockSDClient)
        self.patch(backend, "replication_client", MockReplicationClient())

        self.local_token = DEVICE_TYPE_COMPUTER + TOKEN["token"]
        self.be = backend.ControlBackend()
        self.be.wc.failure = False

        self.memento = MementoHandler()
        backend.logger.addHandler(self.memento)

    def test_backend_creation(self):
        """The backend instance is successfully created."""
        self.assertIsInstance(self.be.wc, MockWebClient)
        self.assertIsInstance(self.be.login_client, MockLoginClient)
        self.assertIsInstance(self.be.sd_client, MockSDClient)

    @inlineCallbacks
    def test_get_token(self):
        """The get_token method returns the right token."""
        self.patch(self.be, 'get_credentials', lambda: defer.succeed(TOKEN))
        token = yield self.be.get_token()
        self.assertEqual(token, TOKEN["token"])

    @inlineCallbacks
    def test_device_is_local(self):
        """The device_is_local returns the right result."""
        self.patch(self.be, 'get_credentials', lambda: defer.succeed(TOKEN))
        result = yield self.be.device_is_local(self.local_token)
        self.assertTrue(result)

        did = EXPECTED_DEVICES_INFO[-1]['device_id']
        result = yield self.be.device_is_local(did)
        self.assertFalse(result)

    def test_shutdown_func(self):
        """A shutdown_func can be passed as creation parameter."""
        f = lambda: None
        be = backend.ControlBackend(shutdown_func=f)
        self.assertEqual(be.shutdown_func, f)

    def test_shutdown_func_is_called_on_shutdown(self):
        """The shutdown_func is called on shutdown."""
        self.be.shutdown_func = self._set_called
        self.be.shutdown()
        self.assertEqual(self._called, ((), {}))

    def test_shutdown_func_when_none(self):
        """The shutdown_func can be None."""
        self.be.shutdown_func = None
        self.be.shutdown()
        # nothing explodes

    def test_login_client_is_cached(self):
        """The login_client instance is cached."""
        self.assertIs(self.be.login_client, self.be.login_client)

    def test_sd_client_is_cached(self):
        """The sd_client instance is cached."""
        self.assertIs(self.be.sd_client, self.be.sd_client)


class SignIriTestCase(BackendBasicTestCase):
    """Test cases for the IRI signing function."""

    @defer.inlineCallbacks
    def setUp(self):
        yield super(SignIriTestCase, self).setUp()
        self.patch(self.be, 'get_credentials', lambda: defer.succeed(TOKEN))

    @inlineCallbacks
    def test_without_ubuntuone_prefix(self):
        """If given url is not an ubuntuone url, don't sign it."""
        iri = 'bad_prefix' + UBUNTUONE_LINK
        result = yield self.be.build_signed_iri(iri)

        self.assertEqual(result, iri)

    @inlineCallbacks
    def test_with_ubuntuone_prefix(self):
        """If given url is an ubuntuone url, sign it."""
        iri = UBUNTUONE_LINK + 'foo'
        result = yield self.be.build_signed_iri(iri)

        expected = yield self.be.wc.build_signed_iri(UBUNTUONE_FROM_OAUTH,
                                                     {'next': iri})
        self.assertEqual(expected, result)


class SignIriNoCredsTestCase(SignIriTestCase):
    """The test suite for the sign url management when there are no creds."""

    @defer.inlineCallbacks
    def setUp(self):
        yield super(SignIriNoCredsTestCase, self).setUp()
        self.patch(self.be, 'get_credentials', lambda: defer.succeed({}))

    @inlineCallbacks
    def test_with_ubuntuone_prefix(self):
        """If given url is an ubuntuone url, don't sign it.

        Since we have no credentials, the url should not be signed.

        """
        iri = UBUNTUONE_LINK + 'foo'
        result = yield self.be.build_signed_iri(iri)

        self.assertEqual(result, iri)


class BackendCredentialsTestCase(BackendBasicTestCase):
    """Credentials tests for the backend."""

    @inlineCallbacks
    def test_credentials_are_cached(self):
        """The credentials are cached."""
        creds1 = yield self.be.get_credentials()
        creds2 = yield self.be.get_credentials()
        self.assertEqual(creds1, creds2)
        self.assertEqual(self.be.login_client._called['find_credentials'], 1)

    @inlineCallbacks
    def test_clear_credentials(self):
        """The credentials can be cleared."""
        # ensure we have creds
        result = yield self.be.login(email='yadda', password='doo')
        assert self.be.login_client.creds == result

        yield self.be.clear_credentials()
        self.assertEqual(self.be.login_client.creds, {})

    @inlineCallbacks
    def test_clear_credentials_invalidates_cached_credentials(self):
        """The cached credentials are cleared."""
        # ensure we have creds
        result = yield self.be.login(email='yadda', password='doo')
        assert self.be.login_client.creds == result

        yield self.be.clear_credentials()

        creds = yield self.be.get_credentials()
        self.assertEqual(creds, {})

    @inlineCallbacks
    def test_login(self):
        """Login can be called."""
        kwargs = dict(email='yadda', password='doo')
        result = yield self.be.login(**kwargs)

        self.assertEqual(self.be.login_client.kwargs, kwargs)
        self.assertEqual(result, TOKEN)

    @inlineCallbacks
    def test_login_caches_credentials(self):
        """Login catches credentials when available."""
        yield self.be.clear_credentials()

        result = yield self.be.login(email='yadda', password='doo')
        creds = yield self.be.get_credentials()

        self.assertEqual(result, TOKEN)
        self.assertEqual(result, creds)
        self.assertEqual(self.be.login_client._called['find_credentials'], 0)

    @inlineCallbacks
    def test_login_fails(self):
        """Login failure is emitted."""
        self.patch(self.be.login_client, 'login_email_password', fail)
        yield self.assertFailure(self.be.login('foo', 'bar'), ValueError)


class BackendAccountTestCase(BackendBasicTestCase):
    """Account tests for the backend."""

    @inlineCallbacks
    def test_account_info(self):
        """The account_info method exercises its callback."""
        self.be.wc.results[ACCOUNT_API] = SAMPLE_ACCOUNT_NO_CURRENT_PLAN
        self.be.wc.results[QUOTA_API] = SAMPLE_QUOTA_JSON
        result = yield self.be.account_info()
        self.assertEqual(result, EXPECTED_ACCOUNT_INFO)

    @inlineCallbacks
    def test_account_info_with_current_plan(self):
        """The account_info method exercises its callback."""
        self.be.wc.results[ACCOUNT_API] = SAMPLE_ACCOUNT_WITH_CURRENT_PLAN
        self.be.wc.results[QUOTA_API] = SAMPLE_QUOTA_JSON
        result = yield self.be.account_info()
        self.assertEqual(result, EXPECTED_ACCOUNT_INFO_WITH_CURRENT_PLAN)

    @inlineCallbacks
    def test_account_info_fails(self):
        """The account_info method exercises its errback."""
        self.be.wc.failure = 404
        yield self.assertFailure(self.be.account_info(),
                                 backend.WebClientError)

    @inlineCallbacks
    def test_account_info_fails_with_unauthorized(self):
        """The account_info clears the credentials on unauthorized."""
        self.be.wc.failure = 401
        d = defer.Deferred()
        self.patch(self.be, 'clear_credentials', lambda: d.callback('called'))
        yield self.assertFailure(self.be.account_info(),
                                 backend.UnauthorizedError)
        yield d


class BackendDevicesTestCase(BackendBasicTestCase):
    """Devices tests for the backend."""

    @inlineCallbacks
    def test_devices_info(self):
        """The devices_info method exercises its callback."""
        self.be.wc.results[DEVICES_API] = SAMPLE_DEVICES_JSON
        result = yield self.be.devices_info()
        self.assertEqual(result, EXPECTED_DEVICES_INFO)

    @inlineCallbacks
    def test_devices_info_fails(self):
        """The devices_info method exercises its errback."""
        self.patch(self.be.wc, 'call_api', fail)
        yield self.assertFailure(self.be.devices_info(), ValueError)

    @inlineCallbacks
    def test_devices_info_with_webclient_error(self):
        """The devices_info returns local info if webclient error."""
        self.be.wc.failure = 404
        result = yield self.be.devices_info()

        self.assertEqual(result, [LOCAL_DEVICE])
        self.assertTrue(self.memento.check_error('devices_info',
                                                 'web client failure'))

    @inlineCallbacks
    def test_devices_info_fails_with_unauthorized(self):
        """The devices_info clears the credentials on unauthorized."""
        self.be.wc.failure = 401
        d = defer.Deferred()
        self.patch(self.be, 'clear_credentials', lambda: d.callback('called'))
        yield self.assertFailure(self.be.devices_info(),
                                 backend.UnauthorizedError)
        yield d

    @inlineCallbacks
    def test_devices_info_if_files_disabled(self):
        """The devices_info returns device only info if files is disabled."""
        yield self.be.disable_files()
        status = yield self.be.file_sync_status()
        assert status['status'] == backend.FILE_SYNC_DISABLED

        self.be.wc.results[DEVICES_API] = SAMPLE_DEVICES_JSON
        result = yield self.be.devices_info()

        expected = []
        for device in EXPECTED_DEVICES_INFO:
            device = device.copy()
            device.pop('limit_bandwidth', None)
            device.pop(backend.DOWNLOAD_KEY, None)
            device.pop(backend.UPLOAD_KEY, None)
            device.pop(backend.AUTOCONNECT_KEY, None)
            device.pop(backend.SHOW_ALL_NOTIFICATIONS_KEY, None)
            device.pop(backend.SHARE_AUTOSUBSCRIBE_KEY, None)
            device.pop(backend.UDF_AUTOSUBSCRIBE_KEY, None)
            device['configurable'] = False
            expected.append(device)
        self.assertEqual(result, expected)

    @inlineCallbacks
    def test_devices_info_when_token_name_is_empty(self):
        """The devices_info can handle empty token names."""
        self.be.wc.results[DEVICES_API] = EMPTY_DESCRIPTION_JSON
        result = yield self.be.devices_info()
        expected = {'configurable': False,
                    'device_id': 'ComputerABCDEF01234token',
                    'is_local': False, 'name': self.be.NAME_NOT_SET,
                    'type': DEVICE_TYPE_COMPUTER}
        self.assertEqual(result, [expected])

    @inlineCallbacks
    def test_devices_info_does_not_log_device_id(self):
        """The devices_info does not log the device_id."""
        self.be.wc.results[DEVICES_API] = SAMPLE_DEVICES_JSON
        yield self.be.devices_info()

        dids = (d['device_id'] for d in EXPECTED_DEVICES_INFO)
        device_id_logged = all(all(did not in r.getMessage()
                                   for r in self.memento.records)
                               for did in dids)
        self.assertTrue(device_id_logged)

    @inlineCallbacks
    def test_device_names_info(self):
        """The device_names_info returns the correct info."""
        self.be.wc.results[DEVICES_API] = SAMPLE_DEVICES_JSON
        result = yield self.be.device_names_info()
        self.assertEqual(result, EXPECTED_DEVICE_NAMES_INFO)

    @inlineCallbacks
    def test_device_names_info_fails(self):
        """The device_names_info method exercises its errback."""
        self.patch(self.be.wc, 'call_api', fail)
        yield self.assertFailure(self.be.device_names_info(), ValueError)

    @inlineCallbacks
    def test_device_names_info_with_webclient_error(self):
        """The device_names_info returns local info if webclient error."""
        self.be.wc.failure = 404
        result = yield self.be.device_names_info()

        device = LOCAL_DEVICE.copy()
        device.pop('configurable', None)
        device.pop('limit_bandwidth', None)
        device.pop(backend.DOWNLOAD_KEY, None)
        device.pop(backend.UPLOAD_KEY, None)
        device.pop(backend.AUTOCONNECT_KEY, None)
        device.pop(backend.SHOW_ALL_NOTIFICATIONS_KEY, None)
        device.pop(backend.SHARE_AUTOSUBSCRIBE_KEY, None)
        device.pop(backend.UDF_AUTOSUBSCRIBE_KEY, None)
        self.assertEqual(result, [device])
        self.assertTrue(self.memento.check_error('device_names_info',
                                                 'web client failure'))

    @inlineCallbacks
    def test_device_names_info_fails_with_unauthorized(self):
        """The device_names_info clears the credentials on unauthorized."""
        self.be.wc.failure = 401
        d = defer.Deferred()
        self.patch(self.be, 'clear_credentials', lambda: d.callback('called'))
        yield self.assertFailure(self.be.device_names_info(),
                                 backend.UnauthorizedError)
        yield d

    @inlineCallbacks
    def test_device_names_info_when_token_name_is_empty(self):
        """The device_names_info can handle empty token names."""
        self.be.wc.results[DEVICES_API] = EMPTY_DESCRIPTION_JSON
        result = yield self.be.device_names_info()
        expected = {'device_id': 'ComputerABCDEF01234token',
                    'is_local': False, 'name': self.be.NAME_NOT_SET,
                    'type': DEVICE_TYPE_COMPUTER}
        self.assertEqual(result, [expected])

    @inlineCallbacks
    def test_device_names_info_does_not_log_device_id(self):
        """The device_names_info does not log the device_id."""
        self.be.wc.results[DEVICES_API] = SAMPLE_DEVICES_JSON
        yield self.be.device_names_info()

        dids = (d['device_id'] for d in EXPECTED_DEVICES_INFO)
        device_id_logged = all(all(did not in r.getMessage()
                                   for r in self.memento.records)
                               for did in dids)
        self.assertTrue(device_id_logged)

    @inlineCallbacks
    def test_remove_device(self):
        """The remove_device method calls the right api."""
        self.patch(self.be, 'clear_credentials', self._set_called)

        dtype, did = DEVICE_TYPE_COMPUTER, "SAMPLE-TOKEN"
        device_id = dtype + did
        apiurl = DEVICE_REMOVE_API % (dtype.lower(), did)
        self.be.wc.results[apiurl] = SAMPLE_DEVICES_JSON
        result = yield self.be.remove_device(device_id)
        self.assertEqual(result, device_id)

        # credentials were not cleared
        self.assertFalse(self._called)

    @inlineCallbacks
    def test_remove_device_clear_credentials_if_local_device(self):
        """The remove_device method clears the credentials if is local."""
        self.patch(self.be, 'clear_credentials', self._set_called)

        apiurl = DEVICE_REMOVE_API % ('computer', TOKEN['token'])
        self.be.wc.results[apiurl] = SAMPLE_DEVICES_JSON
        yield self.be.remove_device(self.local_token)

        # credentials were cleared
        self.assertEqual(self._called, ((), {}))

    @inlineCallbacks
    def test_remove_device_fails(self):
        """The remove_device method fails as expected."""
        self.be.wc.failure = 404
        yield self.assertFailure(self.be.remove_device(self.local_token),
                                 backend.WebClientError)

    @inlineCallbacks
    def test_remove_device_fails_with_unauthorized(self):
        """The remove_device clears the credentials on unauthorized."""
        self.be.wc.failure = 401
        d = defer.Deferred()
        self.patch(self.be, 'clear_credentials', lambda: d.callback('called'))
        yield self.assertFailure(self.be.remove_device(self.local_token),
                                 backend.UnauthorizedError)
        yield d

    @inlineCallbacks
    def test_remove_device_does_not_log_device_id(self):
        """The remove_device does not log the device_id."""
        device_id = DEVICE_TYPE_COMPUTER + TOKEN['token']
        apiurl = DEVICE_REMOVE_API % ('computer', TOKEN['token'])
        self.be.wc.results[apiurl] = SAMPLE_DEVICES_JSON
        yield self.be.remove_device(device_id)

        device_id_logged = all(device_id not in r.getMessage()
                               for r in self.memento.records)
        self.assertTrue(device_id_logged)

    @inlineCallbacks
    def test_change_show_all_notifications(self):
        """The device settings are updated."""
        self.be.sd_client.show_all_notifications = False
        yield self.be.change_device_settings(self.local_token,
                                    {backend.SHOW_ALL_NOTIFICATIONS_KEY: True})
        self.assertEqual(self.be.sd_client.show_all_notifications, True)
        yield self.be.change_device_settings(self.local_token,
                                   {backend.SHOW_ALL_NOTIFICATIONS_KEY: False})
        self.assertEqual(self.be.sd_client.show_all_notifications, False)

    @inlineCallbacks
    def test_change_limit_bandwidth(self):
        """The device settings are updated."""
        self.be.sd_client.throttling = False
        yield self.be.change_device_settings(self.local_token,
                                        {backend.LIMIT_BW_KEY: True})
        self.assertEqual(self.be.sd_client.throttling, True)
        yield self.be.change_device_settings(self.local_token,
                                        {backend.LIMIT_BW_KEY: False})
        self.assertEqual(self.be.sd_client.throttling, False)

    @inlineCallbacks
    def test_change_upload_speed_limit(self):
        """The device settings are updated."""
        self.be.sd_client.limits = {"download": -1, "upload": -1}
        yield self.be.change_device_settings(self.local_token,
                                        {backend.UPLOAD_KEY: 1111})
        self.assertEqual(self.be.sd_client.limits["upload"], 1111)
        self.assertEqual(self.be.sd_client.limits["download"], -1)

    @inlineCallbacks
    def test_change_download_speed_limit(self):
        """The device settings are updated."""
        self.be.sd_client.limits = {"download": -1, "upload": -1}
        yield self.be.change_device_settings(self.local_token,
                                        {backend.DOWNLOAD_KEY: 99})
        self.assertEqual(self.be.sd_client.limits["upload"], -1)
        self.assertEqual(self.be.sd_client.limits["download"], 99)

    @inlineCallbacks
    def test_changing_settings_for_wrong_id_has_no_effect(self):
        """If the id is wrong, no settings are changed."""
        self.be.sd_client.throttling = False
        self.be.sd_client.limits = {"download": -1, "upload": -1}
        new_settings = {
            backend.DOWNLOAD_KEY: 99,
            backend.UPLOAD_KEY: 99,
            backend.LIMIT_BW_KEY: True,
        }
        yield self.be.change_device_settings("wrong token!", new_settings)
        self.assertEqual(self.be.sd_client.throttling, False)
        self.assertEqual(self.be.sd_client.limits["upload"], -1)
        self.assertEqual(self.be.sd_client.limits["download"], -1)

    @inlineCallbacks
    def test_changing_settings_does_not_log_device_id(self):
        """The change_device_settings does not log the device_id."""
        device_id = 'yadda-yadda'
        yield self.be.change_device_settings(device_id, {})

        device_id_logged = all(device_id not in r.getMessage()
                               for r in self.memento.records)
        self.assertTrue(device_id_logged)


class BackendVolumesTestCase(BackendBasicTestCase):
    """Volumes tests for the backend."""

    # Access to a protected member of a client class

    @defer.inlineCallbacks
    def setUp(self):
        yield super(BackendVolumesTestCase, self).setUp()
        # fake quota info and calculate free bytes
        self.be.wc.results[ACCOUNT_API] = SAMPLE_ACCOUNT_NO_CURRENT_PLAN
        self.be.wc.results[QUOTA_API] = SAMPLE_QUOTA_JSON

        # root dir info
        display_name = yield self.be._process_path(ROOT_PATH)
        self.root_volume = {
            u'volume_id': u'', u'path': ROOT_PATH, u'subscribed': True,
            u'type': self.be.ROOT_TYPE,
            u'display_name': display_name,
        }

        self.patch(self.be.sd_client, 'shares', SAMPLE_SHARES)
        self.patch(self.be.sd_client, 'folders', SAMPLE_FOLDERS)

    @inlineCallbacks
    def expected_volumes(self, sample_shares=None, sample_folders=None,
                         with_storage_info=True):
        """Get shares and group by sharing user, get folders and free space."""
        if sample_shares is None:
            sample_shares = self.be.sd_client.shares

        if sample_folders is None:
            sample_folders = self.be.sd_client.folders

        free_bytes = self.be.FREE_BYTES_NOT_AVAILABLE
        if with_storage_info:
            try:
                result = yield self.be.account_info()
            except Exception:  # pylint: disable=W0703
                pass
            else:
                free_bytes = result['quota_total'] - result['quota_used']

        # get shares and group by sharing user
        shares = defaultdict(list)
        for share in sample_shares:
            # filter out non accepted values
            if not bool(share['accepted']):
                continue

            share = share.copy()

            share[u'realpath'] = share[u'path']
            nicer_path = share[u'path'].replace(SHARES_PATH, SHARES_PATH_LINK)
            share[u'path'] = nicer_path
            share[u'type'] = self.be.SHARE_TYPE
            share[u'subscribed'] = bool(share[u'subscribed'])
            share[u'display_name'] = share[u'name']

            username = share['other_visible_name']
            if not username:
                username = share['other_username'] + \
                           ' (%s)' % self.be.NAME_NOT_SET

            shares[username].append(share)

        folders = []
        for folder in sample_folders:
            folder = folder.copy()
            folder[u'type'] = self.be.FOLDER_TYPE
            folder[u'subscribed'] = bool(folder[u'subscribed'])
            display_name = yield self.be._process_path(folder[u'path'])
            folder[u'display_name'] = display_name
            folders.append(folder)

        # sort folders by path
        folders.sort(key=operator.itemgetter('path'))
        expected = [(u'', free_bytes, [self.root_volume] + folders)]

        for other_user, data in shares.iteritems():
            send_freebytes = any(d['access_level'] == 'Modify' for d in data)
            if send_freebytes:
                free_bytes = int(data[0][u'free_bytes'])
            else:
                free_bytes = self.be.FREE_BYTES_NOT_AVAILABLE

            # sort data by path
            data.sort(key=operator.itemgetter('path'))

            expected.append((other_user, free_bytes, data))

        returnValue(expected)

    @inlineCallbacks
    def test_volumes_info(self):
        """The volumes_info method exercises its callback."""
        expected = yield self.expected_volumes()
        result = yield self.be.volumes_info()

        self.assertEqual(result, expected)
        self.assertFalse(self.be.sd_client.volumes_refreshed)

    @inlineCallbacks
    def test_volumes_info_can_refresh_volumes(self):
        """The volumes_info can be refreshed."""
        expected = yield self.expected_volumes()
        result = yield self.be.volumes_info(refresh=True)

        self.assertTrue(self.be.sd_client.volumes_refreshed)
        self.assertEqual(result, expected)

    @inlineCallbacks
    def test_volumes_info_process_path(self):
        """The volumes_info method exercises its callback."""
        root_path = USER_HOME + os.path.sep + 'My Ubuntu' + USER_HOME
        self.patch(self.be.sd_client, 'get_root_dir', lambda: root_path)
        for item in SAMPLE_FOLDERS:
            path = item[u'path']
            path = path[len(USER_HOME) + 1:]
            item[u'path'] = os.path.join(root_path, path)
        self.patch(self.be.sd_client, 'shares', SAMPLE_SHARES)

        yield self.be.volumes_info()
        for item in SAMPLE_FOLDERS:
            key = item[u'volume_id']
            folder = self.be._volumes[key]
            display_name = folder['display_name']
            prefix = 'My Ubuntu' + USER_HOME
            self.assertTrue(display_name.startswith(prefix))

    @inlineCallbacks
    def test_volumes_info_without_storage_info(self):
        """The volumes_info method exercises its callback."""
        expected = yield self.expected_volumes(with_storage_info=False)
        result = yield self.be.volumes_info(with_storage_info=False)
        self.assertEqual(result, expected)

    def test_cached_volumes_are_initially_empty(self):
        """The cached volume list is empty."""
        self.assertEqual(self.be._volumes, {})

    @inlineCallbacks
    def test_volumes_are_cached(self):
        """The volume list is cached."""
        expected = {}
        info = yield self.expected_volumes()
        for _, _, data in info:
            for volume in data:
                sid = volume['volume_id']
                assert sid not in expected
                expected[sid] = volume

        _ = yield self.be.volumes_info()

        self.assertEqual(len(self.be._volumes), len(expected))
        self.assertEqual(self.be._volumes, expected)

    @inlineCallbacks
    def test_cached_volumes_are_updated_with_volume_info(self):
        """The cached volume list is updated."""
        yield self.test_volumes_are_cached()
        yield self.test_volumes_are_cached()

    @inlineCallbacks
    def test_volumes_info_no_quota_for_read_onlys(self):
        """The volumes_info does not send qupota info for RO shares."""
        read_only_shares = [
            {u'accepted': u'True',
             u'subscribed': u'True',
             u'access_level': u'View',
             u'free_bytes': u'39892622746',
             u'generation': u'2704',
             u'name': u're',
             u'node_id': u'c483f419-ed28-490a-825d-a8c074e2d795',
             u'other_username': u'otheruser',
             u'other_visible_name': u'Other User',
             u'path': os.path.join(SHARES_PATH, 're from Other User'),
             u'type': u'Share',
             u'volume_id': u'4a1b263b-a2b3-4f66-9e66-4cd18050810d'},
            {u'accepted': u'True',
             u'subscribed': u'True',
             u'access_level': u'View',
             u'free_bytes': u'39892622746',
             u'generation': u'2704',
             u'name': u'do',
             u'node_id': u'84544ea4-aefe-4f91-9bb9-ed7b0a805baf',
             u'other_username': u'otheruser',
             u'other_visible_name': u'Other User',
             u'path': os.path.join(SHARES_PATH, 'do from Other User'),
             u'type': u'Share',
             u'volume_id': u'7d130dfe-98b2-4bd5-8708-9eeba9838ac0'},
        ]

        self.patch(self.be.sd_client, 'shares', read_only_shares)

        expected = yield self.expected_volumes()
        result = yield self.be.volumes_info()
        self.assertEqual(result, expected)

    @inlineCallbacks
    def test_volumes_info_no_quota_for_root(self):
        """The volumes_info returns info even if quota call fails."""
        self.be.wc.failure = 500

        expected = yield self.expected_volumes()
        result = yield self.be.volumes_info()

        self.assertEqual(len(result), len(expected))
        self.assertEqual(result, expected)

    @inlineCallbacks
    def test_subscribe_volume_folder(self):
        """The subscribe_volume method subscribes a folder."""
        fid = '0123-4567'
        self.be._volumes[fid] = {u'type': self.be.FOLDER_TYPE}

        yield self.be.subscribe_volume(volume_id=fid)
        self.addCleanup(self.be.sd_client.subscribed_folders.remove, fid)

        self.assertEqual(self.be.sd_client.subscribed_folders, [fid])

    @inlineCallbacks
    def test_unsubscribe_volume_folder(self):
        """The unsubscribe_volume method unsubscribes a folder."""
        fid = '0123-4567'
        self.be._volumes[fid] = {u'type': self.be.FOLDER_TYPE}

        yield self.be.subscribe_volume(volume_id=fid)
        self.assertEqual(self.be.sd_client.subscribed_folders, [fid])

        yield self.be.unsubscribe_volume(volume_id=fid)

        self.assertEqual(self.be.sd_client.subscribed_folders, [])

    @inlineCallbacks
    def test_subscribe_volume_share(self):
        """The subscribe_volume method subscribes a share."""
        sid = '0123-4567'
        self.be._volumes[sid] = {u'type': self.be.SHARE_TYPE}

        yield self.be.subscribe_volume(volume_id=sid)
        self.addCleanup(self.be.sd_client.subscribed_shares.remove, sid)

        self.assertEqual(self.be.sd_client.subscribed_shares, [sid])

    @inlineCallbacks
    def test_unsubscribe_volume_share(self):
        """The unsubscribe_volume method unsubscribes a share."""
        sid = '0123-4567'
        self.be._volumes[sid] = {u'type': self.be.SHARE_TYPE}

        yield self.be.subscribe_volume(volume_id=sid)
        self.assertEqual(self.be.sd_client.subscribed_shares, [sid])

        yield self.be.unsubscribe_volume(volume_id=sid)

        self.assertEqual(self.be.sd_client.subscribed_shares, [])

    @inlineCallbacks
    def test_change_volume_settings_folder(self):
        """The volume settings can be changed."""
        fid = '0123-4567'
        self.be._volumes[fid] = {u'type': self.be.FOLDER_TYPE}

        yield self.be.change_volume_settings(fid, {'subscribed': True})
        self.assertEqual(self.be.sd_client.subscribed_folders, [fid])

        yield self.be.change_volume_settings(fid, {'subscribed': False})
        self.assertEqual(self.be.sd_client.subscribed_folders, [])

    @inlineCallbacks
    def test_change_volume_settings_share(self):
        """The volume settings can be changed."""
        sid = '0123-4567'
        self.be._volumes[sid] = {u'type': self.be.SHARE_TYPE}

        yield self.be.change_volume_settings(sid, {'subscribed': True})
        self.assertEqual(self.be.sd_client.subscribed_shares, [sid])

        yield self.be.change_volume_settings(sid, {'subscribed': False})
        self.assertEqual(self.be.sd_client.subscribed_shares, [])

    @inlineCallbacks
    def test_change_volume_settings_no_setting(self):
        """The change volume settings does not fail on empty settings."""
        yield self.be.change_volume_settings('test', {})
        self.assertEqual(self.be.sd_client.subscribed_folders, [])
        self.assertEqual(self.be.sd_client.subscribed_shares, [])

    @inlineCallbacks
    def test_create_folder(self):
        """New folders can be created."""
        self.patch(self.be.sd_client, 'folders', [])

        folder_path = os.path.join(USER_HOME, 'Test Me')
        yield self.be.create_folder(folder_path=folder_path)

        self.assertEqual(self.be.sd_client.folders, [folder_path])

    @defer.inlineCallbacks
    def test_validate_path_for_folder(self):
        """Test proper validation of path for creating folders."""
        folder_path = os.path.join(USER_HOME, 'Test Me')

        result = yield self.be.validate_path_for_folder(folder_path)
        self.assertTrue(result,
            '%r must be a valid path for creating a folder.' % folder_path)

    @defer.inlineCallbacks
    def test_validate_path_for_folder_invalid(self):
        """Test proper validation of path for creating folders."""
        result = yield self.be.validate_path_for_folder(USER_HOME)
        self.assertFalse(result,
            '%r must not be a valid path for creating a folder.' % USER_HOME)


class BackendSyncStatusTestCase(BackendBasicTestCase):
    """Syncdaemon state for the backend."""

    was_disabled = False

    @defer.inlineCallbacks
    def setUp(self):
        yield super(BackendSyncStatusTestCase, self).setUp()
        self.be.file_sync_disabled = self.was_disabled

    def _build_msg(self):
        """Build expected message regarding file sync status."""
        return '%s (%s)' % (self.be.sd_client.status['description'],
                            self.be.sd_client.status['name'])

    @inlineCallbacks
    def assert_correct_status(self, status, msg=None):
        """Check that the resulting status is correct."""
        expected = {MSG_KEY: self._build_msg() if msg is None else msg,
                    STATUS_KEY: status}
        result = yield self.be.file_sync_status()
        self.assertEqual(expected, result)

    @inlineCallbacks
    def test_disabled(self):
        """The syncdaemon status is processed and emitted."""
        self.patch(self.be.sd_client, 'file_sync', False)
        yield self.assert_correct_status(FILE_SYNC_DISABLED, msg='')
        self.assertTrue(self.be.file_sync_disabled)

    @inlineCallbacks
    def test_error(self):
        """The syncdaemon status is processed and emitted."""
        self.be.sd_client.status = {
            'is_error': 'True',  # nothing else matters
            'is_online': '', 'is_connected': '',
            'name': 'AUTH_FAILED', 'connection': '', 'queues': '',
            'description': 'auth failed',
        }
        yield self.assert_correct_status(FILE_SYNC_ERROR)
        # self.be.file_sync_disabled does not change
        self.assertEqual(self.was_disabled, self.be.file_sync_disabled)

    @inlineCallbacks
    def test_starting_when_init_not_user(self):
        """The syncdaemon status is processed and emitted."""
        self.be.sd_client.status = {
            'is_error': '', 'is_online': '', 'is_connected': '',
            'connection': 'Not User With Network', 'queues': '',
            'name': 'INIT', 'description': 'something new',
        }
        yield self.assert_correct_status(FILE_SYNC_STARTING)
        self.assertFalse(self.be.file_sync_disabled)

    @inlineCallbacks
    def test_starting_when_init_with_user(self):
        """The syncdaemon status is processed and emitted."""
        self.be.sd_client.status = {
            'is_error': '', 'is_online': '', 'is_connected': '',
            'connection': 'With User With Network', 'queues': '',
            'name': 'INIT', 'description': 'something new',
        }
        yield self.assert_correct_status(FILE_SYNC_STARTING)
        self.assertFalse(self.be.file_sync_disabled)

    @inlineCallbacks
    def test_starting_when_local_rescan_not_user(self):
        """The syncdaemon status is processed and emitted."""
        self.be.sd_client.status = {
            'is_error': '', 'is_online': '', 'is_connected': '',
            'connection': 'Not User With Network', 'queues': '',
            'name': 'LOCAL_RESCAN', 'description': 'something new',
        }
        yield self.assert_correct_status(FILE_SYNC_STARTING)
        self.assertFalse(self.be.file_sync_disabled)

    @inlineCallbacks
    def test_starting_when_local_rescan_with_user(self):
        """The syncdaemon status is processed and emitted."""
        self.be.sd_client.status = {
            'is_error': '', 'is_online': '', 'is_connected': '',
            'connection': 'With User With Network', 'queues': '',
            'name': 'LOCAL_RESCAN', 'description': 'something new',
        }
        yield self.assert_correct_status(FILE_SYNC_STARTING)
        self.assertFalse(self.be.file_sync_disabled)

    @inlineCallbacks
    def test_starting_when_ready_with_user(self):
        """The syncdaemon status is processed and emitted."""
        self.be.sd_client.status = {
            'is_error': '', 'is_online': '', 'is_connected': '',
            'connection': 'With User With Network', 'queues': '',
            'name': 'READY', 'description': 'something nicer',
        }
        yield self.assert_correct_status(FILE_SYNC_STARTING)
        self.assertFalse(self.be.file_sync_disabled)

    @inlineCallbacks
    def test_disconnected(self):
        """The syncdaemon status is processed and emitted."""
        self.be.sd_client.status = {
            'is_error': '', 'is_online': '', 'is_connected': '',
            'queues': '', 'description': 'something new',
            'connection': 'Not User With Network',  # user didn't connect
            'name': 'READY',  # must be READY, otherwise is STARTING
        }
        yield self.assert_correct_status(FILE_SYNC_DISCONNECTED)

        # self.be.file_sync_disabled does not change
        self.assertEqual(self.was_disabled, self.be.file_sync_disabled)

    @inlineCallbacks
    def test_disconnected_when_waiting(self):
        """The syncdaemon status is processed and emitted."""
        self.be.sd_client.status = {
            'is_error': '', 'is_online': '', 'is_connected': '',
            'connection': 'With User With Network', 'queues': '',
            'name': 'WAITING', 'description': 'what a long wait!',
        }
        yield self.assert_correct_status(FILE_SYNC_DISCONNECTED)

        # self.be.file_sync_disabled does not change
        self.assertEqual(self.was_disabled, self.be.file_sync_disabled)

    @inlineCallbacks
    def test_syncing_if_online(self):
        """The syncdaemon status is processed and emitted."""
        self.be.sd_client.status = {
            'is_error': '', 'is_online': 'True', 'is_connected': 'True',
            'name': 'QUEUE_MANAGER', 'connection': '',
            'queues': 'WORKING_ON_CONTENT',  # anything but IDLE
            'description': 'something nicer',
        }
        yield self.assert_correct_status(FILE_SYNC_SYNCING)

        # self.be.file_sync_disabled does not change
        self.assertEqual(self.was_disabled, self.be.file_sync_disabled)

    @inlineCallbacks
    def test_syncing_even_if_not_online(self):
        """The syncdaemon status is processed and emitted."""
        self.be.sd_client.status = {
            'is_error': '', 'is_online': '', 'is_connected': 'True',
            'name': 'CHECK_VERSION', 'connection': '',
            'queues': 'WORKING_ON_CONTENT',
            'description': 'something nicer',
        }
        yield self.assert_correct_status(FILE_SYNC_SYNCING)

        # self.be.file_sync_disabled does not change
        self.assertEqual(self.was_disabled, self.be.file_sync_disabled)

    @inlineCallbacks
    def test_idle(self):
        """The syncdaemon status is processed and emitted."""
        self.be.sd_client.status = {
            'is_error': '', 'is_online': 'True', 'is_connected': 'True',
            'name': 'QUEUE_MANAGER', 'connection': '',
            'queues': 'IDLE',
            'description': 'something nice',
        }
        yield self.assert_correct_status(FILE_SYNC_IDLE)

        # self.be.file_sync_disabled does not change
        self.assertEqual(self.was_disabled, self.be.file_sync_disabled)

    @inlineCallbacks
    def test_stopped(self):
        """The syncdaemon status is processed and emitted."""
        self.be.sd_client.status = {
            'is_error': '', 'is_online': '', 'is_connected': '',
            'name': 'SHUTDOWN', 'connection': '',
            'queues': 'IDLE',
            'description': 'something nice',
        }
        yield self.assert_correct_status(FILE_SYNC_STOPPED)

        # self.be.file_sync_disabled does not change
        self.assertEqual(self.was_disabled, self.be.file_sync_disabled)

    @inlineCallbacks
    def test_unknown(self):
        """The syncdaemon status is processed and emitted."""
        self.be.sd_client.status = status = {
            'is_error': '', 'is_online': '', 'is_connected': '',
            'name': '', 'connection': '', 'queues': '',
            'description': '',
        }
        yield self.assert_correct_status(FILE_SYNC_UNKNOWN)

        has_warning = self.memento.check_warning('file_sync_status: unknown',
                                                 repr(status))
        self.assertTrue(has_warning)

        # self.be.file_sync_disabled does not change
        self.assertEqual(self.was_disabled, self.be.file_sync_disabled)

    def test_status_changed(self):
        """The file_sync_status is the status changed handler."""
        self.be.status_changed_handler = self._set_called
        status = {'name': 'foo', 'description': 'bar', 'is_error': '',
                  'is_connected': '', 'is_online': '', 'queues': ''}
        self.be.sd_client.status_changed_handler(status)

        expected_status = self.be._process_file_sync_status(status)
        self.assertEqual(self._called, ((expected_status,), {}))

    def test_invalid_status_type(self):
        """Check that the method return None with invalid types."""
        self.be.status_changed_handler = self._set_called
        status = 3
        self.be.sd_client.status_changed_handler(status)

        self.assertEqual(self._called, ((None,), {}))


class BackendSyncStatusIfDisabledTestCase(BackendSyncStatusTestCase):
    """Syncdaemon state for the backend when file sync is disabled."""

    was_disabled = True

    @inlineCallbacks
    def assert_correct_status(self, status, msg=None):
        """Check that the resulting status is correct."""
        sup = super(BackendSyncStatusIfDisabledTestCase, self)
        if status != FILE_SYNC_STARTING:
            yield sup.assert_correct_status(FILE_SYNC_DISABLED, msg='')
        else:
            yield sup.assert_correct_status(status, msg=msg)


class BackendFileSyncOpsTestCase(BackendBasicTestCase):
    """Syncdaemon operations for the backend."""

    @defer.inlineCallbacks
    def setUp(self):
        yield super(BackendFileSyncOpsTestCase, self).setUp()
        self.be.sd_client.actions = []

    @inlineCallbacks
    def test_enable_files(self):
        """Files service is enabled."""
        yield self.be.disable_files()

        yield self.be.enable_files()
        self.assertTrue(self.be.sd_client.file_sync)
        self.assertFalse(self.be.file_sync_disabled)

    @inlineCallbacks
    def test_disable_files(self):
        """Files service is disabled."""
        yield self.be.enable_files()

        yield self.be.disable_files()
        self.assertFalse(self.be.sd_client.file_sync)
        self.assertTrue(self.be.file_sync_disabled)

    @inlineCallbacks
    def test_connect_files(self):
        """Connect files service."""
        yield self.be.connect_files()

        self.assertEqual(self.be.sd_client.actions, ['connect'])
        self.assertFalse(self.be.file_sync_disabled)

    @inlineCallbacks
    def test_disconnect_files(self):
        """Disconnect files service."""
        yield self.be.disconnect_files()

        self.assertEqual(self.be.sd_client.actions, ['disconnect'])
        self.assertFalse(self.be.file_sync_disabled)

    @inlineCallbacks
    def test_restart_files(self):
        """Restart the files service."""
        yield self.be.restart_files()

        self.assertEqual(self.be.sd_client.actions, ['stop', 'start'])
        self.assertFalse(self.be.file_sync_disabled)

    @inlineCallbacks
    def test_start_files(self):
        """Start the files service."""
        yield self.be.start_files()

        self.assertEqual(self.be.sd_client.actions, ['start'])
        self.assertFalse(self.be.file_sync_disabled)

    @inlineCallbacks
    def test_stop_files(self):
        """Stop the files service."""
        yield self.be.stop_files()

        self.assertEqual(self.be.sd_client.actions, ['stop'])
        self.assertFalse(self.be.file_sync_disabled)


class BackendReplicationsTestCase(BackendBasicTestCase):
    """Replications tests for the backend."""

    @inlineCallbacks
    def test_replications_info(self):
        """The replications_info method exercises its callback."""
        result = yield self.be.replications_info()

        # replications_info will use exclusions information
        expected = []
        for name in MockReplicationClient.replications:
            enabled = name not in MockReplicationClient.exclusions
            dependency = ''
            if name == MockReplicationClient.CONTACTS:
                dependency = backend.CONTACTS_PKG

            item = {'replication_id': name, 'name': name,
                    'enabled': enabled, 'dependency': dependency}
            expected.append(item)
        self.assertEqual(sorted(expected), sorted(result))

    @inlineCallbacks
    def test_change_replication_settings(self):
        """The replication settings can be changed."""
        rid = '0123-4567'
        MockReplicationClient.replications.add(rid)
        self.addCleanup(MockReplicationClient.replications.remove, rid)

        yield self.be.change_replication_settings(rid, {'enabled': False})
        self.assertIn(rid, MockReplicationClient.exclusions)

        yield self.be.change_replication_settings(rid, {'enabled': True})
        self.assertNotIn(rid, MockReplicationClient.exclusions)

    @inlineCallbacks
    def test_change_replication_settings_not_in_replications(self):
        """The settings can not be changed for an item not in replications."""
        rid = '0123-4567'
        assert rid not in MockReplicationClient.replications

        d = self.be.change_replication_settings(rid, {'enabled': True})
        yield self.assertFailure(d, replication_client.ReplicationError)

        d = self.be.change_replication_settings(rid, {'enabled': False})
        yield self.assertFailure(d, replication_client.ReplicationError)

    @inlineCallbacks
    def test_change_replication_settings_no_setting(self):
        """The change replication settings does not fail on empty settings."""
        rid = '0123-4567'
        MockReplicationClient.replications.add(rid)
        self.addCleanup(MockReplicationClient.replications.remove, rid)

        prior = MockReplicationClient.exclusions.copy()
        yield self.be.change_replication_settings(rid, {})

        self.assertEqual(MockReplicationClient.exclusions, prior)


class BackendFileSyncSettingsTestCase(BackendBasicTestCase):
    """File Sync Settings tests for the backend."""

    default_settings = backend.ControlBackend.DEFAULT_FILE_SYNC_SETTINGS

    @inlineCallbacks
    def setUp(self):
        yield super(BackendFileSyncSettingsTestCase, self).setUp()
        # restore default settings
        yield self.be.change_file_sync_settings(self.default_settings)

    @inlineCallbacks
    def assert_boolean_setting_is_correct(self, setting_name):
        """The 'setting_name' can be successfully changed."""
        new_value = not self.default_settings[setting_name]
        yield self.be.change_file_sync_settings({setting_name: new_value})

        self.assertEqual(getattr(self.be.sd_client, setting_name), new_value)

        actual = yield self.be.file_sync_settings_info()
        expected = self.default_settings.copy()
        expected[setting_name] = new_value
        self.assertEqual(expected, actual)

    @inlineCallbacks
    def test_file_sync_settings_info(self):
        """The settings_info method exercises its callback."""
        self.patch(self.be.sd_client, "throttling", True)
        self.be.sd_client.limits = {"download": 1000, "upload": 100}
        expected = {
            backend.AUTOCONNECT_KEY: self.be.sd_client.autoconnect,
            backend.SHOW_ALL_NOTIFICATIONS_KEY:
                self.be.sd_client.show_all_notifications,
            backend.SHARE_AUTOSUBSCRIBE_KEY:
                self.be.sd_client.share_autosubscribe,
            backend.UDF_AUTOSUBSCRIBE_KEY:
                self.be.sd_client.udf_autosubscribe,
            backend.DOWNLOAD_KEY: self.be.sd_client.limits['download'],
            backend.UPLOAD_KEY: self.be.sd_client.limits['upload'],
        }
        result = yield self.be.file_sync_settings_info()
        self.assertEqual(expected, result)

    @inlineCallbacks
    def test_file_sync_settings_info_with_limit(self):
        """The settings_info method exercises its callback."""
        self.patch(self.be.sd_client, "throttling", False)
        self.be.sd_client.limits = {"download": 987456, "upload": 125698}
        expected = {
            backend.AUTOCONNECT_KEY: self.be.sd_client.autoconnect,
            backend.SHOW_ALL_NOTIFICATIONS_KEY:
                self.be.sd_client.show_all_notifications,
            backend.SHARE_AUTOSUBSCRIBE_KEY:
                self.be.sd_client.share_autosubscribe,
            backend.UDF_AUTOSUBSCRIBE_KEY:
                self.be.sd_client.udf_autosubscribe,
            backend.DOWNLOAD_KEY: -1,
            backend.UPLOAD_KEY: -1,
        }
        result = yield self.be.file_sync_settings_info()
        self.assertEqual(expected, result)

    @inlineCallbacks
    def test_change_file_sync_setting_autoconnect(self):
        """The settings can be changed for autoconnect."""
        yield self.assert_boolean_setting_is_correct(backend.AUTOCONNECT_KEY)

    @inlineCallbacks
    def test_change_file_sync_setting_show_all_notifications(self):
        """The settings can be changed for show_all_notifications."""
        yield self.assert_boolean_setting_is_correct(
            backend.SHOW_ALL_NOTIFICATIONS_KEY)

    @inlineCallbacks
    def test_change_file_sync_setting_share_autosubscribe(self):
        """The settings can be changed for share_autosubscribe."""
        yield self.assert_boolean_setting_is_correct(
            backend.SHARE_AUTOSUBSCRIBE_KEY)

    @inlineCallbacks
    def test_change_file_sync_setting_udf_autosubscribe(self):
        """The settings can be changed for udf_autosubscribe."""
        yield self.assert_boolean_setting_is_correct(
            backend.UDF_AUTOSUBSCRIBE_KEY)

    @inlineCallbacks
    def test_change_file_sync_setting_download_bandwidth_limit(self):
        """The settings can be changed for download_bandwidth_limit."""
        new_value = 834
        setting_name = backend.DOWNLOAD_KEY
        yield self.be.change_file_sync_settings({setting_name: new_value})

        self.assertEqual(self.be.sd_client.throttling, True)
        self.assertEqual(self.be.sd_client.limits,
                         {'download': new_value, 'upload': -1})

    @inlineCallbacks
    def test_change_file_sync_setting_upload_bandwidth_limit(self):
        """The settings can be changed for upload_bandwidth_limit."""
        new_value = 932
        setting_name = backend.UPLOAD_KEY
        yield self.be.change_file_sync_settings({setting_name: new_value})

        self.assertEqual(self.be.sd_client.throttling, True)
        self.assertEqual(self.be.sd_client.limits,
                         {'download': -1, 'upload': new_value})

    @inlineCallbacks
    def test_no_download_limit_and_upload_change_to_no_limit(self):
        """The settings can be changed for download_bandwidth_limit.

        If the download limit was not set and the upload was set, when
        unsetting the upload limit the bandwidth_throttling_enabled is False.

        """
        only_one_limit_set = {
            backend.DOWNLOAD_KEY: -1,
            backend.UPLOAD_KEY: 52,
        }
        yield self.be.change_file_sync_settings(only_one_limit_set)

        # unset upload_bandwidth_limit
        setting_name = backend.UPLOAD_KEY
        yield self.be.change_file_sync_settings({setting_name: -1})

        self.assertEqual(self.be.sd_client.throttling, False)
        self.assertEqual(self.be.sd_client.limits,
                         {'download': -1, 'upload': -1})

    @inlineCallbacks
    def test_no_upload_limit_and_download_change_to_no_limit(self):
        """The settings can be changed for upload_bandwidth_limit.

        If the upload limit was not set and the download was set, when
        unsetting the download limit the bandwidth_throttling_enabled is False.

        """
        only_one_limit_set = {
            backend.DOWNLOAD_KEY: 33,
            backend.UPLOAD_KEY: -1,
        }
        yield self.be.change_file_sync_settings(only_one_limit_set)

        # unset download_bandwidth_limit
        setting_name = backend.DOWNLOAD_KEY
        yield self.be.change_file_sync_settings({setting_name: -1})

        self.assertEqual(self.be.sd_client.throttling, False)
        self.assertEqual(self.be.sd_client.limits,
                         {'download': -1, 'upload': -1})

    @inlineCallbacks
    def test_restore_defaults(self):
        """The defaults can be restored."""
        not_defaults = {
            backend.AUTOCONNECT_KEY: False,
            backend.SHOW_ALL_NOTIFICATIONS_KEY: False,
            backend.SHARE_AUTOSUBSCRIBE_KEY: True,
            backend.UDF_AUTOSUBSCRIBE_KEY: True,
            backend.DOWNLOAD_KEY: 204800,
            backend.UPLOAD_KEY: 2048,
        }
        yield self.be.change_file_sync_settings(not_defaults)

        # the call we want to test
        yield self.be.restore_file_sync_settings()

        self.assertEqual(self.be.sd_client.autoconnect, True)
        self.assertEqual(self.be.sd_client.show_all_notifications, True)
        self.assertEqual(self.be.sd_client.share_autosubscribe, False)
        self.assertEqual(self.be.sd_client.udf_autosubscribe, False)
        self.assertEqual(self.be.sd_client.throttling, False)
        self.assertEqual(self.be.sd_client.limits,
                         {'download': -1, 'upload': -1})

        result = yield self.be.file_sync_settings_info()
        self.assertEqual(self.default_settings, result)

    @inlineCallbacks
    def test_sync_menu(self):
        """Check that we get the right data to create the menu."""
        result = yield self.be.sync_menu()
        expected = {'recent-transfers': (), 'uploading': ()}
        self.assertEqual(result, expected)

    @inlineCallbacks
    def test_get_public_files(self):
        """Check that we get list of public files."""
        data = []

        def get_public_files_handler(publicfiles):
            """Callback for get_public_files."""
            data.append(publicfiles)

        self.be.set_public_files_list_handler(get_public_files_handler)
        yield self.be.get_public_files()
        expected = [[{'path': '/home/file1',
                      'public_url': 'http:ubuntuone.com/asd123'},
                     {'path': '/home/file2',
                      'public_url': 'http:ubuntuone.com/qwe456'}]]
        self.assertEqual(expected, data)

    @inlineCallbacks
    def test_get_shares(self):
        """Check that we get the list of shares."""
        result = yield self.be.get_shares()
        self.assertEqual(result, [])

    @inlineCallbacks
    def test_change_public_access_set_public(self):
        """Check that we get a notification when the access type change."""
        data = []

        def public_access_change_handler(info):
            """Callback for get_public_files."""
            data.append(info)

        self.be.set_public_access_changed_handler(public_access_change_handler)
        yield self.be.change_public_access('file1', True)
        expected = {'public_url': 'http:ubuntuone.com/zxc789'}
        self.assertEqual([expected], data)