Merge lp:~benji/charmworld/even-more-bundles into lp:~juju-jitsu/charmworld/trunk

[Benji York]

Make the queue worker process the basket queue.

Generalize the queue worker to process the basket queue as well as the
charm queue.

[Aaron Bentley (abentley)]

This all looks good. The only thing I see is a style issue. In several places, you have multiple imports on a single line, e.g. "from charmworld.jobs.config import CHARM_QUEUE, BASKET_QUEUE, settings". Charmworld follows Launchpad style, and puts these on multiple lines to reduce merge conflicts. e.g.:

from charmworld.jobs.config import (
    BASKET_QUEUE,
    CHARM_QUEUE,
    settings,
)

review: Approve

[lp:~benji/charmworld/even-more-bundles]

331. By Benji York on 2013-07-22

tweak import style

332. By Benji York on 2013-07-22

more import style tweaking

=== modified file 'charmworld/jobs/ingest.py'
--- charmworld/jobs/ingest.py	2013-07-18 21:49:30 +0000
+++ charmworld/jobs/ingest.py	2013-07-22 15:53:24 +0000
@@ -41,6 +41,7 @@
     getdb,
     getfs,
     options_to_storage,
+    store_bundles,
 )
 from charmworld.search import (
     ElasticSearchClient,
@@ -277,6 +278,7 @@
         fs = getfs(self.db, collection='hashed-files')
         self.decorate_basket(payload, fs)
         self.db.baskets.save(payload)
+        self.store_bundles(fs, payload, self.db.bundles)
 
     @staticmethod
     def construct_id(basket_data, revno):
@@ -290,8 +292,13 @@
         tree = branch.repository.revision_tree(basket_data['commit'])
         basket_data['_id'] = self.construct_id(basket_data, revno)
         basket_data['file_hashes'] = slurp_files(fs, tree)
-        #Basket(basket_data, fs).get_file_contents('')
-        #basket = get_basket(fs, basket_data['file_hashes'])
+
+    @staticmethod
+    def store_bundles(fs, basket_data, collection):
+        hashes = basket_data['file_hashes']
+        deployer_config_bytes = fs.get(hashes['bundles.yaml'])
+        deployer_config = yaml.safe_load(deployer_config_bytes)
+        store_bundles(collection, deployer_config, basket_data['_id'])
 
 
 def slurp_files(fs, tree):
@@ -529,7 +536,7 @@
         cfile = CharmFileSet.get_by_id(
             fs, files[metadata_file]['fileid'])
         try:
-            metadata = quote_yaml(yaml.load(cfile.read()))
+            metadata = quote_yaml(yaml.safe_load(cfile.read()))
         except Exception, exc:
             raise IngestError(
                 'Invalid charm metadata %s: %s' % (
@@ -543,7 +550,7 @@
         config_raw = cfile.read()
 
         try:
-            config_yaml = yaml.load(config_raw)
+            config_yaml = yaml.safe_load(config_raw)
             if 'options' in config_yaml:
                 config_yaml['options'] = options_to_storage(
                     config_yaml['options'])

=== modified file 'charmworld/jobs/lp.py'
--- charmworld/jobs/lp.py	2013-07-19 12:51:01 +0000
+++ charmworld/jobs/lp.py	2013-07-22 15:53:24 +0000
@@ -131,7 +131,7 @@
     try:
         lease_time = int(settings['script_lease_time']) * 60
         with lock('ingest-queue', lease_time, db, log):
-            queue = get_queue(CHARM_QUEUE)
+            queue = get_queue(db, CHARM_QUEUE)
             queue_size = queue.size()
             queue.clear()
             log.info("Dequeued %s charms", queue_size)
@@ -139,6 +139,7 @@
         log.warn(str(error))
 
 
+# XXX bug=1130732: This function does not appear to be tested.
 def main():
     configure_logging()
     parser = argparse.ArgumentParser()
@@ -164,8 +165,8 @@
     try:
         with lock('ingest-queue', int(settings['script_lease_time']) * 60,
                   db, log):
-            charm_queue = get_queue(CHARM_QUEUE)
-            basket_queue = get_queue(BASKET_QUEUE)
+            charm_queue = get_queue(db, CHARM_QUEUE)
+            basket_queue = get_queue(db, BASKET_QUEUE)
             queue_from_branches(db, charm_queue, basket_queue,
                                 charm_import_limit, args.prefix)
     except LockHeld, error:

=== modified file 'charmworld/jobs/supervisord.conf'
--- charmworld/jobs/supervisord.conf	2013-04-01 15:07:06 +0000
+++ charmworld/jobs/supervisord.conf	2013-07-22 15:53:24 +0000
@@ -29,11 +29,9 @@
 ; supervisor.
 
 [program:ingest]
-command=bin/python charmworld/jobs/worker.py ; the program (relative uses PATH, can take args)
-numprocs=1                    ; number of processes copies to start (def 1)
-;directory=/tmp                ; directory to cwd to before exec (def no cwd)
-;autostart=true                ; start at supervisord start (default: true)
-autorestart=unexpected        ; whether/when to restart (default: unexpected)
-startsecs=1                   ; number of secs prog must stay running (def. 1)
-startretries=3                ; max # of serial start failures (default 3)
-exitcodes=0,2                 ; 'expected' exit codes for process (default 0,2)
+command=bin/ingest-queued --run-forever
+numprocs=1
+autorestart=unexpected
+startsecs=1
+startretries=3
+exitcodes=0,2

=== modified file 'charmworld/jobs/tests/test_ingest.py'
--- charmworld/jobs/tests/test_ingest.py	2013-07-18 21:49:30 +0000
+++ charmworld/jobs/tests/test_ingest.py	2013-07-22 15:53:24 +0000
@@ -11,6 +11,7 @@
 import os.path
 import shutil
 import subprocess
+from textwrap import dedent
 
 import bzrlib
 from bzrlib.bzrdir import BzrDir
@@ -237,12 +238,13 @@
         side_effect=checkout_charm_mock
     ) as checkout_branch:
         with bzr_isolation() as working_dir:
-            try:
-                yield checkout_branch, working_dir
-            finally:
-                # Clean up despite exceptions raised by the context manager
-                # client.
-                shutil.rmtree(CHARM_DIR, ignore_errors=True)
+            with patch('charmworld.jobs.ingest.UpdateBundleJob.store_bundles'):
+                try:
+                    yield checkout_branch, working_dir
+                finally:
+                    # Clean up despite exceptions raised by the context
+                    # manager client.
+                    shutil.rmtree(CHARM_DIR, ignore_errors=True)
 
     # TODO complete refactoring of checkout_branch, et al.
 
@@ -368,6 +370,42 @@
         basket_id = UpdateBundleJob.construct_id(payload, 5)
         self.assertEqual('~charmers/mysql-5', basket_id)
 
+    def test_job_can_store_bundles_in_the_db(self):
+        basket_data = factory.get_payload_json(name='wordpress')
+        basket_id = UpdateBundleJob.construct_id(basket_data, 5)
+        bundle_id = '%s/%s' % (basket_id, 'stage')
+        basket_data['_id'] = basket_id
+        job = UpdateBundleJob()
+        job.setup(self.db)
+        DEPLOYER_CONFIG_HASH = '31337'
+        basket_data['file_hashes'] = {
+            'bundles.yaml': DEPLOYER_CONFIG_HASH,
+        }
+        fs = getfs(self.db)
+        deployer_config = dedent("""\
+            stage:
+                series: precise
+                services:
+                    blog:
+                        charm: cs:precise/wordpress
+                        constraints: mem=2
+                        options:
+                            tuning: optimized
+                            engine: apache
+        """)
+        fs.put(deployer_config, _id=DEPLOYER_CONFIG_HASH)
+        job.store_bundles(fs, basket_data, self.db.bundles)
+        self.assertIsNotNone(self.db.bundles.find_one(bundle_id))
+
+    def test_job_run_stores_bundles_in_the_db(self):
+        basket_data = factory.get_payload_json(name='wordpress')
+        job = UpdateBundleJob()
+        job.setup(self.db)
+        with patch.object(job, 'decorate_basket'):
+            with patch.object(job, 'store_bundles') as store_bundles:
+                job.run(basket_data)
+                self.assertTrue(store_bundles.called)
+
 
     # TODO add bundles to index
 

=== modified file 'charmworld/jobs/tests/test_lp.py'
--- charmworld/jobs/tests/test_lp.py	2013-07-19 12:51:01 +0000
+++ charmworld/jobs/tests/test_lp.py	2013-07-22 15:53:24 +0000
@@ -258,8 +258,8 @@
 
     @patch('charmworld.jobs.lp.all_branch_data', all_branch_data_mock)
     def test_queue_from_branches(self):
-        charm_queue = get_queue('test_charm_queue')
-        basket_queue = get_queue('test_basket_queue')
+        charm_queue = get_queue(self.db, 'test_charm_queue')
+        basket_queue = get_queue(self.db, 'test_basket_queue')
         self.addCleanup(basket_queue.clear)
         queue_from_branches(self.db, charm_queue, basket_queue)
         item = charm_queue.next()
@@ -269,7 +269,7 @@
         self.assertEqual({'foo': 'bar'}, item.payload)
 
     def test_dequeue(self):
-        queue = get_queue('test_queue')
+        queue = get_queue(self.db, 'test_queue')
         self.addCleanup(queue.clear)
         charm_data = factory.get_charm_json()
         queue.put(charm_data)
@@ -281,7 +281,7 @@
                 with patch(lp_get_queue, return_value=queue) as gq:
                     dequeue()
         getdb.assert_called_with(ANY, MONGO_DATABASE)
-        gq.assert_called_with(CHARM_QUEUE)
+        gq.assert_called_with(self.db, CHARM_QUEUE)
         self.assertEqual(0, queue.size())
 
     @patch('requests.get', failing_requests_get)
@@ -290,7 +290,7 @@
         # charm branches, no ingest jobs are queued. Otherwise, the
         # promulgated status may be incorrectly reset for promulgated
         # charms that are not returned by all_branch_data().
-        out_queue = get_queue('test_queue')
+        out_queue = get_queue(self.db, 'test_queue')
         factory.makeCharm(self.db)
         self.assertRaises(
             ConnectionError, queue_from_branches, self.db,

=== modified file 'charmworld/jobs/tests/test_worker.py'
--- charmworld/jobs/tests/test_worker.py	2013-07-17 20:40:55 +0000
+++ charmworld/jobs/tests/test_worker.py	2013-07-22 15:53:24 +0000
@@ -1,9 +1,10 @@
 # Copyright 2013 Canonical Ltd.  This software is licensed under the
 # GNU Affero General Public License version 3 (see the file LICENSE).
 
+__metaclass__ = type
+
 from contextlib import contextmanager
 
-from mongoqueue import MongoQueue
 from mock import (
     create_autospec,
     patch,
@@ -11,14 +12,21 @@
 from pyelasticsearch import ElasticSearch
 
 from charmworld.testing import logger_buffer
-from charmworld.jobs.config import settings
+from charmworld.jobs.config import (
+    CHARM_QUEUE,
+    BASKET_QUEUE,
+    settings,
+)
 from charmworld.jobs.worker import (
-    CharmWorker,
-    DEFAULT_JOBS,
+    QueueWorker,
     main,
 )
-from charmworld.jobs.ingest import IngestJob
+from charmworld.jobs.ingest import (
+    DBIngestJob,
+    UpdateCharmJob,
+)
 from charmworld.jobs.tests.test_ingest import charm_update_environment
+from charmworld.jobs.utils import get_queue
 from charmworld.search import ElasticSearchClient
 from charmworld.testing import (
     factory,
@@ -26,13 +34,13 @@
 )
 
 
-class SuccessfulRun(IngestJob):
+class SuccessfulRun(DBIngestJob):
 
     def run(self, charm_data):
         charm_data['ingested'] = True
 
 
-class FailureRun(IngestJob):
+class FailureRun(DBIngestJob):
 
     def run(self, charm_data):
         if charm_data['charm'] == 'foo':
@@ -41,9 +49,10 @@
             charm_data['ingested'] = True
 
 
-class GatheringRun(IngestJob):
+class GatheringRun(DBIngestJob):
 
-    def setup(self):
+    def setup(self, db=None):
+        super(GatheringRun, self).setup(db)
         self.charms = []
 
     def run(self, charm_data):
@@ -54,23 +63,21 @@
 def main_environment(charm, index_client, queue):
     """An environment in which worker.main() can be run."""
     with patch('charmworld.jobs.worker.configure_logging'):
-        with patch('sys.argv', ['foo', '--quit-on-empty']):
-            with patch('charmworld.jobs.worker.get_queue',
-                       lambda x: queue):
-                with charm_update_environment(charm, index_client):
-                    yield
+        with charm_update_environment(charm, index_client):
+            yield
 
 
 class WorkerTest(MongoTestBase):
 
     def setUp(self):
         super(WorkerTest, self).setUp()
-        self.queue = MongoQueue(self.db.queue, 'queue')
-        self.queue.put({'charm': 'foo'})
-        self.queue.put({'charm': 'bar'})
+        self.charm_queue = get_queue(self.db, CHARM_QUEUE)
+        self.basket_queue = get_queue(self.db, BASKET_QUEUE)
+        self.charm_queue.put({'charm': 'foo'})
+        self.charm_queue.put({'charm': 'bar'})
 
     def tearDown(self):
-        self.queue.clear()
+        self.charm_queue.clear()
         super(WorkerTest, self).tearDown()
 
     def test_basic_running(self):
@@ -80,10 +87,13 @@
             SuccessfulRun(),
             GatheringRun(),
         )
-        worker = CharmWorker(ingest_jobs, self.queue)
-        worker.run(quit_on_empty=True)
+        queue_jobs = {
+            self.charm_queue: ingest_jobs
+        }
+        worker = QueueWorker(self.db, queue_jobs)
+        worker.run()
 
-        self.assertEqual(0, self.queue.size())
+        self.assertEqual(0, self.charm_queue.size())
         expected = [
             {'ingested': True, 'charm': 'foo'},
             {'ingested': True, 'charm': 'bar'},
@@ -93,10 +103,13 @@
 
     def test_failure_while_running(self):
         ingest_jobs = (FailureRun(), GatheringRun())
-        worker = CharmWorker(ingest_jobs, self.queue)
-        worker.run(quit_on_empty=True)
+        queue_jobs = {
+            self.charm_queue: ingest_jobs
+        }
+        worker = QueueWorker(self.db, queue_jobs)
+        worker.run()
 
-        self.assertEqual(0, self.queue.size())
+        self.assertEqual(0, self.charm_queue.size())
         expected = [
             {'ingested': True, 'charm': 'bar'},
         ]
@@ -122,40 +135,81 @@
             SuccessfulRun(),
             GatheringRun(),
         )
-        worker = CharmWorker(ingest_jobs, LockedJobQueue)
-        worker.run(quit_on_empty=True)
+        worker = QueueWorker(self.db, {LockedJobQueue: ingest_jobs})
+        worker.run()
 
     def test_ingest_job_contract(self):
         # All ingest jobs accept the right inputs.
-        ingest_jobs = [create_autospec(job)() for job in DEFAULT_JOBS]
-        worker = CharmWorker(ingest_jobs, self.queue)
-        worker.run(quit_on_empty=True)
-        self.assertEqual(0, self.queue.size())
+        ingest_jobs = [create_autospec(UpdateCharmJob)()]
+        worker = QueueWorker(self.db, {self.charm_queue: ingest_jobs})
+        worker.run()
+        self.assertEqual(0, self.charm_queue.size())
 
     def test_interval_is_int(self):
-        worker = CharmWorker([], self.queue)
+        worker = QueueWorker(self.db, {self.charm_queue: []})
         self.assertIs(int, type(worker.interval),
-                      'CharmWorker.interval is a %s, not an int.' %
+                      'QueueWorker.interval is a %s, not an int.' %
                       type(worker.interval))
 
     def run_main_with_exit(self):
-        with self.assertRaises(SystemExit) as e:
-            main()
+        with patch('charmworld.jobs.worker.configure_logging'):
+            with self.assertRaises(SystemExit) as e:
+                main(['ingest-queued'], db=self.db)
         return e.exception.code
 
     def test_elasticsearch_failure(self):
         payload = factory.get_payload_json()
-        self.queue.clear()
-        self.queue.put(payload)
+        self.charm_queue.clear()
+        self.charm_queue.put(payload)
         charm = factory.get_charm_json(payload=payload)
         index_client = ElasticSearchClient(
             ElasticSearch('http://localhost:70'), 'foo')
         with patch.dict(settings,
                         values={'es_urls': 'http://localhost:70'}):
-            with main_environment(charm, index_client, self.queue):
+            with main_environment(charm, index_client, self.charm_queue):
                 with logger_buffer('charm.worker') as handler:
                     exit_code = self.run_main_with_exit()
         self.assertEqual(40, handler.buffer[1].levelno)
         self.assertIn('Could not connect', handler.buffer[1].msg)
         self.assertEqual(1, exit_code)
-        self.assertEqual(0, self.queue.size())
+        self.assertEqual(0, self.charm_queue.size())
+
+    def test_job_for_bundles_is_run_if_bundle_queue_has_entries(self):
+        # Basket queue entries are processed by run_job.
+        queue = get_queue(self.db, BASKET_QUEUE)
+        queue_entry = {'bundle': 'foo'}
+        queue.put(queue_entry)
+        with patch('charmworld.jobs.worker.run_job') as run_job:
+            self.run_main_with_exit()
+            queue_entries_run = [x[0][1] for x in run_job.call_args_list]
+            self.assertIn(queue_entry, queue_entries_run)
+
+    def test_worker_does_not_run_forever_by_default(self):
+        run_called = []
+        class FauxWorker:
+            def __init__(self, *args):
+                pass
+
+            @staticmethod
+            def run(run_forever):
+                self.assertFalse(run_forever)
+                run_called.append(True)
+
+        with patch('charmworld.jobs.worker.QueueWorker', new=FauxWorker):
+            self.run_main_with_exit()
+        self.assertTrue(run_called)
+
+    def test_worker_runs_forever_if_command_line_says_to(self):
+        run_called = []
+        class FauxWorker:
+            def __init__(self, *args):
+                pass
+
+            @staticmethod
+            def run(run_forever):
+                self.assertFalse(run_forever)
+                run_called.append(True)
+
+        with patch('charmworld.jobs.worker.QueueWorker', new=FauxWorker):
+            self.run_main_with_exit()
+        self.assertTrue(run_called)

=== modified file 'charmworld/jobs/utils.py'
--- charmworld/jobs/utils.py	2013-06-10 14:24:39 +0000
+++ charmworld/jobs/utils.py	2013-07-22 15:53:24 +0000
@@ -10,19 +10,12 @@
 from socket import gethostname
 from time import time
 
-from charmworld.models import getconnection
-from charmworld.models import getdb
-from charmworld.utils import get_ini
-
-
-def get_queue(queue_name):
+
+def get_queue(db, queue_name):
     """Get the required mongoqueues for the provided queue name.
 
     As some jobs only require one queue, only one queue name is mandatory.
     """
-    settings = get_ini()
-    connection = getconnection(settings)
-    db = getdb(connection, settings.get('mongo.database'))
     return MongoQueue(db[queue_name], '%s-queue' % queue_name)
 
 

=== modified file 'charmworld/jobs/worker.py'
--- charmworld/jobs/worker.py	2013-06-14 21:08:36 +0000
+++ charmworld/jobs/worker.py	2013-07-22 15:53:24 +0000
@@ -6,102 +6,93 @@
 import sys
 import time
 
-from charmworld.utils import (
-    configure_logging,
-    get_ini,
+from charmworld.jobs.config import (
+    CHARM_QUEUE,
+    BASKET_QUEUE,
 )
-from charmworld.jobs.config import CHARM_QUEUE
 from charmworld.jobs.ingest import (
     run_job,
+    UpdateBundleJob,
     UpdateCharmJob,
 )
 from charmworld.jobs.utils import get_queue
+from charmworld.models import getconnection
+from charmworld.models import getdb
 from charmworld.search import SearchServiceNotAvailable
-
-
-DEFAULT_JOBS = (
-    UpdateCharmJob,
+from charmworld.utils import (
+    configure_logging,
+    get_ini,
 )
 
+
 settings = get_ini()
 INTERVAL = int(settings.get('worker_interval'))
 
 
-class CharmWorker(object):
+class QueueWorker(object):
     """Processes raw charm data"""
 
-    def __init__(self, ingest_jobs, queue, interval=INTERVAL):
-        self.ingest_jobs = ingest_jobs
-        self.queue = queue
+    def __init__(self, db, queue_jobs, interval=INTERVAL):
+        self.db = db
+        self.queue_jobs = queue_jobs
         self.log = logging.getLogger('charm.worker')
         self.interval = interval
 
-    def run(self, quit_on_empty=False):
+    def run(self, run_forever=False):
         self.log.info('Starting processing')
         return_code = 0
-        for job in self.ingest_jobs:
-            job.setup()
+        for jobs in self.queue_jobs.values():
+            for job in jobs:
+                job.setup(db=self.db)
         while True:
-            item = self.queue.next()
-            if item is None:
-                if quit_on_empty:
-                    self.log.info(
-                        'Empty queue with quit_on_empty set. Quitting.')
-                    break
-                else:
-                    wait_time_in_minutes = self.interval / 60
-                    self.log.info(
-                        'Empty queue. Waiting %s minutes to recheck.' %
-                        wait_time_in_minutes)
-                    time.sleep(self.interval)
-                    continue
-            charm_data = item.payload
-            try:
-                for job in self.ingest_jobs:
+            for queue, jobs in self.queue_jobs.items():
+                while True:
+                    item = queue.next()
+                    if item is None:
+                        break
+                    charm_data = item.payload
                     try:
-                        if not run_job(job, charm_data, needs_setup=False):
-                            break
-                    except SearchServiceNotAvailable as e:
-                        self.log.exception(str(e))
-                        self.queue.clear()
-                        return_code = 1
-
-            finally:
-                item.complete()
+                        for job in jobs:
+                            try:
+                                if not run_job(
+                                        job, charm_data, needs_setup=False):
+                                    break
+                            except SearchServiceNotAvailable as e:
+                                self.log.exception(str(e))
+                                queue.clear()
+                                return_code = 1
+                    finally:
+                        item.complete()
+
+            if not run_forever:
+                self.log.info('Empty queue and not running forever; quitting.')
+                break
+
+            wait_time_in_minutes = self.interval / 60
+            self.log.info(
+                'Empty queue.  Waiting %s minutes to recheck.' %
+                wait_time_in_minutes)
+            time.sleep(self.interval)
         return return_code
 
 
-# XXX j.c.sackett Feb 21 2013 Bug:1111708
-# Functions not currently bound to jobs; probably need new jobs created for
-# them outside of ingest worker.
-#def reindex(db, indexer):
-    #log = logging.getLogger("charm.index")
-    #count = 0
-    #for charm in db.find():
-        #log.debug("Indexing %s", charm['branch_spec'])
-        #try:
-            #index_charm(indexer, charm)
-        #except:
-            #log.error("Indexing charm %s", charm['branch_spec'])
-            #raise
-        #count += 1
-    #log.info("Indexed %d Charms" % count)
-
-
-def main():
+def main(argv=sys.argv, db=None):
+    if db is None:
+        settings = get_ini()
+        connection = getconnection(settings)
+        db = getdb(connection, settings.get('mongo.database'))
     configure_logging()
     parser = argparse.ArgumentParser()
     parser.add_argument(
-        "--quit-on-empty",
-        help="Shut down the worker if the queue is empty",
-        action="store_true")
+        '--run-forever',
+        help='Keep running after the queues are processed and process more '
+        'later.',
+        action='store_true', dest='run_forever')
     # Explicitly pass sys.argv to make monkeypatching more reliable.
-    args = parser.parse_args(sys.argv[1:])
-    queue = get_queue(CHARM_QUEUE)
-    jobs = [job() for job in DEFAULT_JOBS]
-    worker = CharmWorker(jobs, queue)
-    sys.exit(worker.run(quit_on_empty=args.quit_on_empty))
-
-
-if __name__ == "__main__":
-    main()
+    args = parser.parse_args(argv[1:])
+    queue_jobs = {
+        get_queue(db, CHARM_QUEUE): [UpdateCharmJob()],
+        get_queue(db, BASKET_QUEUE): [UpdateBundleJob()],
+    }
+    worker = QueueWorker(db, queue_jobs)
+    sys.exit(worker.run(run_forever=args.run_forever))

=== modified file 'charmworld/models.py'
--- charmworld/models.py	2013-07-19 12:51:01 +0000
+++ charmworld/models.py	2013-07-22 15:53:24 +0000
@@ -7,6 +7,7 @@
 from calendar import timegm
 import logging
 from mimetypes import guess_type
+import os
 from os.path import dirname
 from os.path import join
 from os.path import normpath
@@ -36,7 +37,7 @@
     :param settings: dict of application settings including the mongo.X info.
 
     """
-    url_or_host = settings.get('mongo.url')
+    url_or_host = os.environ.get('TEST_MONGODB', settings.get('mongo.url'))
     if not url_or_host:
         url_or_host = settings.get('mongo.host')
         port = int(settings.get('mongo.port'))

=== modified file 'charmworld/teams.py'
--- charmworld/teams.py	2013-01-04 16:42:25 +0000
+++ charmworld/teams.py	2013-07-22 15:53:24 +0000
@@ -65,8 +65,10 @@
 @since: 2.1.1
 """
 
-from openid.message import registerNamespaceAlias, \
-     NamespaceAliasRegistrationError
+from openid.message import (
+    registerNamespaceAlias,
+    NamespaceAliasRegistrationError,
+)
 from openid.extension import Extension
 from openid import oidutil
 

=== modified file 'charmworld/tests/test_models.py'
--- charmworld/tests/test_models.py	2013-07-19 12:51:01 +0000
+++ charmworld/tests/test_models.py	2013-07-22 15:53:24 +0000
@@ -1192,6 +1192,7 @@
 class TestStoreBundles(MongoTestBase):
 
     def test_store_bundles(self):
+        # The bundles can be stored in the database.
         deployer_config = dedent("""\
             wordpress-stage:
                 series: precise

=== modified file 'docs/index.rst'
--- docs/index.rst	2013-07-15 20:55:10 +0000
+++ docs/index.rst	2013-07-22 15:53:24 +0000
@@ -59,7 +59,7 @@
 ::
 
     $ bin/enqueue
-    $ bin/ingest-queued --quit-on-empty
+    $ bin/ingest-queued
 
 Output verbosity is controlled by the logging config in charmworld.ini
 

=== removed file 'scripts/worker'
--- scripts/worker	2013-02-21 19:14:36 +0000
+++ scripts/worker	1970-01-01 00:00:00 +0000
@@ -1,16 +0,0 @@
-#!/bin/bash
-
-# Copyright 2012, 2013 Marco Ceppi, Canonical Ltd.  This software is
-# licensed under the GNU Affero General Public License version 3 (see
-# the file LICENSE).
-
-set -e
-
-SCRIPT_DIR="$( cd "$( dirname "${BASH_SOURCE[0]}" )" && pwd )"
-PROJECT_DIR="$( dirname $SCRIPT_DIR )"
-PYTHON_BIN=$PROJECT_DIR/bin/python
-: ${LOGDIR:="$( dirname $PROJECT_DIR )/var"}
-JOBDIR=$PROJECT_DIR/charmworld/jobs
-
-echo "Starting worker..."
-$PYTHON_BIN $JOBDIR/worker.py --quit-on-empty &>> $LOGDIR/ingest-worker.log