Rev 21: Add add_url_map and remove_url_map DBus methods to the Broadcast object. in file:///home/robertc/source/baz/plugins/dbus/trunk/

[Robert Collins]

At file:///home/robertc/source/baz/plugins/dbus/trunk/

------------------------------------------------------------
revno: 21
revision-id: robertc at robertcollins.net-20070325223808-t3hben50faont9u5
parent: robertc at robertcollins.net-20070325222024-xxnx7foea3rgdm7w
committer: Robert Collins <robertc at robertcollins.net>
branch nick: trunk
timestamp: Mon 2007-03-26 08:38:08 +1000
message:
  Add add_url_map and remove_url_map DBus methods to the Broadcast object.
modified:
  activity.py                    activity.py-20070206034725-q208d0jtkshwu0fx-1
  tests/test_activity.py         test_activity.py-20070206035206-bnhzvtm6m6hpgylz-1
=== modified file 'activity.py'
--- a/activity.py	2007-03-25 22:13:28 +0000
+++ b/activity.py	2007-03-25 22:38:08 +0000
@@ -26,6 +26,7 @@
 import gobject
 
 from bzrlib.revision import NULL_REVISION
+from bzrlib.plugins.dbus import mapper
 
 
 class Activity(object):
@@ -139,6 +140,13 @@
         bus_name = dbus.service.BusName(
             Broadcast.DBUS_NAME, bus=bus)
         dbus.service.Object.__init__(self, bus, Broadcast.DBUS_PATH, bus_name)
+        self.url_mapper = mapper.URLMapper()
+
+    @dbus.service.method(DBUS_INTERFACE,
+                         in_signature='ss', out_signature='')
+    def add_url_map(self, source_prefix, target_prefix):
+        """Add a url prefix to be mapped when advertising revisions."""
+        self.url_mapper.maps.setdefault(source_prefix, []).append(target_prefix)
 
     @dbus.service.method(DBUS_INTERFACE,
                          in_signature='ss', out_signature='')
@@ -150,6 +158,15 @@
         """
         self.Revision(revision_id, [url])
 
+    @dbus.service.method(DBUS_INTERFACE,
+                         in_signature='ss', out_signature='')
+    def remove_url_map(self, source_prefix, target_prefix):
+        """Remove a previously mapped url prefix."""
+        pos = self.url_mapper.maps[source_prefix].index(target_prefix)
+        del self.url_mapper.maps[source_prefix][pos]
+        if not self.url_mapper.maps[source_prefix]:
+            del self.url_mapper.maps[source_prefix]
+
     @dbus.service.signal(DBUS_INTERFACE, 'sas')
     def Revision(self, revision, urls):
         """A revision has been observed at url."""

=== modified file 'tests/test_activity.py'
--- a/tests/test_activity.py	2007-03-25 22:13:28 +0000
+++ b/tests/test_activity.py	2007-03-25 22:38:08 +0000
@@ -327,3 +327,80 @@
             signals1)
         self.assertEqual([('revision1', ['url1']), ('revision2', ['url2'])],
             signals2)
+
+    def test_add_url_map(self):
+        """Calling add_url_map via dbus should succeed and return nothing."""
+        # attach a Broadcaster.
+        broadcaster = activity.Broadcast(self.bus)
+        dbus_object = self.bus.get_object(activity.Broadcast.DBUS_NAME,
+            activity.Broadcast.DBUS_PATH)
+        dbus_iface = dbus.Interface(dbus_object,
+            activity.Broadcast.DBUS_INTERFACE)
+
+        # call add_url_map.
+        errors = []
+        results = []
+        def handle_reply():
+            results.append(None)
+            if len(results) + len(errors) == 2:
+                mainloop.quit() # end the mainloop
+        def handle_error(error):
+            errors.append(error)
+            if len(results) + len(errors) == 2:
+                mainloop.quit() # end the mainloop
+        dbus_iface.add_url_map('foo/', 'bar/',
+            reply_handler=handle_reply,
+            error_handler=handle_error)
+        # for two locations at the same url
+        dbus_iface.add_url_map('foo/', 'baz/',
+            reply_handler=handle_reply,
+            error_handler=handle_error)
+        mainloop = gobject.MainLoop()
+        mainloop.run()
+        if errors:
+            raise errors[0]
+        self.assertEqual([None, None], results)
+        self.assertEqual([], errors)
+        self.assertEqual({'foo/':['bar/', 'baz/']}, broadcaster.url_mapper.maps)
+
+    def test_remove_url_map(self):
+        """Calling remove_url_map via dbus should succeed and return nothing."""
+        # attach a Broadcaster.
+        broadcaster = activity.Broadcast(self.bus)
+        dbus_object = self.bus.get_object(activity.Broadcast.DBUS_NAME,
+            activity.Broadcast.DBUS_PATH)
+        dbus_iface = dbus.Interface(dbus_object,
+            activity.Broadcast.DBUS_INTERFACE)
+
+        # call add_url_map.
+        errors = []
+        results = []
+        def handle_reply():
+            results.append(None)
+            if len(results) + len(errors) == 4:
+                mainloop.quit() # end the mainloop
+        def handle_error(error):
+            errors.append(error)
+            if len(results) + len(errors) == 4:
+                mainloop.quit() # end the mainloop
+        # add two locations at the same prefix
+        dbus_iface.add_url_map('foo/', 'bar/',
+            reply_handler=handle_reply,
+            error_handler=handle_error)
+        dbus_iface.add_url_map('foo/', 'baz/',
+            reply_handler=handle_reply,
+            error_handler=handle_error)
+        # and remove them both
+        dbus_iface.remove_url_map('foo/', 'bar/',
+            reply_handler=handle_reply,
+            error_handler=handle_error)
+        dbus_iface.remove_url_map('foo/', 'baz/',
+            reply_handler=handle_reply,
+            error_handler=handle_error)
+        mainloop = gobject.MainLoop()
+        mainloop.run()
+        if errors:
+            raise errors[0]
+        self.assertEqual([None, None, None, None], results)
+        self.assertEqual([], errors)
+        self.assertEqual({}, broadcaster.url_mapper.maps)