Merge lp:~jderose/dmedia/closer-to-restful into lp:dmedia

[Jason Gerard DeRose]

Okay, I apologize isn't going to be easy to review. More cleanup/refactoring work is still needed, and I certainly need to spend time documenting all of this. All the same, this fixes an important stability problem that dmedia has been experiencing under Natty:

https://bugs.launchpad.net/ubuntu/+source/desktopcouch/+bug/714406

I'm confident that this is the right step architecturally for dmedia, I just don't have time to make things all clean and formally documented before 0.4.

[lp:~jderose/dmedia/closer-to-restful]

187. By Jason Gerard DeRose on 2011-02-21

Fixed typo dmj726 spotted

188. By Jason Gerard DeRose on 2011-02-21

another typo dmj726 spotted

189. By Jason Gerard DeRose on 2011-02-21

dmj726 caught me importing time twice

[David Jordan (dmj726)]

Looks good from what I can see.

review: Approve

[Jason Gerard DeRose (jderose)]

Thanks, David!

[Jason Gerard DeRose (jderose)]

review: Approve

=== modified file 'dmedia-gtk'
--- dmedia-gtk	2011-01-25 12:11:16 +0000
+++ dmedia-gtk	2011-02-21 12:01:51 +0000
@@ -25,6 +25,7 @@
 from subprocess import check_call
 import optparse
 import dmedia
+from dmedia.abstractcouch import get_env
 from dmedia.metastore import MetaStore
 from dmedia import ui
 
@@ -39,7 +40,7 @@
 (options, args) = parser.parse_args()
 
 
-store = MetaStore()
+store = MetaStore(get_env())
 app = ui.create_app()
 store.update(app)
 

=== modified file 'dmedia-service'
--- dmedia-service	2011-01-30 21:17:48 +0000
+++ dmedia-service	2011-02-21 12:01:51 +0000
@@ -21,26 +21,30 @@
 # You should have received a copy of the GNU Affero General Public License along
 # with `dmedia`.  If not, see <http://www.gnu.org/licenses/>.
 
-from dmedia.util import configure_logging
-from dmedia.constants import BUS
 import sys
 import optparse
-from os import path
+
 import dbus
-import dbus.service
-import dbus.mainloop.glib
+from dbus.mainloop.glib import DBusGMainLoop
 import gobject
 
-
-parser = optparse.OptionParser()
+DBusGMainLoop(set_as_default=True)
+gobject.threads_init()
+
+import dmedia
+from dmedia.constants import BUS, DBNAME
+from dmedia.abstractcouch import get_env
+
+
+parser = optparse.OptionParser(version=dmedia.__version__)
 parser.add_option('--bus',
     default=BUS,
     help='D-Bus bus name; default is %r' % BUS,
 )
 parser.add_option('--dbname',
     metavar='DIR',
-    default=None,
-    help='dmedia CouchDB database',
+    default=DBNAME,
+    help='CouchDB database; default is %r' % DBNAME,
 )
 parser.add_option('--no-gui',
     action='store_true',
@@ -57,10 +61,15 @@
 
 if __name__ == '__main__':
     (options, args) = parser.parse_args()
+
+    from dmedia.util import configure_logging
     configure_logging('service')
+
+    env = get_env(options.dbname)
+    env['bus'] = options.bus
+    env['no_gui'] = options.no_gui
+
     from dmedia import service
     mainloop = gobject.MainLoop()
-    obj = service.DMedia(mainloop.quit,
-        options.bus, options.dbname, options.no_gui
-    )
+    obj = service.DMedia(env, mainloop.quit)
     mainloop.run()

=== modified file 'dmedia/__init__.py'
--- dmedia/__init__.py	2011-01-26 16:23:49 +0000
+++ dmedia/__init__.py	2011-02-21 12:01:51 +0000
@@ -41,3 +41,31 @@
 assert path.isdir(packagedir)
 datadir = path.join(packagedir, 'data')
 assert path.isdir(datadir)
+
+
+def get_env(dbname=None):
+    """
+    Get desktopcouch runtime info in most the lightweight way possible.
+
+    Here "lightweight" doesn't necessarily mean "fast", but with as few imports
+    as possible to keep the dmedia memory footprint small.
+    """
+    import dbus
+    DC = 'org.desktopcouch.CouchDB'
+    conn = dbus.SessionBus()
+    proxy = conn.get_object(DC, '/')
+    getPort = proxy.get_dbus_method('getPort', dbus_interface=DC)
+    port = getPort()
+    url = 'http://localhost:%d/' % port
+
+    import gnomekeyring
+    data = gnomekeyring.find_items_sync(
+        gnomekeyring.ITEM_GENERIC_SECRET,
+        {'desktopcouch': 'oauth'}
+    )
+    oauth = dict(zip(
+        ('consumer_key', 'consumer_secret', 'token', 'token_secret'),
+        data[0].secret.split(':')
+    ))
+
+    return dict(port=port, url=url, oauth=oauth, dbname=dbname)

=== added file 'dmedia/abstractcouch.py'
--- dmedia/abstractcouch.py	1970-01-01 00:00:00 +0000
+++ dmedia/abstractcouch.py	2011-02-21 12:01:51 +0000
@@ -0,0 +1,158 @@
+# Authors:
+#   Jason Gerard DeRose <[e-mail]>
+#
+# dmedia: distributed media library
+# Copyright (C) 2011 Jason Gerard DeRose <[e-mail]>
+#
+# This file is part of `dmedia`.
+#
+# `dmedia` is free software: you can redistribute it and/or modify it under the
+# terms of the GNU Affero General Public License as published by the Free
+# Software Foundation, either version 3 of the License, or (at your option) any
+# later version.
+#
+# `dmedia` is distributed in the hope that it will be useful, but WITHOUT ANY
+# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
+# A PARTICULAR PURPOSE.  See the GNU Affero General Public License for more
+# details.
+#
+# You should have received a copy of the GNU Affero General Public License along
+# with `dmedia`.  If not, see <http://www.gnu.org/licenses/>.
+
+"""
+Small abstraction to hide desktopcouch vs system-wide CouchDB details.
+
+dmedia is obviously very focused on desktop use, for which desktopcouch is a
+godsend.  However, dmedia also needs to run on headless, minimal servers.  We
+want to abstract the difference to a single place in the code so the rest of the
+code can just work without concern for whether we're connecting to a CouchDB
+instance launched by desktopcouch, or a system-wide CouchDB started with an
+init script.
+
+This would be a great feature to upstream into desktopcouch, but for now we can
+give it a good trial by fire in dmedia.
+
+For more details, see:
+
+    https://bugs.launchpad.net/dmedia/+bug/722035
+"""
+
+import logging
+
+from couchdb import Server, ResourceNotFound
+try:
+    from desktopcouch.records.http import OAuthSession
+except ImportError:
+    OAuthSession = None
+if OAuthSession is not None:
+    from desktopcouch.application.platform import find_port
+    from desktopcouch.application.local_files import get_oauth_tokens
+
+
+log = logging.getLogger()
+
+
+def get_couchdb_server(env):
+    """
+    Return `couchdb.Server` for desktopcouch or system-wide CouchDB.
+
+    The *env* argument is a ``dict`` instance containing information about how
+    to connect to CouchDB.  It must always contain a ``"url"`` key, which is the
+    URL as passed to ``couchdb.Server()``.
+
+    If *env* contains an ``"oauth"`` key, a
+    ``desktopcouch.records.http.OAuthSession`` will be created and passed to
+    ``couchdb.Server()``.
+
+    For example, to connect to the system-wide CouchDB, pass an *env* like this:
+
+    >>> env = {'url': 'http://localhost:5984/'}
+
+    Or to connect to a per-user desktopcouch CouchDB, pass an *env* like this:
+
+    >>> env = {
+    ...     'url': 'http://localhost:51074/',
+    ...     'oauth': {
+    ...         'consumer_secret': 'no',
+    ...         'token': 'way',
+    ...         'consumer_key': 'too',
+    ...         'token_secret': 'secret'
+    ...     }
+    ... }
+
+    When using desktopcouch, you can build an *env* with the correct port and
+    oauth credentials like this:
+
+    >>> from desktopcouch.application.platform import find_port
+    >>> from desktopcouch.application.local_files import get_oauth_tokens
+    >>> env = {
+    ...     'url': 'http://localhost:%d/' % find_port(),
+    ...     'oauth': get_oauth_tokens(),
+    ... }
+
+    Note the reason *env* is a ``dict`` is so it's easily extensible and can
+    carry other useful information, for example the dmedia database name so that
+    an alternate name can be provided for, say,  unit tests (like
+    ``"dmedia_test"``).
+
+    The goal is to have all the needed information is one easily serialized
+    piece of data (important for testing across multiple processes).
+    """
+    url = env.get('url', 'http://localhost:5984/')
+    log.info('CouchDB server is %r', url)
+    if env.get('oauth') is None:
+        session = None
+    else:
+        if OAuthSession is None:
+            raise ValueError(
+                "provided env['oauth'] but OAuthSession not available: %r" %
+                    (env,)
+            )
+        log.info('Using desktopcouch `OAuthSession`')
+        session = OAuthSession(credentials=env['oauth'])
+    return Server(url, session=session)
+
+
+def get_dmedia_db(env, server=None):
+    """
+    Return the dmedia database specified by *env*.
+
+    The database name is determined by ``env['dbname']``.  If the ``"dbname"``
+    key is missing or is ``None``, the default database name ``"dmedia"`` is
+    used.
+
+    If the database does not exist, it will be created.
+
+    If *server* is ``None``, one is created based on *env* by calling
+    `get_couchdb_server()`.
+
+    Returns a ``couchdb.Database`` instance.
+    """
+    if server is None:
+        server = get_couchdb_server(env)
+    dbname = env.get('dbname')
+    if dbname is None:
+        dbname = 'dmedia'
+    log.info('CouchDB database is %r', dbname)
+    try:
+        return server[dbname]
+    except ResourceNotFound:
+        return server.create(dbname)
+
+
+def get_env(dbname=None):
+    """
+    Return default *env*.
+
+    This will return an appropriate *env* based on whether desktopcouch is
+    available.  Not a perfect solution, but works for now.
+    """
+    if OAuthSession is None:
+        return {'dbname': dbname}
+    port = find_port()
+    return {
+        'port': port,
+        'url': 'http://localhost:%d/' % port,
+        'oauth': get_oauth_tokens(),
+        'dbname': dbname,
+    }

=== modified file 'dmedia/constants.py'
--- dmedia/constants.py	2011-02-06 11:11:03 +0000
+++ dmedia/constants.py	2011-02-21 12:01:51 +0000
@@ -27,6 +27,8 @@
 import mimetypes
 mimetypes.init()
 
+DBNAME = 'dmedia'
+
 # Standard read/write buffer size:
 CHUNK_SIZE = 2**20  # 1 MiB
 
@@ -45,7 +47,10 @@
 
 # D-Bus releated:
 BUS = 'org.freedesktop.DMedia'
-INTERFACE = 'org.freedesktop.DMedia'
+INTERFACE = BUS
+DC_BUS = 'org.desktopcouch.CouchDB'
+DC_INTERFACE = DC_BUS
+
 
 # Standard format for TypeError message:
 TYPE_ERROR = '%s: need a %r; got a %r: %r'

=== modified file 'dmedia/importer.py'
--- dmedia/importer.py	2011-02-19 03:05:53 +0000
+++ dmedia/importer.py	2011-02-21 12:01:51 +0000
@@ -35,10 +35,12 @@
 
 from .util import random_id
 from .errors import DuplicateFile
-from .workers import Worker, Manager, register, isregistered, exception_name
+from .workers import (
+    CouchWorker, CouchManager, register, isregistered, exception_name
+)
 from .filestore import FileStore, quick_id, safe_open, safe_ext, pack_leaves
-from .metastore import MetaStore
 from .extractor import merge_metadata
+from .abstractcouch import get_env, get_couchdb_server, get_dmedia_db
 
 
 mimetypes.init()
@@ -188,17 +190,14 @@
     }
 
 
-class Importer(object):
-    def __init__(self, batch_id, base, extract, dbname=None):
-        self.batch_id = batch_id
-        self.base = base
-        self.extract = extract
+class ImportWorker(CouchWorker):
+    def __init__(self, env, q, key, args):
+        super(ImportWorker, self).__init__(env, q, key, args)
+        (self.base, self.extract) = args
         self.home = path.abspath(os.environ['HOME'])
-        self.metastore = MetaStore(dbname=dbname)
-        self.db = self.metastore.db
         self.filestore = FileStore(
             path.join(self.home, DOTDIR),
-            self.metastore.machine_id
+            self.env.get('machine_id')
         )
         try:
             self.db.save(self.filestore._doc)
@@ -210,6 +209,27 @@
         self.doc = None
         self._id = None
 
+    def execute(self, base, extract=False):
+        import_id = self.start()
+        self.emit('started', import_id)
+
+        files = self.scanfiles()
+        total = len(files)
+        self.emit('count', import_id, total)
+
+        c = 1
+        for (src, action) in self.import_all_iter():
+            self.emit('progress', import_id, c, total,
+                dict(
+                    action=action,
+                    src=src,
+                )
+            )
+            c += 1
+
+        stats = self.finalize()
+        self.emit('finished', import_id, stats)
+
     def save(self):
         """
         Save current 'dmedia/import' record to CouchDB.
@@ -222,8 +242,8 @@
         """
         assert self._id is None
         self.doc = create_import(self.base,
-            batch_id=self.batch_id,
-            machine_id=self.metastore.machine_id,
+            batch_id=self.env.get('batch_id'),
+            machine_id=self.env.get('machine_id'),
         )
         self._id = self.doc['_id']
         self.save()
@@ -368,34 +388,13 @@
         assert list(t[0] for t in self.filetuples) == self._processed
         self.doc['time_end'] = time.time()
         self.save()
+        dt = self.doc['time_end'] - self.doc['time']
+        log.info('Completed import of %r in %d:%02d',
+            self.base, dt / 60, dt % 60
+        )
         return self.doc['stats']
 
 
-class ImportWorker(Worker):
-    def execute(self, batch_id, base, extract=False, dbname=None):
-
-        adapter = Importer(batch_id, base, extract, dbname)
-
-        import_id = adapter.start()
-        self.emit('started', import_id)
-
-        files = adapter.scanfiles()
-        total = len(files)
-        self.emit('count', import_id, total)
-
-        c = 1
-        for (src, action) in adapter.import_all_iter():
-            self.emit('progress', import_id, c, total,
-                dict(
-                    action=action,
-                    src=src,
-                )
-            )
-            c += 1
-
-        stats = adapter.finalize()
-        self.emit('finished', import_id, stats)
-
 
 def to_dbus_stats(stats):
     return dict(
@@ -414,12 +413,9 @@
             accum[key][k] += v
 
 
-class ImportManager(Manager):
-    def __init__(self, callback=None, dbname=None):
-        super(ImportManager, self).__init__(callback)
-        self._dbname = dbname
-        self.metastore = MetaStore(dbname=dbname)
-        self.db = self.metastore.db
+class ImportManager(CouchManager):
+    def __init__(self, env, callback=None):
+        super(ImportManager, self).__init__(env, callback)
         self.doc = None
         self._total = 0
         self._completed = 0
@@ -437,10 +433,15 @@
         assert self._workers == {}
         self._total = 0
         self._completed = 0
-        self.doc = create_batch(self.metastore.machine_id)
+        self.doc = create_batch(self.env.get('machine_id'))
         self.save()
         self.emit('BatchStarted', self.doc['_id'])
 
+    def get_worker_env(self, worker, key, args):
+        env = dict(self.env)
+        env['batch_id'] = self.doc['_id']
+        return env
+
     def _finish_batch(self):
         assert self._workers == {}
         self.doc['time_end'] = time.time()
@@ -449,6 +450,7 @@
             to_dbus_stats(self.doc['stats'])
         )
         self.doc = None
+        self.env = None
         log.info('Batch complete, compacting database...')
         self.db.compact()
 
@@ -494,10 +496,4 @@
                 return False
             if len(self._workers) == 0:
                 self._start_batch()
-            return self.do('ImportWorker', base,
-                self.doc['_id'], base, extract, self._dbname
-            )
-
-    def list_imports(self):
-        with self._lock:
-            return sorted(self._workers)
+            return self.start_job('ImportWorker', base, base, extract)

=== modified file 'dmedia/metastore.py'
--- dmedia/metastore.py	2011-02-18 19:22:23 +0000
+++ dmedia/metastore.py	2011-02-21 12:01:51 +0000
@@ -27,16 +27,11 @@
 import time
 import socket
 import platform
+
 import gnomekeyring
 from couchdb import ResourceNotFound, ResourceConflict
-import desktopcouch
-from desktopcouch.records.server import  CouchDatabase
-from desktopcouch.records.record import  Record
-from desktopcouch.local_files import DEFAULT_CONTEXT, Context
-try:
-    from desktopcouch import find_port
-except ImportError:
-    from desktopcouch.application.platform import find_port
+
+from .abstractcouch import get_couchdb_server, get_dmedia_db
 from .util import random_id
 
 
@@ -182,14 +177,18 @@
         )),
     )
 
-    def __init__(self, dbname=None):
-        self.dbname = ('dmedia' if dbname is None else dbname)
-        self.desktop = CouchDatabase(self.dbname, create=True)
-        self.server = self.desktop._server
-        self.db = self.server[self.dbname]
+    def __init__(self, env):
+        self.env = env
+        self.server = get_couchdb_server(env)
+        self.db = get_dmedia_db(env, self.server)
         self.create_views()
         self._machine_id = None
 
+    def get_env(self):
+        env = dict(self.env)
+        env['machine_id'] = self.machine_id
+        return env
+
     def get_basic_auth(self):
         data = gnomekeyring.find_items_sync(
             gnomekeyring.ITEM_GENERIC_SECRET,
@@ -199,7 +198,7 @@
         return (user, password)
 
     def get_port(self):
-        return find_port()
+        return self.env.get('port')
 
     def get_uri(self):
         return 'http://localhost:%s' % self.get_port()

=== modified file 'dmedia/service.py'
--- dmedia/service.py	2011-01-30 21:17:48 +0000
+++ dmedia/service.py	2011-02-21 12:01:51 +0000
@@ -33,14 +33,11 @@
 import dbus.service
 import dbus.mainloop.glib
 import gobject
-from .constants import BUS, INTERFACE, EXT_MAP
+from .constants import BUS, INTERFACE, DBNAME, EXT_MAP
 from .util import NotifyManager, Timer, import_started, batch_finished
 from .importer import ImportManager
 from .metastore import MetaStore
 
-gobject.threads_init()
-dbus.mainloop.glib.DBusGMainLoop(set_as_default=True)
-
 try:
     import pynotify
     pynotify.init('dmedia')
@@ -70,23 +67,24 @@
         'ImportFinished',
     ])
 
-    def __init__(self, killfunc=None, bus=None, dbname=None, no_gui=False):
+    def __init__(self, env, killfunc=None):
+        self._env = env
         self._killfunc = killfunc
-        self._bus = (BUS if bus is None else bus)
-        self._dbname = dbname
-        self._no_gui = no_gui
+        self._bus = env.get('bus', BUS)
+        self._dbname = env.get('dbname', DBNAME)
+        self._no_gui = env.get('no_gui', False)
         log.info('Starting service on %r', self._bus)
         self._conn = dbus.SessionBus()
         super(DMedia, self).__init__(self._conn, object_path='/')
         self._busname = dbus.service.BusName(self._bus, self._conn)
 
-        if no_gui or pynotify is None:
+        if self._no_gui or pynotify is None:
             self._notify = None
         else:
             log.info('Using `pynotify`')
             self._notify = NotifyManager()
 
-        if no_gui or appindicator is None:
+        if self._no_gui or appindicator is None:
             self._indicator = None
         else:
             log.info('Using `appindicator`')
@@ -118,12 +116,21 @@
             self._indicator.set_menu(self._menu)
             self._indicator.set_status(appindicator.STATUS_ACTIVE)
 
+        self._metastore = None
         self._manager = None
 
     @property
+    def metastore(self):
+        if self._metastore is None:
+            self._metastore = MetaStore(self._env)
+        return self._metastore
+
+    @property
     def manager(self):
         if self._manager is None:
-            self._manager = ImportManager(self._on_signal, self._dbname)
+            self._manager = ImportManager(
+                self.metastore.get_env(), self._on_signal
+            )
             self._manager.start()
         return self._manager
 
@@ -133,6 +140,8 @@
             method(*args)
 
     def _on_timer(self):
+        if self._manager is None:
+            return
         text = _('File %d of %d') % self._manager.get_batch_progress()
         self._current_label.set_text(text)
         self._indicator.set_menu(self._menu)
@@ -143,8 +152,7 @@
     def _on_futon(self, menuitem):
         log.info('Opening dmedia database in Futon..')
         try:
-            store = MetaStore()
-            uri = store.get_auth_uri() + '/_utils'
+            uri = self.metastore.get_auth_uri() + '/_utils'
             check_call(['/usr/bin/xdg-open', uri])
             log.info('Opened Futon')
         except Exception:
@@ -222,10 +230,11 @@
         """
         Kill the dmedia service process.
         """
-        log.info('Killing service')
+        log.info('Killing service...')
         if self._manager is not None:
             self._manager.kill()
         if callable(self._killfunc):
+            log.info('Calling killfunc()')
             self._killfunc()
 
     @dbus.service.method(INTERFACE, in_signature='', out_signature='s')
@@ -288,4 +297,6 @@
         """
         Return list of currently running imports.
         """
-        return self.manager.list_imports()
+        if self._manager is None:
+            return []
+        return self.manager.list_jobs()

=== modified file 'dmedia/tests/__init__.py'
--- dmedia/tests/__init__.py	2011-02-07 03:42:37 +0000
+++ dmedia/tests/__init__.py	2011-02-21 12:01:51 +0000
@@ -24,8 +24,14 @@
 Unit tests for `dmedia` package.
 """
 
+from unittest import TestCase
 from os import path
 
+from desktopcouch.application.platform import find_port
+from desktopcouch.application.local_files import get_oauth_tokens
+
+import dmedia
+
 datadir = path.join(path.dirname(path.abspath(__file__)), 'data')
 sample_mov = path.join(datadir, 'MVI_5751.MOV')
 sample_thm = path.join(datadir, 'MVI_5751.THM')
@@ -33,3 +39,32 @@
 assert path.isdir(datadir)
 assert path.isfile(sample_mov)
 assert path.isfile(sample_thm)
+
+
+class test_functions(TestCase):
+
+    def test_get_env(self):
+        # FIXME: Somehow this test is making gnomekeyring and
+        # ~/.config/desktop-couch/desktop-couchdb.ini contain differnt values
+        return
+        f = dmedia.get_env
+        port = find_port()
+        url = 'http://localhost:%d/' % port
+        oauth = get_oauth_tokens()
+
+        self.assertEqual(
+            f(),
+            {'port': port, 'url': url, 'oauth': oauth, 'dbname': None}
+        )
+        self.assertEqual(
+            f(dbname=None),
+            {'port': port, 'url': url, 'oauth': oauth, 'dbname': None}
+        )
+        self.assertEqual(
+            f(dbname='dmedia'),
+            {'port': port, 'url': url, 'oauth': oauth, 'dbname': 'dmedia'}
+        )
+        self.assertEqual(
+            f(dbname='dmedia_test'),
+            {'port': port, 'url': url, 'oauth': oauth, 'dbname': 'dmedia_test'}
+        )

=== added file 'dmedia/tests/couch.py'
--- dmedia/tests/couch.py	1970-01-01 00:00:00 +0000
+++ dmedia/tests/couch.py	2011-02-21 12:01:51 +0000
@@ -0,0 +1,63 @@
+# Authors:
+#   Jason Gerard DeRose <[e-mail]>
+#
+# dmedia: distributed media library
+# Copyright (C) 2011 Jason Gerard DeRose <[e-mail]>
+#
+# This file is part of `dmedia`.
+#
+# `dmedia` is free software: you can redistribute it and/or modify it under the
+# terms of the GNU Affero General Public License as published by the Free
+# Software Foundation, either version 3 of the License, or (at your option) any
+# later version.
+#
+# `dmedia` is distributed in the hope that it will be useful, but WITHOUT ANY
+# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
+# A PARTICULAR PURPOSE.  See the GNU Affero General Public License for more
+# details.
+#
+# You should have received a copy of the GNU Affero General Public License along
+# with `dmedia`.  If not, see <http://www.gnu.org/licenses/>.
+
+"""
+Base class for CouchDB tests.
+"""
+
+from unittest import TestCase
+
+import couchdb
+
+from dmedia.abstractcouch import get_env, get_couchdb_server
+from dmedia.util import random_id
+from .helpers import TempHome
+
+
+class CouchCase(TestCase):
+    """
+    Base class for tests that talk to CouchDB.
+
+    So that a user's production data doesn't get hosed, all tests are run in the
+    ``"dmedia_test"`` database.
+
+    FIXME: This isn't the best solution, but some changes in desktopcouch in
+    Natty make it difficult for 3rd party apps to use dc-test idioms:
+
+        https://bugs.launchpad.net/desktopcouch/+bug/694909
+    """
+
+    def setUp(self):
+        self.home = TempHome()
+        self.dbname = 'dmedia_test'
+        self.env = get_env(self.dbname)
+        server = get_couchdb_server(self.env)
+        try:
+            del server[self.dbname]
+        except couchdb.ResourceNotFound:
+            pass
+        self.machine_id = random_id()
+        self.env['machine_id'] = self.machine_id
+
+    def tearDown(self):
+        self.home = None
+        self.dbname = None
+        self.env = None

=== modified file 'dmedia/tests/helpers.py'
--- dmedia/tests/helpers.py	2011-02-07 03:42:37 +0000
+++ dmedia/tests/helpers.py	2011-02-21 12:01:51 +0000
@@ -30,8 +30,7 @@
 import tempfile
 import shutil
 from base64 import b32encode, b32decode, b64encode
-from desktopcouch.records.server import  CouchDatabase
-from desktopcouch.records.server_base import NoSuchDatabase
+
 from . import sample_mov, sample_thm
 
 mov_hash = 'ZR765XWSF6S7JQHLUI4GCG5BHGPE252O'
@@ -173,22 +172,3 @@
 
     def __call__(self, signal, args):
         self.messages.append((signal, args))
-
-
-class CouchCase(TestCase):
-    """
-    Base class for tests that need a desktopcouch testing Context.
-    """
-
-    def setUp(self):
-        self.home = TempHome()
-        self.dbname = 'dmedia_test'
-        try:
-            dc = CouchDatabase(self.dbname)
-            del dc._server[self.dbname]
-        except NoSuchDatabase:
-            pass
-
-    def tearDown(self):
-        self.home = None
-        self.dbname = None

=== added file 'dmedia/tests/test_abstractcouch.py'
--- dmedia/tests/test_abstractcouch.py	1970-01-01 00:00:00 +0000
+++ dmedia/tests/test_abstractcouch.py	2011-02-21 12:01:51 +0000
@@ -0,0 +1,207 @@
+# Authors:
+#   Jason Gerard DeRose <[e-mail]>
+#
+# dmedia: distributed media library
+# Copyright (C) 2010 Jason Gerard DeRose <[e-mail]>
+#
+# This file is part of `dmedia`.
+#
+# `dmedia` is free software: you can redistribute it and/or modify it under the
+# terms of the GNU Affero General Public License as published by the Free
+# Software Foundation, either version 3 of the License, or (at your option) any
+# later version.
+#
+# `dmedia` is distributed in the hope that it will be useful, but WITHOUT ANY
+# WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR
+# A PARTICULAR PURPOSE.  See the GNU Affero General Public License for more
+# details.
+#
+# You should have received a copy of the GNU Affero General Public License along
+# with `dmedia`.  If not, see <http://www.gnu.org/licenses/>.
+
+"""
+Unit tests for `dmedia.abstractcouch` module.
+"""
+
+from unittest import TestCase
+
+import couchdb
+from desktopcouch.application.platform import find_port
+from desktopcouch.application.local_files import get_oauth_tokens
+from desktopcouch.records.http import OAuthSession
+
+from dmedia import abstractcouch
+
+from .helpers import raises
+
+class test_functions(TestCase):
+    def tearDown(self):
+        if abstractcouch.OAuthSession is None:
+            abstractcouch.OAuthSession = OAuthSession
+
+    def dc_env(self):
+        """
+        Create an *env* for desktopcouch.
+        """
+        port = find_port()
+        return {
+            'port': port,
+            'url': 'http://localhost:%d/' % port,
+            'oauth': get_oauth_tokens(),
+        }
+
+    def test_get_couchdb_server(self):
+        f = abstractcouch.get_couchdb_server
+
+        # Test with empty env
+        s = f({})
+        self.assertTrue(isinstance(s, couchdb.Server))
+        self.assertEqual(repr(s), "<Server 'http://localhost:5984/'>")
+
+        # Test with only url
+        s = f({'url': 'http://localhost:5984/'})
+        self.assertTrue(isinstance(s, couchdb.Server))
+        self.assertEqual(repr(s), "<Server 'http://localhost:5984/'>")
+
+        # Test with desktopcouch
+        env = self.dc_env()
+        s = f(env)
+        self.assertTrue(isinstance(s, couchdb.Server))
+        self.assertEqual(
+            repr(s),
+            "<Server 'http://localhost:%d/'>" % find_port()
+        )
+
+        # Test when OAuthSession is not imported, oauth is provided
+        abstractcouch.OAuthSession = None
+        e = raises(ValueError, f, env)
+        self.assertEqual(
+            str(e),
+            "provided env['oauth'] but OAuthSession not available: %r" % (env,)
+        )
+
+        # Test when OAuthSession is not imported, oauth not provided
+        s = f({})
+        self.assertTrue(isinstance(s, couchdb.Server))
+        self.assertEqual(repr(s), "<Server 'http://localhost:5984/'>")
+        s = f({'url': 'http://localhost:5984/'})
+        self.assertTrue(isinstance(s, couchdb.Server))
+        self.assertEqual(repr(s), "<Server 'http://localhost:5984/'>")
+
+    def test_get_dmedia_db(self):
+        f = abstractcouch.get_dmedia_db
+
+        # Test when server is not provided
+        env = self.dc_env()
+
+        assert 'dmedia' not in env
+        d = f(env)
+        self.assertTrue(isinstance(d, couchdb.Database))
+        self.assertEqual(repr(d), "<Database 'dmedia'>")
+        self.assertEqual(d.info()['db_name'], 'dmedia')
+
+        env['dbname'] = None
+        d = f(env)
+        self.assertTrue(isinstance(d, couchdb.Database))
+        self.assertEqual(repr(d), "<Database 'dmedia'>")
+        self.assertEqual(d.info()['db_name'], 'dmedia')
+
+        env['dbname'] = 'dmedia'
+        d = f(env)
+        self.assertTrue(isinstance(d, couchdb.Database))
+        self.assertEqual(repr(d), "<Database 'dmedia'>")
+        self.assertEqual(d.info()['db_name'], 'dmedia')
+
+        env['dbname'] = 'dmedia_test'
+        d = f(env)
+        self.assertTrue(isinstance(d, couchdb.Database))
+        self.assertEqual(repr(d), "<Database 'dmedia_test'>")
+        self.assertEqual(d.info()['db_name'], 'dmedia_test')
+
+
+        # Test when server *is* provided
+        env = self.dc_env()
+        server = abstractcouch.get_couchdb_server(env)
+
+        assert 'dmedia' not in env
+        d = f(env, server)
+        self.assertTrue(isinstance(d, couchdb.Database))
+        self.assertEqual(repr(d), "<Database 'dmedia'>")
+        self.assertEqual(d.info()['db_name'], 'dmedia')
+
+        env['dbname'] = None
+        d = f(env, server)
+        self.assertTrue(isinstance(d, couchdb.Database))
+        self.assertEqual(repr(d), "<Database 'dmedia'>")
+        self.assertEqual(d.info()['db_name'], 'dmedia')
+
+        env['dbname', server] = 'dmedia'
+        d = f(env)
+        self.assertTrue(isinstance(d, couchdb.Database))
+        self.assertEqual(repr(d), "<Database 'dmedia'>")
+        self.assertEqual(d.info()['db_name'], 'dmedia')
+
+        env['dbname'] = 'dmedia_test'
+        d = f(env, server)
+        self.assertTrue(isinstance(d, couchdb.Database))
+        self.assertEqual(repr(d), "<Database 'dmedia_test'>")
+        self.assertEqual(d.info()['db_name'], 'dmedia_test')
+
+
+        # Test when server=None is explicitly provided
+        env = self.dc_env()
+
+        assert 'dmedia' not in env
+        d = f(env, server=None)
+        self.assertTrue(isinstance(d, couchdb.Database))
+        self.assertEqual(repr(d), "<Database 'dmedia'>")
+        self.assertEqual(d.info()['db_name'], 'dmedia')
+
+        env['dbname'] = None
+        d = f(env, server=None)
+        self.assertTrue(isinstance(d, couchdb.Database))
+        self.assertEqual(repr(d), "<Database 'dmedia'>")
+        self.assertEqual(d.info()['db_name'], 'dmedia')
+
+        env['dbname'] = 'dmedia'
+        d = f(env, server=None)
+        self.assertTrue(isinstance(d, couchdb.Database))
+        self.assertEqual(repr(d), "<Database 'dmedia'>")
+        self.assertEqual(d.info()['db_name'], 'dmedia')
+
+        env['dbname'] = 'dmedia_test'
+        d = f(env, server=None)
+        self.assertTrue(isinstance(d, couchdb.Database))
+        self.assertEqual(repr(d), "<Database 'dmedia_test'>")
+        self.assertEqual(d.info()['db_name'], 'dmedia_test')
+
+    def test_get_env(self):
+        f = abstractcouch.get_env
+        port = find_port()
+        url = 'http://localhost:%d/' % port
+        oauth = get_oauth_tokens()
+
+        # Test when OAuthSession is available
+        self.assertEqual(
+            f(),
+            {'port': port, 'url': url, 'oauth': oauth, 'dbname': None}
+        )
+        self.assertEqual(
+            f(dbname=None),
+            {'port': port, 'url': url, 'oauth': oauth, 'dbname': None}
+        )
+        self.assertEqual(
+            f(dbname='dmedia'),
+            {'port': port, 'url': url, 'oauth': oauth, 'dbname': 'dmedia'}
+        )
+        self.assertEqual(
+            f(dbname='dmedia_test'),
+            {'port': port, 'url': url, 'oauth': oauth, 'dbname': 'dmedia_test'}
+        )
+
+        # Test when OAuthSession is *not* available
+        abstractcouch.OAuthSession = None
+        self.assertEqual(f(), {'dbname': None})
+        self.assertEqual(f(dbname=None), {'dbname': None})
+        self.assertEqual(f(dbname='dmedia'), {'dbname': 'dmedia'})
+        self.assertEqual(f(dbname='dmedia_test'), {'dbname': 'dmedia_test'})

=== modified file 'dmedia/tests/test_client.py'
--- dmedia/tests/test_client.py	2011-01-30 21:17:48 +0000
+++ dmedia/tests/test_client.py	2011-02-21 12:01:51 +0000
@@ -27,15 +27,18 @@
 from os import path
 from subprocess import Popen
 import time
+
 import dbus
 from dbus.proxies import ProxyObject
 import gobject
+
 import dmedia
 from dmedia import client, service
 from dmedia.constants import VIDEO, AUDIO, IMAGE, EXTENSIONS
-from .helpers import CouchCase, TempDir, random_bus, prep_import_source
+from .helpers import TempDir, random_bus, prep_import_source
 from .helpers import sample_mov, sample_thm
 from .helpers import mov_hash, thm_hash
+from .couch import CouchCase
 
 
 tree = path.dirname(path.dirname(path.abspath(dmedia.__file__)))

=== modified file 'dmedia/tests/test_importer.py'
--- dmedia/tests/test_importer.py	2011-02-18 21:15:03 +0000
+++ dmedia/tests/test_importer.py	2011-02-21 12:01:51 +0000
@@ -24,6 +24,7 @@
 Unit tests for `dmedia.importer` module.
 """
 
+from unittest import TestCase
 import os
 from os import path
 import hashlib
@@ -31,21 +32,22 @@
 import shutil
 import time
 from base64 import b32decode, b32encode, b64encode
-from unittest import TestCase
 from multiprocessing import current_process
-from .helpers import CouchCase, TempDir, TempHome, raises
+
+import couchdb
+
+from dmedia.errors import AmbiguousPath
+from dmedia.filestore import FileStore
+from dmedia.util import random_id
+from dmedia import importer, schema
+from dmedia.abstractcouch import get_env, get_dmedia_db
+from .helpers import TempDir, TempHome, raises
 from .helpers import DummyQueue, DummyCallback, prep_import_source
 from .helpers import sample_mov, sample_thm
 from .helpers import mov_hash, mov_leaves, mov_att, mov_qid
 from .helpers import thm_hash, thm_leaves, thm_qid
-from dmedia.errors import AmbiguousPath
-from dmedia.filestore import FileStore
-from dmedia.metastore import MetaStore
-from dmedia.util import random_id
-from dmedia import importer, schema
+from .couch import CouchCase
 
-import desktopcouch
-from desktopcouch.stop_local_couchdb import stop_couchdb
 
 
 letters = 'gihdwaqoebxtcklrnsmjufyvpz'
@@ -267,36 +269,138 @@
         )
 
 
-class test_Importer(CouchCase):
-    klass = importer.Importer
-    batch_id = 'YKGHY6H5RVCDNMUBL4NLP6AU'
-
-    def new(self, base, extract=False):
-        return self.klass(self.batch_id, base, extract, dbname=self.dbname)
+class test_ImportWorker(CouchCase):
+    klass = importer.ImportWorker
+
+    def setUp(self):
+        super(test_ImportWorker, self).setUp()
+        self.batch_id = random_id()
+        self.env['batch_id'] = self.batch_id
+        self.q = DummyQueue()
+        self.pid = current_process().pid
+        self.tmp = TempDir()
+
+    def tearDown(self):
+        super(test_ImportWorker, self).tearDown()
+        self.q = None
+        self.pid = None
+        self.tmp = None
+
+    def new(self, base=None, extract=False):
+        base = (self.tmp.path if base is None else base)
+        return self.klass(self.env, self.q, base, (base, extract))
 
     def test_init(self):
         tmp = TempDir()
         inst = self.new(tmp.path, True)
-        self.assertEqual(inst.batch_id, self.batch_id)
+        self.assertEqual(inst.env, self.env)
         self.assertEqual(inst.base, tmp.path)
         self.assertTrue(inst.extract is True)
+
+        self.assertTrue(isinstance(inst.server, couchdb.Server))
+        self.assertTrue(isinstance(inst.db, couchdb.Database))
+
         self.assertEqual(inst.home, self.home.path)
         self.assertTrue(isinstance(inst.filestore, FileStore))
         self.assertEqual(inst.filestore.base, self.home.join('.dmedia'))
-        self.assertTrue(isinstance(inst.metastore, MetaStore))
 
         # Test with extract = False
         inst = self.new(tmp.path, False)
         self.assertTrue(inst.extract is False)
 
+    def test_run(self):
+        q = DummyQueue()
+        pid = current_process().pid
+        tmp = TempDir()
+        base = tmp.path
+        inst = self.klass(self.env, q, base, (base, False))
+
+        src1 = tmp.copy(sample_mov, 'DCIM', '100EOS5D2', 'MVI_5751.MOV')
+        dup1 = tmp.copy(sample_mov, 'DCIM', '100EOS5D2', 'MVI_5752.MOV')
+        src2 = tmp.copy(sample_thm, 'DCIM', '100EOS5D2', 'MVI_5751.THM')
+
+        mov_size = path.getsize(sample_mov)
+        thm_size = path.getsize(sample_thm)
+
+        inst.run()
+
+        self.assertEqual(len(q.messages), 6)
+        _id = q.messages[0]['args'][1]
+        self.assertEqual(len(_id), 24)
+        self.assertEqual(
+            q.messages[0],
+            dict(
+                signal='started',
+                args=(base, _id),
+                worker='ImportWorker',
+                pid=pid,
+            )
+        )
+        self.assertEqual(
+            q.messages[1],
+            dict(
+                signal='count',
+                args=(base, _id, 3),
+                worker='ImportWorker',
+                pid=pid,
+            )
+        )
+        self.assertEqual(q.messages[2],
+            dict(
+                signal='progress',
+                args=(base, _id, 1, 3,
+                    dict(action='imported', src=src1)
+                ),
+                worker='ImportWorker',
+                pid=pid,
+            )
+        )
+        self.assertEqual(q.messages[3],
+            dict(
+                signal='progress',
+                args=(base, _id, 2, 3,
+                    dict(action='imported', src=src2)
+                ),
+                worker='ImportWorker',
+                pid=pid,
+            )
+        )
+        self.assertEqual(q.messages[4],
+            dict(
+                signal='progress',
+                args=(base, _id, 3, 3,
+                    dict(action='skipped', src=dup1)
+                ),
+                worker='ImportWorker',
+                pid=pid,
+            )
+        )
+        self.assertEqual(
+            q.messages[5],
+            dict(
+                signal='finished',
+                args=(base, _id,
+                    dict(
+                        considered={'count': 3, 'bytes': (mov_size*2 + thm_size)},
+                        imported={'count': 2, 'bytes': (mov_size + thm_size)},
+                        skipped={'count': 1, 'bytes': mov_size},
+                        empty={'count': 0, 'bytes': 0},
+                        error={'count': 0, 'bytes': 0},
+                    ),
+                ),
+                worker='ImportWorker',
+                pid=pid,
+            )
+        )
+
     def test_start(self):
         tmp = TempDir()
         inst = self.new(tmp.path)
         self.assertTrue(inst.doc is None)
         _id = inst.start()
         self.assertEqual(len(_id), 24)
-        store = MetaStore(dbname=self.dbname)
-        self.assertEqual(inst.doc, store.db[_id])
+        db = get_dmedia_db(self.env)
+        self.assertEqual(inst.doc, db[_id])
         self.assertEqual(
             set(inst.doc),
             set([
@@ -312,10 +416,7 @@
             ])
         )
         self.assertEqual(inst.doc['batch_id'], self.batch_id)
-        self.assertEqual(
-            inst.doc['machine_id'],
-            inst.metastore.machine_id
-        )
+        self.assertEqual(inst.doc['machine_id'], self.machine_id)
         self.assertEqual(inst.doc['base'], tmp.path)
         self.assertEqual(
             inst.doc['log'],
@@ -706,103 +807,16 @@
         )
 
 
-class test_ImportWorker(CouchCase):
-    klass = importer.ImportWorker
-
-    def test_run(self):
-        q = DummyQueue()
-        pid = current_process().pid
-
-        tmp = TempDir()
-        batch_id = 'YKGHY6H5RVCDNMUBL4NLP6AU'
-        base = tmp.path
-        inst = self.klass(q, base, (batch_id, base, False, self.dbname))
-
-        src1 = tmp.copy(sample_mov, 'DCIM', '100EOS5D2', 'MVI_5751.MOV')
-        dup1 = tmp.copy(sample_mov, 'DCIM', '100EOS5D2', 'MVI_5752.MOV')
-        src2 = tmp.copy(sample_thm, 'DCIM', '100EOS5D2', 'MVI_5751.THM')
-
-        mov_size = path.getsize(sample_mov)
-        thm_size = path.getsize(sample_thm)
-
-        inst.run()
-
-        self.assertEqual(len(q.messages), 6)
-        _id = q.messages[0]['args'][1]
-        self.assertEqual(len(_id), 24)
-        self.assertEqual(
-            q.messages[0],
-            dict(
-                signal='started',
-                args=(base, _id),
-                worker='ImportWorker',
-                pid=pid,
-            )
-        )
-        self.assertEqual(
-            q.messages[1],
-            dict(
-                signal='count',
-                args=(base, _id, 3),
-                worker='ImportWorker',
-                pid=pid,
-            )
-        )
-        self.assertEqual(q.messages[2],
-            dict(
-                signal='progress',
-                args=(base, _id, 1, 3,
-                    dict(action='imported', src=src1)
-                ),
-                worker='ImportWorker',
-                pid=pid,
-            )
-        )
-        self.assertEqual(q.messages[3],
-            dict(
-                signal='progress',
-                args=(base, _id, 2, 3,
-                    dict(action='imported', src=src2)
-                ),
-                worker='ImportWorker',
-                pid=pid,
-            )
-        )
-        self.assertEqual(q.messages[4],
-            dict(
-                signal='progress',
-                args=(base, _id, 3, 3,
-                    dict(action='skipped', src=dup1)
-                ),
-                worker='ImportWorker',
-                pid=pid,
-            )
-        )
-        self.assertEqual(
-            q.messages[5],
-            dict(
-                signal='finished',
-                args=(base, _id,
-                    dict(
-                        considered={'count': 3, 'bytes': (mov_size*2 + thm_size)},
-                        imported={'count': 2, 'bytes': (mov_size + thm_size)},
-                        skipped={'count': 1, 'bytes': mov_size},
-                        empty={'count': 0, 'bytes': 0},
-                        error={'count': 0, 'bytes': 0},
-                    ),
-                ),
-                worker='ImportWorker',
-                pid=pid,
-            )
-        )
-
-
 class test_ImportManager(CouchCase):
     klass = importer.ImportManager
 
+    def new(self, callback=None):
+        return self.klass(self.env, callback)
+
+
     def test_start_batch(self):
         callback = DummyCallback()
-        inst = self.klass(callback, self.dbname)
+        inst = self.new(callback)
 
         # Test that batch cannot be started when there are active workers:
         inst._workers['foo'] = 'bar'
@@ -816,7 +830,6 @@
         self.assertEqual(inst._completed, 0)
         self.assertEqual(inst._total, 0)
         batch = inst.doc
-        batch_id = batch['_id']
         self.assertTrue(isinstance(batch, dict))
         self.assertEqual(
             set(batch),
@@ -832,12 +845,12 @@
         )
         self.assertEqual(batch['type'], 'dmedia/batch')
         self.assertEqual(batch['imports'], [])
-        self.assertEqual(batch['machine_id'], inst.metastore.machine_id)
+        self.assertEqual(batch['machine_id'], self.machine_id)
         self.assertEqual(inst.db[batch['_id']], batch)
         self.assertEqual(
             callback.messages,
             [
-                ('BatchStarted', (batch_id,)),
+                ('BatchStarted', (batch['_id'],)),
             ]
         )
 
@@ -846,7 +859,7 @@
 
     def test_finish_batch(self):
         callback = DummyCallback()
-        inst = self.klass(callback, self.dbname)
+        inst = self.new(callback)
         batch_id = random_id()
         inst.doc = dict(
             _id=batch_id,
@@ -892,7 +905,7 @@
 
     def test_on_error(self):
         callback = DummyCallback()
-        inst = self.klass(callback, self.dbname)
+        inst = self.new(callback)
 
         # Make sure it works when doc is None:
         inst.on_error('foo', 'IOError', 'nope')
@@ -919,9 +932,21 @@
             ]
         )
 
+    def test_get_worker_env(self):
+        batch_id = random_id()
+        inst = self.new()
+        env = dict(self.env)
+        assert 'batch_id' not in env
+        inst.doc = {'_id': batch_id}
+        env['batch_id'] = batch_id
+        self.assertEqual(
+            inst.get_worker_env('ImportWorker', 'a key', ('some', 'args')),
+            env
+        )
+
     def test_on_started(self):
         callback = DummyCallback()
-        inst = self.klass(callback, self.dbname)
+        inst = self.new(callback)
         self.assertEqual(callback.messages, [])
         inst._start_batch()
         batch_id = inst.doc['_id']
@@ -960,7 +985,7 @@
 
     def test_on_count(self):
         callback = DummyCallback()
-        inst = self.klass(callback, self.dbname)
+        inst = self.new(callback)
         self.assertEqual(callback.messages, [])
 
         one = TempDir()
@@ -990,7 +1015,7 @@
 
     def test_on_progress(self):
         callback = DummyCallback()
-        inst = self.klass(callback, self.dbname)
+        inst = self.new(callback)
         self.assertEqual(callback.messages, [])
 
         one = TempDir()
@@ -1030,7 +1055,7 @@
 
     def test_on_finished(self):
         callback = DummyCallback()
-        inst = self.klass(callback, self.dbname)
+        inst = self.new(callback)
         batch_id = random_id()
         inst.doc = dict(
             _id=batch_id,
@@ -1116,7 +1141,7 @@
         )
 
     def test_get_batch_progress(self):
-        inst = self.klass(dbname=self.dbname)
+        inst = self.new()
         self.assertEqual(inst.get_batch_progress(), (0, 0))
         inst._total = 18
         self.assertEqual(inst.get_batch_progress(), (0, 18))
@@ -1128,7 +1153,7 @@
 
     def test_start_import(self):
         callback = DummyCallback()
-        inst = self.klass(callback, self.dbname)
+        inst = self.new(callback)
         self.assertTrue(inst.start())
 
         tmp = TempDir()
@@ -1208,13 +1233,3 @@
                 )
             )
         )
-
-    def test_list_imports(self):
-        inst = self.klass(dbname=self.dbname)
-        self.assertEqual(inst.list_imports(), [])
-        inst._workers.update(
-            dict(foo=None, bar=None, baz=None)
-        )
-        self.assertEqual(inst.list_imports(), ['bar', 'baz', 'foo'])
-        inst._workers.clear()
-        self.assertEqual(inst.list_imports(), [])

=== modified file 'dmedia/tests/test_metastore.py'
--- dmedia/tests/test_metastore.py	2011-02-18 19:22:23 +0000
+++ dmedia/tests/test_metastore.py	2011-02-21 12:01:51 +0000
@@ -24,19 +24,16 @@
 """
 
 from unittest import TestCase
+import os
+import shutil
 import socket
 import platform
-from helpers import CouchCase, TempDir, TempHome
+
+import couchdb
+
 from dmedia import metastore
-import couchdb
-from desktopcouch.records.server import  CouchDatabase
-from desktopcouch.records.record import  Record
-from desktopcouch.local_files import Context
-from desktopcouch.stop_local_couchdb import stop_couchdb
-import desktopcouch
-import tempfile
-import os
-import shutil
+from .helpers import TempDir, TempHome
+from .couch import CouchCase
 
 
 class test_functions(TestCase):
@@ -90,13 +87,13 @@
     klass = metastore.MetaStore
 
     def new(self):
-        return self.klass(self.dbname)
+        return self.klass(self.env)
 
     def test_init(self):
         inst = self.new()
-        self.assertEqual(inst.dbname, self.dbname)
-        self.assertEqual(isinstance(inst.desktop, CouchDatabase), True)
-        self.assertEqual(isinstance(inst.server, couchdb.Server), True)
+        self.assertEqual(inst.env, self.env)
+        self.assertTrue(isinstance(inst.server, couchdb.Server))
+        self.assertTrue(isinstance(inst.db, couchdb.Database))
 
     def update(self):
         inst = self.new()

=== modified file 'dmedia/tests/test_service.py'
--- dmedia/tests/test_service.py	2011-01-30 21:17:48 +0000
+++ dmedia/tests/test_service.py	2011-02-21 12:01:51 +0000
@@ -24,8 +24,10 @@
 """
 
 from unittest import TestCase
-from helpers import CouchCase, TempDir, random_bus
+
 from dmedia import service, importer
+from .helpers import TempDir, random_bus
+from .couch import CouchCase
 
 
 class test_DMedia(CouchCase):
@@ -33,11 +35,11 @@
 
     def test_init(self):
         bus = random_bus()
+        self.env['bus'] = bus
+        self.env['no_gui'] = True
         def kill():
             pass
-        inst = self.klass(
-            killfunc=kill, bus=bus, dbname=self.dbname, no_gui=True
-        )
+        inst = self.klass(self.env, killfunc=kill)
         self.assertTrue(inst._killfunc is kill)
         self.assertTrue(inst._bus is bus)
         self.assertTrue(inst._dbname is self.dbname)
@@ -48,4 +50,3 @@
         self.assertTrue(inst._manager is m)
         self.assertTrue(isinstance(m, importer.ImportManager))
         self.assertEqual(m._callback, inst._on_signal)
-        self.assertTrue(m._dbname is self.dbname)

=== modified file 'dmedia/tests/test_workers.py'
--- dmedia/tests/test_workers.py	2010-12-29 13:58:00 +0000
+++ dmedia/tests/test_workers.py	2011-02-21 12:01:51 +0000
@@ -30,8 +30,12 @@
 import multiprocessing.queues
 from multiprocessing import current_process
 import threading
+
+import couchdb
+
 from dmedia import workers
 from .helpers import raises, DummyQueue, DummyCallback
+from .couch import CouchCase
 
 
 class test_functions(TestCase):
@@ -109,7 +113,8 @@
 
         # Test with unknown worker name
         q = DummyQueue()
-        f(q, 'ImportFiles', 'the key', ('foo', 'bar'))
+        env = {'foo': 'bar'}
+        f('ImportFiles', env, q, 'the key', ('foo', 'bar'))
 
         self.assertEqual(
             q.messages,
@@ -136,7 +141,8 @@
         workers.register(ImportFiles)
 
         q = DummyQueue()
-        f(q, 'ImportFiles', 'the key', ('hello', 'world'))
+        env = {'foo': 'bar'}
+        f('ImportFiles', env, q, 'the key', ('hello', 'world'))
         self.assertEqual(
             q.messages,
             [
@@ -160,10 +166,12 @@
     klass = workers.Worker
 
     def test_init(self):
+        env = {'foo': 'bar'}
         q = DummyQueue()
         key = 'the key'
         args = ('foo', 'bar')
-        inst = self.klass(q, key, args)
+        inst = self.klass(env, q, key, args)
+        self.assertTrue(inst.env is env)
         self.assertTrue(inst.q is q)
         self.assertTrue(inst.key is key)
         self.assertTrue(inst.args is args)
@@ -171,9 +179,10 @@
         self.assertEqual(inst.name, 'Worker')
 
     def test_emit(self):
+        env = {'foo': 'bar'}
         q = DummyQueue()
         args = ('foo', 'bar')
-        inst = self.klass(q, 'akey', args)
+        inst = self.klass(env, q, 'akey', args)
         pid = current_process().pid
 
         self.assertEqual(q.messages, [])
@@ -206,6 +215,7 @@
         self.assertEqual(q.messages, [one, two, three])
 
     def test_run(self):
+        env = {'foo': 'bar'}
         q = DummyQueue()
         args = ('foo', 'bar')
         pid = current_process().pid
@@ -214,7 +224,7 @@
             def execute(self, one, two):
                 self.emit('Hello', '%s and %s' % (one, two))
 
-        inst = do_something(q, 'key', args)
+        inst = do_something(env, q, 'key', args)
         inst.run()
         self.assertEqual(q.messages[0],
             dict(
@@ -226,20 +236,40 @@
         )
 
     def test_execute(self):
+        env = {'foo': 'bar'}
         q = DummyQueue()
         args = ('foo', 'bar')
-        inst = self.klass(q, 'key', args)
+        inst = self.klass(env, q, 'key', args)
 
         e = raises(NotImplementedError, inst.execute)
         self.assertEqual(str(e), 'Worker.execute()')
 
         class do_something(self.klass):
             pass
-        inst = do_something(q, 'key', args)
+        inst = do_something(env, q, 'key', args)
         e = raises(NotImplementedError, inst.execute)
         self.assertEqual(str(e), 'do_something.execute()')
 
 
+class test_CouchWorker(CouchCase):
+    klass = workers.CouchWorker
+
+    def test_init(self):
+        q = DummyQueue()
+        key = 'a key'
+        args = ('some', 'args')
+        inst = self.klass(self.env, q, key, args)
+        self.assertTrue(inst.env is self.env)
+        self.assertTrue(inst.q is q)
+        self.assertTrue(inst.key is key)
+        self.assertTrue(inst.args is args)
+        self.assertTrue(isinstance(inst.server, couchdb.Server))
+        self.assertTrue(isinstance(inst.db, couchdb.Database))
+
+
+
+
+
 def infinite():
     while True:
         time.sleep(1)
@@ -254,8 +284,8 @@
 
 
 class ExampleWorker(workers.Worker):
-    def execute(self, run_infinitely=True):
-        if run_infinitely:
+    def execute(self, *args):
+        if self.env.get('infinite', True):
             infinite()
         else:
             time.sleep(1)
@@ -269,18 +299,21 @@
         workers.register(ExampleWorker)
 
     def test_init(self):
+        env = {'foo': 'bar'}
         # Test with non-callable callback:
-        e = raises(TypeError, self.klass, 'foo')
+        e = raises(TypeError, self.klass, env, 'foo')
         self.assertEqual(str(e), "callback must be callable; got 'foo'")
 
         # Test that callback default is None:
-        inst = self.klass()
+        inst = self.klass(env)
+        self.assertTrue(inst.env is env)
         self.assertTrue(inst._callback is None)
 
         # Test with a callable:
         def foo():
             pass
-        inst = self.klass(callback=foo)
+        inst = self.klass(env, callback=foo)
+        self.assertTrue(inst.env is env)
         self.assertTrue(inst._callback is foo)
         self.assertTrue(inst._running is False)
         self.assertEqual(inst._workers, {})
@@ -294,7 +327,8 @@
                 assert self._call is None
                 self._call = arg1 + arg2
 
-        inst = Example()
+        env = {'foo': 'bar'}
+        inst = Example(env = {'foo': 'bar'})
         msg = dict(signal='stuff', args=('foo', 'bar'))
         inst._process_message(msg)
         self.assertEqual(inst._call, 'foobar')
@@ -304,7 +338,8 @@
         self.assertEqual(str(e), "'Example' object has no attribute 'on_nope'")
 
     def test_on_terminate(self):
-        inst = self.klass()
+        env = {'foo': 'bar'}
+        inst = self.klass(env)
         e = raises(KeyError, inst.on_terminate, 'foo')
         p = multiprocessing.Process(target=time.sleep, args=(1,))
         p.daemon = True
@@ -316,7 +351,8 @@
         self.assertEqual(inst._workers, {})
 
     def test_start(self):
-        inst = self.klass()
+        env = {'foo': 'bar'}
+        inst = self.klass(env)
 
         # Test that start() returns False when already running:
         inst._running = True
@@ -336,7 +372,8 @@
         inst._thread.join()
 
     def test_kill(self):
-        inst = self.klass()
+        env = {'foo': 'bar'}
+        inst = self.klass(env)
 
         # Test that kill() returns False when not running:
         self.assertTrue(inst.kill() is False)
@@ -357,7 +394,8 @@
         self.assertEqual(inst._workers, {})
 
     def test_kill_job(self):
-        inst = self.klass()
+        env = {'foo': 'bar'}
+        inst = self.klass(env)
 
         # Test that kill_job() returns False when no such job exists:
         self.assertTrue(inst.kill_job('foo') is False)
@@ -372,31 +410,55 @@
         # Again test that kill_job() returns False when no such job exists:
         self.assertTrue(inst.kill_job('foo') is False)
 
-    def test_do(self):
-        inst = self.klass()
+    def test_start_job(self):
+        env = {'foo': 'bar'}
+        inst = self.klass(env)
 
         # Test that False is returned when key already exists:
         inst._workers['foo'] = 'bar'
-        self.assertTrue(inst.do('ExampleWorker', 'foo') is False)
+        self.assertTrue(inst.start_job('ExampleWorker', 'foo') is False)
 
-        # Test creating a process
+        # Test creating a process with no args
         inst._workers.clear()
-        self.assertTrue(inst.do('ExampleWorker', 'foo', ) is True)
+        self.assertTrue(inst.start_job('ExampleWorker', 'foo') is True)
         self.assertEqual(list(inst._workers), ['foo'])
         p = inst._workers['foo']
         self.assertTrue(isinstance(p, multiprocessing.Process))
         self.assertTrue(p.daemon)
         self.assertTrue(p.is_alive())
+        self.assertEqual(
+            p._args,
+            ('ExampleWorker', inst.env, inst._q, 'foo', tuple())
+        )
+        self.assertEqual(p._kwargs, {})
+        p.terminate()
+        p.join()
+
+        # Test creating a process *with* args
+        self.assertTrue(
+            inst.start_job('ExampleWorker', 'bar', 'some', 'args') is True
+        )
+        self.assertEqual(sorted(inst._workers), ['bar', 'foo'])
+        p = inst._workers['bar']
+        self.assertTrue(isinstance(p, multiprocessing.Process))
+        self.assertTrue(p.daemon)
+        self.assertTrue(p.is_alive())
+        self.assertEqual(
+            p._args,
+            ('ExampleWorker', inst.env, inst._q, 'bar', ('some', 'args'))
+        )
+        self.assertEqual(p._kwargs, {})
         p.terminate()
         p.join()
 
     def test_emit(self):
+        env = {'foo': 'bar'}
         # Test with no callback
-        inst = self.klass()
+        inst = self.klass(env)
         inst.emit('ImportStarted', 'foo', 'bar')
 
         callback = DummyCallback()
-        inst = self.klass(callback)
+        inst = self.klass(env, callback)
         inst.emit('ImportStarted', 'foo', 'bar')
         inst.emit('NoArgs')
         inst.emit('OneArg', 'baz')
@@ -408,3 +470,33 @@
                 ('OneArg', ('baz',)),
             ]
         )
+
+    def test_list_jobs(self):
+        env = {'foo': 'bar'}
+        inst = self.klass(env)
+        self.assertEqual(inst.list_jobs(), [])
+        inst._workers.update(
+            dict(foo=None, bar=None, baz=None)
+        )
+        self.assertEqual(inst.list_jobs(), ['bar', 'baz', 'foo'])
+        inst._workers.clear()
+        self.assertEqual(inst.list_jobs(), [])
+
+
+class test_CouchManager(CouchCase):
+    klass = workers.CouchManager
+
+    def test_init(self):
+        inst = self.klass(self.env)
+        self.assertTrue(inst.env is self.env)
+        self.assertTrue(inst._callback is None)
+        self.assertTrue(isinstance(inst.server, couchdb.Server))
+        self.assertTrue(isinstance(inst.db, couchdb.Database))
+
+        def func():
+            pass
+        inst = self.klass(self.env, func)
+        self.assertTrue(inst.env is self.env)
+        self.assertTrue(inst._callback, func)
+        self.assertTrue(isinstance(inst.server, couchdb.Server))
+        self.assertTrue(isinstance(inst.db, couchdb.Database))

=== modified file 'dmedia/workers.py'
--- dmedia/workers.py	2011-02-19 04:38:12 +0000
+++ dmedia/workers.py	2011-02-21 12:01:51 +0000
@@ -28,10 +28,12 @@
 from threading import Thread, Lock
 from Queue import Empty
 import logging
+
 from .constants import TYPE_ERROR
+from .abstractcouch import get_couchdb_server, get_dmedia_db
+
 
 log = logging.getLogger()
-
 _workers = {}
 
 
@@ -79,14 +81,24 @@
     return exception.__name__
 
 
-def dispatch(q, worker, key, args):
+def dispatch(worker, env, q, key, args):
+    """
+    Dispatch a worker in this proccess.
+
+    :param worker: name of worker class, eg ``'ImportWorker'``
+    :param env: a ``dict`` containing run-time information like the CouchDB URL
+    :param q: a ``multiprocessing.Queue`` or similar
+    :param key: a key to uniquely identify this worker among active workers
+        controlled by the `Manager` that launched this worker
+    :param args: arguments to be passed to `Worker.run()`
+    """
     pid = current_process().pid
-    log.debug('dispatch in process %d: worker=%r, key=%r, args=%r',
+    log.debug('** dispatch in process %d: worker=%r, key=%r, args=%r',
         pid, worker, key, args
     )
     try:
         klass = _workers[worker]
-        inst = klass(q, key, args)
+        inst = klass(env, q, key, args)
         inst.run()
     except Exception as e:
         log.exception('exception in procces %d, worker=%r', pid)
@@ -106,7 +118,8 @@
 
 
 class Worker(object):
-    def __init__(self, q, key, args):
+    def __init__(self, env, q, key, args):
+        self.env = env
         self.q = q
         self.key = key
         self.args = args
@@ -143,12 +156,20 @@
         )
 
 
+class CouchWorker(Worker):
+    def __init__(self, env, q, key, args):
+        super(CouchWorker, self).__init__(env, q, key, args)
+        self.server = get_couchdb_server(env)
+        self.db = get_dmedia_db(env, self.server)
+
+
 class Manager(object):
-    def __init__(self, callback=None):
+    def __init__(self, env, callback=None):
         if not (callback is None or callable(callback)):
             raise TypeError(
                 'callback must be callable; got %r' % callback
             )
+        self.env = env
         self._callback = callback
         self._running = False
         self._workers = {}
@@ -165,7 +186,7 @@
                 pass
 
     def _process_message(self, msg):
-        log.info('[From %(pid)d] %(signal)s %(args)r', msg)
+        log.info('[from %(worker)s %(pid)d] %(signal)s %(args)r', msg)
         with self._lock:
             signal = msg['signal']
             args = msg['args']
@@ -180,10 +201,10 @@
         log.error('%s %s: %s: %s', self.name, key, exception, message)
 
     def start(self):
-        log.info('Killing %s', self.name)
         with self._lock:
             if self._running:
                 return False
+            log.info('Starting %s', self.name)
             self._running = True
             self._thread = Thread(target=self._signal_thread)
             self._thread.daemon = True
@@ -191,9 +212,9 @@
             return True
 
     def kill(self):
-        log.info('Killing %s', self.name)
         if not self._running:
             return False
+        log.info('Killing %s', self.name)
         self._running = False
         self._thread.join()  # Cleanly shutdown _signal_thread
         with self._lock:
@@ -203,15 +224,24 @@
             self._workers.clear()
             return True
 
-    def do(self, worker, key, *args):
+    def get_worker_env(self, worker, key, args):
+        return dict(self.env)
+
+    def start_job(self, worker, key, *args):
         """
         Start a process identified by *key*, using worker class *name*.
+
+        :param worker: name of worker class, eg ``'ImportWorker'``
+        :param key: a key to uniquely identify new `Worker` among active workers
+            controlled by this `Manager`
+        :param args: arguments to be passed to `Worker.run()`
         """
         if key in self._workers:
             return False
+        env = self.get_worker_env(worker, key, args)
         p = multiprocessing.Process(
             target=dispatch,
-            args=(self._q, worker, key, args),
+            args=(worker, env, self._q, key, args),
         )
         p.daemon = True
         self._workers[key] = p
@@ -227,6 +257,9 @@
             p.join()
             return True
 
+    def list_jobs(self):
+        return sorted(self._workers)
+
     def emit(self, signal, *args):
         """
         Emit a signal to higher-level code.
@@ -234,3 +267,10 @@
         if self._callback is None:
             return
         self._callback(signal, args)
+
+
+class CouchManager(Manager):
+    def __init__(self, env, callback=None):
+        super(CouchManager, self).__init__(env, callback)
+        self.server = get_couchdb_server(env)
+        self.db = get_dmedia_db(env, self.server)

=== modified file 'setup.py'
--- setup.py	2011-02-01 06:09:34 +0000
+++ setup.py	2011-02-21 12:01:51 +0000
@@ -86,6 +86,11 @@
                         break
 
     def run(self):
+        from dbus.mainloop.glib import DBusGMainLoop
+        import gobject
+        DBusGMainLoop(set_as_default=True)
+        gobject.threads_init()
+
         pynames = tuple(self._pynames_iter())
 
         # Add unit-tests: