Rev 4: Really have a private dbus. in file:///home/robertc/source/baz/plugins/dbus/trunk/

[Robert Collins]

------------------------------------------------------------
revno: 4
revision-id: robertc at robertcollins.net-20070206050548-fr9um4fo1070o1sw
parent: robertc at robertcollins.net-20070206041636-nfxqkxb7jrmgndx0
committer: Robert Collins <robertc at robertcollins.net>
branch nick: trunk
timestamp: Tue 2007-02-06 16:05:48 +1100
message:
  Really have a private dbus.
modified:
  tests/test_activity.py         test_activity.py-20070206035206-bnhzvtm6m6hpgylz-1
=== modified file 'tests/test_activity.py'
--- a/tests/test_activity.py	2007-02-06 04:16:36 +0000
+++ b/tests/test_activity.py	2007-02-06 05:05:48 +0000
@@ -18,20 +18,106 @@
 
 """Tests for the dbus activity service."""
 
+import os
+import signal
+import subprocess
+import tempfile
+import thread
+import weakref
+
 import dbus
+import _dbus_bindings
 
 from bzrlib.tests import TestCaseWithMemoryTransport
 
 from bzrlib.plugins.dbus import activity
 
 
+def create_daemon():
+    """Create a dbus daemon."""
+    config_file = tempfile.NamedTemporaryFile()
+    config_file.write('''
+<!DOCTYPE busconfig PUBLIC "-//freedesktop//DTD D-Bus Bus Configuration 1.0//EN"
+ "http://www.freedesktop.org/standards/dbus/1.0/busconfig.dtd">
+<busconfig>
+  <!-- Our well-known bus type, don't change this -->
+  <type>session</type>
+
+  <listen>unix:tmpdir=/tmp</listen>
+
+  <standard_session_servicedirs />
+
+  <policy context="default">
+    <!-- Allow everything to be sent -->
+    <allow send_destination="*"/>
+    <!-- Allow everything to be received -->
+    <allow eavesdrop="true"/>
+    <!-- Allow anyone to own anything -->
+    <allow own="*"/>
+  </policy>
+
+</busconfig>
+''')
+    config_file.flush()
+    proc = subprocess.Popen(
+        ['dbus-daemon', '--config-file=%s' % config_file.name,
+        '--print-address'], stdout=subprocess.PIPE)
+    address = proc.stdout.readline()
+    return address.strip(), proc
+
+
+class PrivateBus(dbus.Bus):
+    """A PrivateBus created within this application."""
+
+    def __new__(cls):
+        address, proc = create_daemon()
+        try:
+            bus = _dbus_bindings.BusImplementation.__new__(cls, address)
+        except:
+            os.kill(proc.pid, signal.SIGINT)
+            raise
+        bus._test_process = proc
+        bus._bus_type = address
+        # _bus_names is used by dbus.service.BusName!
+        bus._bus_names = weakref.WeakValueDictionary()
+
+        bus._signal_recipients_by_object_path = {}
+        """Map from object path to dict mapping dbus_interface to dict
+        mapping member to list of SignalMatch objects."""
+
+        bus._signal_sender_matches = {}
+        """Map from sender well-known name to list of match rules for all
+        signal handlers that match on sender well-known name."""
+
+        bus._signals_lock = thread.allocate_lock()
+        """Lock used to protect signal data structures if doing two
+        removals at the same time (everything else is atomic, thanks to
+        the GIL)"""
+
+        bus.add_message_filter(bus.__class__._signal_func)
+
+        return bus
+
+    def nuke(self):
+        """Ensure the daemon is shutdown."""
+        os.kill(self._test_process.pid, signal.SIGINT)
+    
+    def __str__(self):
+        return "Private DBUS (%s)" % self._bus_type
+
+    def __repr__(self):
+        return '<%s.%s on %s at %#x>' % (__name__, self.__class__,
+            self._bus_type, id(self))
+
+
 class TestActivity(TestCaseWithMemoryTransport):
 
     def setUp(self):
         TestCaseWithMemoryTransport.setUp(self)
-        # AFAIK this creates a private dbus session so we dont spam 
+        # setup a private dbus session so we dont spam 
         # the users desktop!
-        self.bus = dbus.StarterBus()
+        self.bus = PrivateBus()
+        self.addCleanup(self.bus.nuke)
         
     def test_advertise_branch_no_service_running(self):
         # should not error: just call it