~ubuntu-branches/ubuntu/oneiric/ubuntuone-client/oneiric

« back to all changes in this revision

Viewing changes to tests/syncdaemon/test_dbus.py

• browse files at revision 1.1.21
• compare with another revision
• download diff
• download tarball
• view history from revision 1.1.21

Show diffs side-by-side

added

removed

tests/syncdaemon/test_dbus.py

37

37

    DBUS_IFACE_PUBLIC_FILES_NAME,

38

38

    EventListener,

39

39

)

40

 

from ubuntuone.syncdaemon.volume_manager import Share, Shared, UDF

 

40

from ubuntuone.syncdaemon.volume_manager import (

 

41

    Share,

 

42

    Shared,

 

43

    UDF,

 

44

    VolumeDoesNotExist,

 

45

)

41

46

from ubuntuone.syncdaemon.tools import DBusClient

42

47

from ubuntuone.syncdaemon import event_queue, states, main, config

43

48

from contrib.testing.testcase import (

799

804

        expected = []

800

805

        udf_id = str(uuid.uuid4())

801

806

        udf_path = self.main.vm._build_udf_path('~/foo/bar')

802

 

        udf = UDF(str(udf_id), str('udf_node_id'), '~/foo/bar', udf_path, True)

 

807

        udf = UDF(str(udf_id), str('udf_node_id'), u'~/foo/bar', udf_path, True)

803

808

        self.main.vm.udfs[udf_id] = udf

804

809

        for i in xrange(5):

805

810

            if i % 2:

981

986

        self.main.event_q.push('AQ_DOWNLOAD_STARTED', '', 'node_id', '')

982

987

        return d

983

988

 

 

989

    def test_download_file_progress(self):

 

990

        """Test the DBus signals in Status."""

 

991

        a_dir = os.path.join(self.root_dir, u'ñoño'.encode('utf-8'))

 

992

        self.fs_manager.create(a_dir, "", is_dir=False)

 

993

        self.fs_manager.set_node_id(a_dir, "node_id")

 

994

 

 

995

        d = defer.Deferred()

 

996

        def download_handler(path, info):

 

997

            """Handler for DownloadFileProgress signal."""

 

998

            self.assertEqual(a_dir, path.encode('utf-8'))

 

999

            self.assertEqual(info, {'n_bytes_read': '10',

 

1000

                                    'deflated_size': '20'})

 

1001

            d.callback(True)

 

1002

 

 

1003

        match = self.bus.add_signal_receiver(download_handler,

 

1004

                                     signal_name='DownloadFileProgress')

 

1005

        self.signal_receivers.add(match)

 

1006

        self.main.event_q.push('AQ_DOWNLOAD_FILE_PROGRESS',

 

1007

                               share_id='',

 

1008

                               node_id='node_id',

 

1009

                               n_bytes_read=10,

 

1010

                               deflated_size=20)

 

1011

        return d

 

1012

 

984

1013

    def test_download_finished(self):

985

1014

        """ Test the DBus signals in Status """

986

1015

        a_dir = os.path.join(self.root_dir, u'ñoño'.encode('utf-8'))

1056

1085

        self.main.event_q.push('AQ_UPLOAD_STARTED', '', 'node_id', '')

1057

1086

        return d

1058

1087

 

 

1088

    def test_upload_file_progress(self):

 

1089

        """Test the DBus signals in Status."""

 

1090

        a_dir = os.path.join(self.root_dir, u'ñoño'.encode('utf-8'))

 

1091

        self.fs_manager.create(a_dir, "", is_dir=False)

 

1092

        self.fs_manager.set_node_id(a_dir, "node_id")

 

1093

 

 

1094

        d = defer.Deferred()

 

1095

        def upload_handler(path, info):

 

1096

            """Handler for UploadFileProgress signal."""

 

1097

            self.assertEqual(a_dir, path.encode('utf-8'))

 

1098

            self.assertEqual(info, {'n_bytes_written': '10',

 

1099

                                     'deflated_size': '20'})

 

1100

            d.callback(True)

 

1101

 

 

1102

        match = self.bus.add_signal_receiver(upload_handler,

 

1103

                                     signal_name='UploadFileProgress')

 

1104

        self.signal_receivers.add(match)

 

1105

        self.main.event_q.push('AQ_UPLOAD_FILE_PROGRESS',

 

1106

                               share_id='',

 

1107

                               node_id='node_id',

 

1108

                               n_bytes_written=10,

 

1109

                               deflated_size=20)

 

1110

        return d

 

1111

 

1059

1112

    def test_upload_finished(self):

1060

1113

        """ Test the DBus signals in Status """

1061

1114

        a_dir = os.path.join(self.root_dir, u'ñoño'.encode('utf-8'))

1821

1874

    def _create_udf(self, id, node_id, suggested_path, subscribed=True):

1822

1875

        """Create an UDF and returns it and the volume"""

1823

1876

        path = self.main.vm._build_udf_path(suggested_path)

 

1877

        # make sure suggested_path is unicode

 

1878

        if isinstance(suggested_path, str):

 

1879

            suggested_path = suggested_path.decode('utf-8')

1824

1880

        udf = UDF(str(id), str(node_id), suggested_path, path, subscribed)

1825

1881

        return udf

1826

1882

 

1837

1893

        self.assertEquals(udf_dict['node_id'], udf.node_id)

1838

1894

        self.assertFalse(udf_dict['subscribed'])

1839

1895

 

 

1896

    def test_get_udf_dict_bad_encoding(self):

 

1897

        """Test for Folders._get_udf_dict."""

 

1898

        suggested_path = u'~/Música'

 

1899

        udf = self._create_udf(uuid.uuid4(), 'node_id', suggested_path,

 

1900

                               subscribed=False)

 

1901

        udf.suggested_path = udf.suggested_path.encode('utf-8')

 

1902

        udf_dict = self.dbus_iface.folders._get_udf_dict(udf)

 

1903

        # check the path it's unicode

 

1904

        self.assertEquals(udf_dict['path'], udf.path.decode('utf-8'))

 

1905

        self.assertEquals(udf_dict['volume_id'], udf.id)

 

1906

        self.assertEquals(repr(udf_dict['suggested_path']),

 

1907

                          repr(udf.suggested_path.decode('utf-8')))

 

1908

        self.assertEquals(udf_dict['node_id'], udf.node_id)

 

1909

        self.assertFalse(udf_dict['subscribed'])

 

1910

 

1840

1911

    @defer.inlineCallbacks

1841

1912

    def test_get_folders(self):

1842

1913

        """Test for Folders.get_folders."""

2018

2089

            """FolderDeleted handler."""

2019

2090

            self.assertRaises(KeyError, self.main.fs.get_by_path,

2020

2091

                              info['path'].decode('utf-8'))

2021

 

            self.assertRaises(KeyError, self.main.vm.get_volume, info['volume_id'])

 

2092

            self.assertRaises(VolumeDoesNotExist,

 

2093

                              self.main.vm.get_volume, info['volume_id'])

2022

2094

            d.callback(True)

2023

2095

        match = self.bus.add_signal_receiver(deleted_handler,

2024

2096

                                             signal_name='FolderDeleted')

2027

2099

            """the callback"""

2028

2100

            self.assertNotIn(udf.volume_id, self.main.vm.udfs)

2029

2101

            self.assertRaises(KeyError, self.main.fs.get_by_path, udf.path)

2030

 

            self.assertRaises(KeyError, self.main.vm.get_volume, udf.volume_id)

 

2102

            self.assertRaises(VolumeDoesNotExist,

 

2103

                              self.main.vm.get_volume, udf.volume_id)

2031

2104

        self.folders_client.call_method('delete', udf.volume_id,

2032

2105

                                        reply_handler=check_deleted,

2033

2106

                                        error_handler=self.error_handler)

2060

2133

        return d

2061

2134

 

2062

2135

    @defer.inlineCallbacks

 

2136

    def test_delete_error_signal_folder_id(self):

 

2137

        """Test for FolderDeleteError for a volume that doesn't exists."""

 

2138

        udf_id = 'foobar'

 

2139

        d = defer.Deferred()

 

2140

        def deleted_error_handler(info, error):

 

2141

            """FolderDeleteError handler"""

 

2142

            d.callback((info, error))

 

2143

        match = self.bus.add_signal_receiver(deleted_error_handler,

 

2144

                                             signal_name='FolderDeleteError')

 

2145

        self.signal_receivers.add(match)

 

2146

        self.folders_client.call_method('delete', udf_id,

 

2147

                                        reply_handler=lambda *args: None,

 

2148

                                        error_handler=d.errback)

 

2149

        info, error = yield d

 

2150

        self.assertEquals(info['volume_id'], udf_id)

 

2151

        self.assertEquals(error, "DOES_NOT_EXIST")

 

2152

 

 

2153

    @defer.inlineCallbacks

2063

2154

    def test_subscribe(self):

2064

2155

        """Test for Folders.subscribe and that it fires a dbus signal."""

2065

2156

        suggested_path = u'~/ñoño'

2164

2255

            """FolderDeleted handler."""

2165

2256

            self.assertRaises(KeyError, self.main.fs.get_by_path,

2166

2257

                              info['path'].decode('utf-8'))

2167

 

            self.assertRaises(KeyError, self.main.vm.get_volume, info['volume_id'])

 

2258

            self.assertRaises(VolumeDoesNotExist,

 

2259

                              self.main.vm.get_volume, info['volume_id'])

2168

2260

            d.callback(True)

2169

2261

        match = self.bus.add_signal_receiver(deleted_handler,

2170

2262

                                             signal_name='ShareDeleted')

2173

2265

            """the callback"""

2174

2266

            self.assertNotIn(share.volume_id, self.main.vm.shares)

2175

2267

            self.assertRaises(KeyError, self.main.fs.get_by_path, share.path)

2176

 

            self.assertRaises(KeyError, self.main.vm.get_volume, share.volume_id)

 

2268

            self.assertRaises(VolumeDoesNotExist,

 

2269

                              self.main.vm.get_volume, share.volume_id)

2177

2270

        client = DBusClient(self.bus, '/shares', DBUS_IFACE_SHARES_NAME)

2178

2271

        client.call_method('delete_share', share.volume_id,

2179

2272

                           reply_handler=check_deleted,

• Older »

Loggerhead is a web-based interface for Breezy
Version: 2.0.1