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
robertc at robertcollins.net
Sun Mar 25 23:38:08 BST 2007
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)
More information about the bazaar-commits
mailing list