← Back to branch summary

~verterok/ubuntuone-client/volumemanager_udfs-2

« back to all changes in this revision

Viewing changes to tests/syncdaemon/test_dbus.py

  • Committer: guillermo.gonzalez at canonical
  • Date: 2010-01-12 18:25:40 UTC
  • Revision ID: guillermo.gonzalez@canonical.com-20100112182540-omkse31tb3plq1bd
add Shares.delete_share method and ShareDeleteErro signal to DBus interface

Show diffs side-by-side

added added

removed removed

Lines of Context:
tests/syncdaemon/test_dbus.py
1576
1576
            d.callback(True)
1577
1577
        match = self.bus.add_signal_receiver(deleted_handler,
1578
1578
                                             signal_name='FolderDeleted')
 
1579
        self.signal_receivers.add(match)
1579
1580
        def check_deleted(info):
1580
1581
            """the callback"""
1581
1582
            self.assertNotIn(udf.id, self.main.vm.udfs)
1582
1583
            self.assertRaises(KeyError, self.main.fs.get_by_path, udf.path)
1583
1584
            self.assertRaises(KeyError, self.main.vm.get_volume, udf.id)
1584
 
        self.signal_receivers.add(match)
1585
1585
        self.folders_client.call_method('delete', udf.id,
1586
1586
                                        reply_handler=check_deleted,
1587
1587
                                        error_handler=self.error_handler)
1702
1702
        return d
1703
1703
 
1704
1704
 
 
1705
class ShareTests(DBusTwistedTestCase):
 
1706
    """Share specific tests"""
 
1707
 
 
1708
    def test_delete_share(self):
 
1709
        """Test for Shares.delete_share."""
 
1710
        share_path = os.path.join(self.main.shares_dir, 'share')
 
1711
        share = Share(path=share_path, share_id='share_id',
 
1712
                      subtree='node_id', accepted=True)
 
1713
        self.main.vm.add_share(share)
 
1714
        d = defer.Deferred()
 
1715
        def delete_volume(volume_id):
 
1716
            """Fake delete_volume"""
 
1717
            self.main.event_q.push("AQ_DELETE_VOLUME_OK", volume_id=volume_id)
 
1718
        self.main.action_q.delete_volume = delete_volume
 
1719
        def deleted_handler(info):
 
1720
            """FolderDeleted handler."""
 
1721
            self.assertRaises(KeyError, self.main.fs.get_by_path,
 
1722
                              info['path'].decode('utf-8'))
 
1723
            self.assertRaises(KeyError, self.main.vm.get_volume, info['id'])
 
1724
            d.callback(True)
 
1725
        match = self.bus.add_signal_receiver(deleted_handler,
 
1726
                                             signal_name='ShareDeleted')
 
1727
        self.signal_receivers.add(match)
 
1728
        def check_deleted(info):
 
1729
            """the callback"""
 
1730
            self.assertNotIn(share.id, self.main.vm.shares)
 
1731
            self.assertRaises(KeyError, self.main.fs.get_by_path, share.path)
 
1732
            self.assertRaises(KeyError, self.main.vm.get_volume, share.id)
 
1733
        client = DBusClient(self.bus, '/shares', DBUS_IFACE_SHARES_NAME)
 
1734
        client.call_method('delete_share', share.id,
 
1735
                           reply_handler=check_deleted,
 
1736
                           error_handler=self.error_handler)
 
1737
        return d
 
1738
 
 
1739
    def test_delete_share_error_signal(self):
 
1740
        """Test for Shares.delete_share with an error."""
 
1741
        share_path = os.path.join(self.main.shares_dir, 'share')
 
1742
        share = Share(path=share_path, share_id='share_id',
 
1743
                      subtree='node_id', accepted=True)
 
1744
        self.main.vm.add_share(share)
 
1745
        d = defer.Deferred()
 
1746
        # patch delete_volume to fail
 
1747
        def delete_volume(volume_id):
 
1748
            """Fake delete_volume"""
 
1749
            self.main.event_q.push("AQ_DELETE_VOLUME_ERROR",
 
1750
                                   volume_id=volume_id, error="I'm broken")
 
1751
        self.main.action_q.delete_volume = delete_volume
 
1752
        def deleted_error_handler(info, error):
 
1753
            """FolderDeleteError handler"""
 
1754
            self.assertEquals(info['id'], share.id)
 
1755
            self.assertEquals(error, "I'm broken")
 
1756
            d.callback(True)
 
1757
        match = self.bus.add_signal_receiver(deleted_error_handler,
 
1758
                                             signal_name='ShareDeleteError')
 
1759
        self.signal_receivers.add(match)
 
1760
        client = DBusClient(self.bus, '/shares', DBUS_IFACE_SHARES_NAME)
 
1761
        client.call_method('delete_share', share.id,
 
1762
                           reply_handler=lambda *args: None,
 
1763
                           error_handler=self.error_handler)
 
1764
        return d
 
1765
 
 
1766
 
1705
1767
def test_suite():
1706
1768
    # pylint: disable-msg=C0111
1707
1769
    return unittest.TestLoader().loadTestsFromName(__name__)