Merge lp:~abentley/lazr.jobrunner/launchpad-via-celery into lp:lazr.jobrunner

[Aaron Bentley]

Update to support run-via-celery in Launchpad

This branch improves the compatibility of the celery job-running code with Launchpad.

The biggest change is that Celery-specific code has been isolated in its own modules. This is because importing from Celery has side-effects (i.e. loading a config module), so only code that intends to relate to Celery should import it.

I've removed src/lazr.jobrunner.egg-info from version control, because AFAICT its files are all auto-generated by bin/buildout.

I've added a "lint" target to the makefile, so "make lint" will check lint. However, it lints all .py files in src. Since we don't have significant lint-broken code, we can ensure a completely lint-clean tree. I've cleaned up what lint we did have.

I've updated the version to 0.2 in setup.py, so that LP code can request the development egg instead of the 0.1 code.

I've moved celeryconfig.py to src/lazr/jobrunner/celeryconfig.py. This is currently used only by the job runner.

I've copied the Launchpad version of LeaseHeld into jobrunner, and updated Launchpad to import it.

I've tweaked runJobHandleError to take a reference to job.fail early, so that it can be invoked when the DB transaction is invalid.

I've added getJobRunner to RunJobHandleError, so that Launchpad can use lp.services.job.runner.BaseJobRunner.

[Abel Deuring (adeuring)]

looks good

review: Approve (code)

=== modified file '.bzrignore'
--- .bzrignore	2012-02-23 17:11:46 +0000
+++ .bzrignore	2012-03-22 18:19:18 +0000
@@ -6,4 +6,4 @@
 tmp
 build
 dist
-src/lazr.jobrunner.egg-info/SOURCES.txt
+src/lazr.jobrunner.egg-info

=== modified file 'Makefile'
--- Makefile	2012-03-01 18:41:07 +0000
+++ Makefile	2012-03-22 18:19:18 +0000
@@ -1,9 +1,9 @@
+develop: bin/python
+	bin/buildout
+
 check: bin/python
 	bin/nosetests
 
-develop: bin/python
-	bin/buildout
-
 bin/buildout:
 	python -S bootstrap.py
 
@@ -13,4 +13,7 @@
 clean:
 	bzr clean-tree --force --ignored --unknown
 
-.PHONY: develop check clean
+lint:
+	pocketlint `find src -name '*.py'`
+
+.PHONY: check clean develop lint

=== modified file 'setup.py'
--- setup.py	2012-03-15 15:14:59 +0000
+++ setup.py	2012-03-22 18:19:18 +0000
@@ -22,12 +22,13 @@
 NEWS = open(os.path.join(here, 'NEWS.txt')).read()
 
 
-version = '0.1'
+version = '0.2'
 
 install_requires = [
     # List your project dependencies here.
     # For more details, see:
     # http://packages.python.org/distribute/setuptools.html#declaring-dependencies
+    'celery',
 ]
 
 

=== removed directory 'src/lazr.jobrunner.egg-info'
=== removed file 'src/lazr.jobrunner.egg-info/PKG-INFO'
--- src/lazr.jobrunner.egg-info/PKG-INFO	2012-03-15 14:59:09 +0000
+++ src/lazr.jobrunner.egg-info/PKG-INFO	1970-01-01 00:00:00 +0000
@@ -1,46 +0,0 @@
-Metadata-Version: 1.0
-Name: lazr.jobrunner
-Version: 0.1
-Summary: A Celery based job runner
-Home-page: UNKNOWN
-Author: UNKNOWN
-Author-email: UNKNOWN
-License: UNKNOWN
-Description: This file requires editing
-        ==========================
-        
-        Note to the author: Please add something informative to this README *before*
-        releasing your software, as `a little documentation goes a long way`_.  Both
-        README.rst (this file) and NEWS.txt (release notes) will be included in your
-        package metadata which gets displayed in the PyPI page for your project.
-        
-        You can take a look at the README.txt of other projects, such as repoze.bfg
-        (http://bfg.repoze.org/trac/browser/trunk/README.txt) for some ideas.
-        
-        .. _`a little documentation goes a long way`: http://www.martinaspeli.net/articles/a-little-documentation-goes-a-long-way
-        
-        Credits
-        -------
-        
-        - `Distribute`_
-        - `Buildout`_
-        - `modern-package-template`_
-        
-        .. _Buildout: http://www.buildout.org/
-        .. _Distribute: http://pypi.python.org/pypi/distribute
-        .. _`modern-package-template`: http://pypi.python.org/pypi/modern-package-template
-        
-        
-        News
-        ====
-        
-        0.1
-        ---
-        
-        *Release date: 15-Mar-2012*
-        
-        * Initial reelease
-        
-        
-        
-Platform: UNKNOWN

=== removed file 'src/lazr.jobrunner.egg-info/dependency_links.txt'
--- src/lazr.jobrunner.egg-info/dependency_links.txt	2012-02-23 10:32:17 +0000
+++ src/lazr.jobrunner.egg-info/dependency_links.txt	1970-01-01 00:00:00 +0000
@@ -1,1 +0,0 @@
-

=== removed file 'src/lazr.jobrunner.egg-info/entry_points.txt'
--- src/lazr.jobrunner.egg-info/entry_points.txt	2012-02-23 10:32:17 +0000
+++ src/lazr.jobrunner.egg-info/entry_points.txt	1970-01-01 00:00:00 +0000
@@ -1,3 +0,0 @@
-[console_scripts]
-lazr.jobrunner = lazr.jobrunner:main
-

=== removed file 'src/lazr.jobrunner.egg-info/namespace_packages.txt'
--- src/lazr.jobrunner.egg-info/namespace_packages.txt	2012-02-23 10:32:17 +0000
+++ src/lazr.jobrunner.egg-info/namespace_packages.txt	1970-01-01 00:00:00 +0000
@@ -1,1 +0,0 @@
-lazr

=== removed file 'src/lazr.jobrunner.egg-info/not-zip-safe'
--- src/lazr.jobrunner.egg-info/not-zip-safe	2012-02-23 10:32:17 +0000
+++ src/lazr.jobrunner.egg-info/not-zip-safe	1970-01-01 00:00:00 +0000
@@ -1,1 +0,0 @@
-

=== removed file 'src/lazr.jobrunner.egg-info/top_level.txt'
--- src/lazr.jobrunner.egg-info/top_level.txt	2012-02-23 10:32:17 +0000
+++ src/lazr.jobrunner.egg-info/top_level.txt	1970-01-01 00:00:00 +0000
@@ -1,1 +0,0 @@
-lazr

=== modified file 'src/lazr/jobrunner/__init__.py'
--- src/lazr/jobrunner/__init__.py	2012-02-23 10:32:17 +0000
+++ src/lazr/jobrunner/__init__.py	2012-03-22 18:19:18 +0000
@@ -1,4 +1,5 @@
 # Example package with a console entry point
 
+
 def main():
     print "Hello World"

=== renamed file 'celeryconfig.py' => 'src/lazr/jobrunner/celeryconfig.py'
=== added file 'src/lazr/jobrunner/celerytask.py'
--- src/lazr/jobrunner/celerytask.py	1970-01-01 00:00:00 +0000
+++ src/lazr/jobrunner/celerytask.py	2012-03-22 18:19:18 +0000
@@ -0,0 +1,45 @@
+# Copyright 2012 Canonical Ltd.  All rights reserved.
+#
+# This file is part of lazr.jobrunner
+#
+# lazr.jobrunner is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Lesser General Public License as published by
+# the Free Software Foundation, version 3 of the License.
+#
+# lazr.jobrunner 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 Lesser General Public
+# License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with lazr.jobrunner. If not, see <http://www.gnu.org/licenses/>.
+
+__metaclass__ = type
+
+
+from celery.task import Task
+from lazr.jobrunner.jobrunner import (
+    JobRunner,
+    LeaseHeld,
+    memory_limit,
+    )
+
+
+class RunJob(Task):
+
+    abstract = True
+
+    oops_config = None
+
+    def getJobRunner(self):
+        return JobRunner(oops_config=self.oops_config)
+
+    def run(self, job_id):
+        job = self.job_source.get(job_id)
+        try:
+            job.acquireLease()
+        except LeaseHeld:
+            return
+        runner = self.getJobRunner()
+        with memory_limit(self.job_source.memory_limit):
+            runner.runJobHandleError(job)

=== modified file 'src/lazr/jobrunner/jobrunner.py'
--- src/lazr/jobrunner/jobrunner.py	2012-03-16 18:27:52 +0000
+++ src/lazr/jobrunner/jobrunner.py	2012-03-22 18:19:18 +0000
@@ -26,8 +26,6 @@
     )
 import sys
 
-from celery.task import Task
-
 
 class SuspendJobException(Exception):
     """Raised when a running job wants to suspend itself."""
@@ -35,7 +33,10 @@
 
 
 class LeaseHeld(Exception):
-    """Raised when a lease cannot be acquired."""
+    """Raised when attempting to acquire a lease that is already held."""
+
+    def __init__(self):
+        Exception.__init__(self, 'Lease is already held.')
 
 
 class JobStatus:
@@ -65,23 +66,6 @@
         FAILED, SUSPENDED])
 
 
-class RunJob(Task):
-
-    abstract = True
-
-    oops_config = None
-
-    def run(self, job_id):
-        job = self.job_source.get(job_id)
-        try:
-            job.acquireLease()
-        except LeaseHeld:
-            return
-        runner = JobRunner(oops_config=self.oops_config)
-        with memory_limit(self.job_source.memory_limit):
-            runner.runJobHandleError(job)
-
-
 @contextmanager
 def memory_limit(limit):
     if limit is not None:
@@ -169,6 +153,7 @@
         self.logger.info(
             'Running %s in status %s' % (
                 self.job_str(job), job.status.title))
+        fail = job.fail
         job.start(manage_transaction=True)
         do_retry = False
         try:
@@ -186,7 +171,7 @@
             job.suspend(manage_transaction=True)
             self.incomplete_jobs.append(job)
         except Exception:
-            job.fail(manage_transaction=True)
+            fail(manage_transaction=True)
             self.incomplete_jobs.append(job)
             raise
         else:
@@ -216,7 +201,7 @@
                 self.logger.exception("Failure in _doOops: %s" % e)
                 info = sys.exc_info()
                 report = job.makeOopsReport(self.oops_config, info)
-                oops_ids = self.oops_config.publish(report)
+                self.oops_config.publish(report)
                 return report
 
     @staticmethod

=== modified file 'src/lazr/jobrunner/tests/config1.py'
--- src/lazr/jobrunner/tests/config1.py	2012-03-09 21:50:46 +0000
+++ src/lazr/jobrunner/tests/config1.py	2012-03-22 18:19:18 +0000
@@ -1,6 +1,6 @@
 BROKER_VHOST = "/"
 CELERY_RESULT_BACKEND = "amqp"
-CELERY_IMPORTS = ("lazr.jobrunner.tests.test_jobrunner", )
+CELERY_IMPORTS = ("lazr.jobrunner.tests.test_celerytask", )
 CELERYD_CONCURRENCY = 1
 import os
 import oops

=== added file 'src/lazr/jobrunner/tests/test_celerytask.py'
--- src/lazr/jobrunner/tests/test_celerytask.py	1970-01-01 00:00:00 +0000
+++ src/lazr/jobrunner/tests/test_celerytask.py	2012-03-22 18:19:18 +0000
@@ -0,0 +1,286 @@
+# Copyright 2012 Canonical Ltd.  All rights reserved.
+#
+# This file is part of lazr.jobrunner
+#
+# lazr.jobrunner is free software: you can redistribute it and/or modify
+# it under the terms of the GNU Lesser General Public License as published by
+# the Free Software Foundation, version 3 of the License.
+#
+# lazr.jobrunner 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 Lesser General Public
+# License for more details.
+#
+# You should have received a copy of the GNU Lesser General Public License
+# along with lazr.jobrunner. If not, see <http://www.gnu.org/licenses/>.
+
+__metaclass__ = type
+
+
+import contextlib
+import errno
+import json
+import os
+import os.path
+from resource import (
+    getrlimit,
+    RLIMIT_AS,
+    )
+import shutil
+import subprocess
+import tempfile
+from time import sleep
+from unittest import TestCase
+os.environ.setdefault('CELERY_CONFIG_MODULE', 'lazr.jobrunner.celeryconfig')
+
+from celery.exceptions import SoftTimeLimitExceeded
+
+from lazr.jobrunner.celerytask import RunJob
+from lazr.jobrunner.jobrunner import (
+    JobStatus,
+    )
+from lazr.jobrunner.tests.test_jobrunner import (
+    FakeJob,
+    )
+
+
+def get_root():
+    import lazr.jobrunner
+    root = os.path.join(os.path.dirname(lazr.jobrunner.__file__), '../../../')
+    return os.path.normpath(root)
+
+
[e-mail]
+def running(cmd_name, cmd_args, env=None, cwd=None):
+    proc = subprocess.Popen((cmd_name,) + cmd_args, env=env,
+                            stderr=subprocess.PIPE, cwd=cwd)
+    try:
+        yield proc
+    finally:
+        proc.terminate()
+        proc.wait()
+
+
+def celeryd(config_module, file_job_dir):
+    cmd_args = ('--config', config_module)
+    environ = dict(os.environ)
+    environ['FILE_JOB_DIR'] = file_job_dir
+    return running('bin/celeryd', cmd_args, environ, cwd=get_root())
+
+
[e-mail]
+def tempdir():
+    dirname = tempfile.mkdtemp()
+    try:
+        yield dirname
+    finally:
+        shutil.rmtree(dirname)
+
+
+class FakeJobSource:
+
+    memory_limit = None
+
+    def __init__(self):
+        self.jobs = {}
+
+    def get(self, job_id):
+        return self.jobs[job_id]
+
+
+class FileJob(FakeJob):
+
+    def __init__(self, job_source, job_id, output=None,
+                 status=JobStatus.WAITING, exception=None, sleep=None):
+        super(FileJob, self).__init__(job_id)
+        self.job_source = job_source
+        self.output = output
+        self.status = status
+        self.exception = exception
+        self.sleep = sleep
+
+    def save(self):
+        self.job_source.set(self)
+
+    def run(self):
+        super(FileJob, self).run()
+        if self.sleep is not None:
+            sleep(self.sleep)
+        if self.exception is not None:
+            raise Exception(self.exception)
+        if self.output is not None:
+            self.job_source.set_output(self, self.output)
+
+
+class FileJobSource:
+
+    memory_limit = None
+
+    def __init__(self, root):
+        self.root = root
+        self.job_root = os.path.join(self.root, 'job')
+        self.output_root = os.path.join(self.root, 'output')
+
+        def ensure_dir(path):
+            try:
+                os.mkdir(path)
+            except OSError, e:
+                if e.errno != errno.EEXIST:
+                    raise
+        ensure_dir(self.job_root)
+        ensure_dir(self.output_root)
+
+    def _job_file(self, job_id, mode):
+        return open(os.path.join(self.job_root, str(job_id)), mode)
+
+    def _job_output_file(self, job_id, mode):
+        return open(os.path.join(self.output_root, str(job_id)), mode)
+
+    def get(self, job_id):
+        with self._job_file(job_id, 'r') as job_file:
+            job_data = json.load(job_file)
+            job_data['status'] = JobStatus.by_value[job_data['status']]
+            return FileJob(self, **job_data)
+
+    def set(self, job):
+        with self._job_file(job.job_id, 'w') as job_file:
+            job_info = {
+                'job_id': job.job_id,
+                'output': job.output,
+                'status': job.status.value,
+                'exception': job.exception,
+                'sleep': job.sleep,
+            }
+            json.dump(job_info, job_file)
+
+    def get_output(self, job):
+        try:
+            with self._job_output_file(job.job_id, 'r') as job_output_file:
+                    return job_output_file.read()
+        except IOError, e:
+            if e.errno == errno.ENOENT:
+                return None
+            raise
+
+    def set_output(self, job, output):
+        with self._job_output_file(job.job_id, 'w') as job_output_file:
+            job_output_file.write(output)
+
+
+class RunFileJob(RunJob):
+
+    name = 'run_file_job'
+
+    def __init__(self):
+        self.file_job_dir = None
+
+    @property
+    def job_source(self):
+        return FileJobSource(self.file_job_dir)
+
+
+class TestRunJob(TestCase):
+
+    @staticmethod
+    def makeFakeJobSource(job=None):
+        js = FakeJobSource()
+        if job is None:
+            job = FakeJob(10)
+        js.jobs[job.job_id] = job
+        return js
+
+    @staticmethod
+    def runJob(js):
+        task = RunJob()
+        task.job_source = js
+        task.run(10)
+
+    def test_run(self):
+        js = self.makeFakeJobSource()
+        self.assertTrue(js.jobs[10].unrun)
+        self.runJob(js)
+        self.assertFalse(js.jobs[10].unrun)
+
+    def test_memory_limit(self):
+
+        class MemoryCheckJob(FakeJob):
+
+            def run(self):
+                super(MemoryCheckJob, self).run()
+                self.current_memory_limit = getrlimit(RLIMIT_AS)[0]
+
+        start_limits = getrlimit(RLIMIT_AS)
+        js = FakeJobSource()
+        job = MemoryCheckJob(10)
+        js.jobs[10] = job
+        js.memory_limit = 1024 ** 3
+        task = RunJob()
+        task.job_source = js
+        task.run(10)
+        self.assertEqual(1024 ** 3, job.current_memory_limit)
+        self.assertEqual(start_limits, getrlimit(RLIMIT_AS))
+
+    def test_acquires_lease(self):
+        js = self.makeFakeJobSource()
+        self.assertFalse(js.jobs[10].lease_held)
+        self.runJob(js)
+        self.assertTrue(js.jobs[10].lease_held)
+
+    def test_skips_failed_acquisition(self):
+        js = self.makeFakeJobSource()
+        js.jobs[10].acquireLease()
+        self.runJob(js)
+        self.assertTrue(js.jobs[10].unrun)
+
+
+class TestCeleryD(TestCase):
+
+    def test_run_job(self):
+        with tempdir() as temp_dir:
+            js = FileJobSource(temp_dir)
+            job = FileJob(js, 10, 'my_output')
+            job.save()
+            result = RunFileJob.delay(10)
+            self.assertIs(None, js.get_output(job))
+            self.assertEqual(JobStatus.WAITING, job.status)
+            with celeryd('lazr.jobrunner.tests.config1', temp_dir):
+                result.wait(10)
+            job = js.get(job.job_id)
+            self.assertEqual('my_output', js.get_output(job))
+            self.assertEqual(JobStatus.COMPLETED, job.status)
+
+    def run_file_job(self, temp_dir, config='lazr.jobrunner.tests.config1',
+                     **kwargs):
+        js = FileJobSource(temp_dir)
+        job = FileJob(js, 10, **kwargs)
+        job.save()
+        result = RunFileJob.delay(10)
+        with celeryd(config, temp_dir) as proc:
+            try:
+                result.wait(10)
+            except SoftTimeLimitExceeded:
+                pass
+        job = js.get(job.job_id)
+        return job, proc
+
+    def test_run_job_emits_oopses(self):
+        with tempdir() as temp_dir:
+            job, proc = self.run_file_job(
+                temp_dir, exception='Catch me if you can!')
+            err = proc.stderr.read()
+            self.assertEqual(JobStatus.FAILED, job.status)
+            self.assertIs(None, job.job_source.get_output(job))
+            self.assertIn(
+                "OOPS while executing job 10: [] Exception(u'Catch me if you"
+                " can!',)", err)
+
+    def test_timeout_long(self):
+        """Raises exception when a job exceeds the configured time limit."""
+        with tempdir() as temp_dir:
+            job, proc = self.run_file_job(
+                temp_dir, config='lazr.jobrunner.tests.time_limit_config',
+                sleep=10)
+        self.assertEqual(JobStatus.FAILED, job.status)
+        err = proc.stderr.read()
+        self.assertIn(
+            'OOPS while executing job 10: [] SoftTimeLimitExceeded', err)

=== modified file 'src/lazr/jobrunner/tests/test_jobrunner.py'
--- src/lazr/jobrunner/tests/test_jobrunner.py	2012-03-16 18:27:52 +0000
+++ src/lazr/jobrunner/tests/test_jobrunner.py	2012-03-22 18:19:18 +0000
@@ -17,21 +17,9 @@
 __metaclass__ = type
 
 import contextlib
-import errno
-import json
 import logging
-import os.path
-from resource import (
-    getrlimit,
-    RLIMIT_AS,
-    )
-import subprocess
-import shutil
-import tempfile
-from time import sleep
 from unittest import TestCase
 
-from celery.exceptions import SoftTimeLimitExceeded
 import oops
 from zope.testing.loghandler import Handler
 
@@ -40,7 +28,6 @@
     JobRunner,
     JobStatus,
     LeaseHeld,
-    RunJob,
     SuspendJobException,
     )
 
@@ -97,243 +84,6 @@
         return [('foo', 'bar')]
 
 
-class FakeJobSource:
-
-    memory_limit = None
-
-    def __init__(self):
-        self.jobs = {}
-
-    def get(self, job_id):
-        return self.jobs[job_id]
-
-
-class FileJob(FakeJob):
-
-    def __init__(self, job_source, job_id, output=None,
-                 status=JobStatus.WAITING, exception=None, sleep=None):
-        super(FileJob, self).__init__(job_id)
-        self.job_source = job_source
-        self.output = output
-        self.status = status
-        self.exception = exception
-        self.sleep = sleep
-
-    def save(self):
-        self.job_source.set(self)
-
-    def run(self):
-        super(FileJob, self).run()
-        if self.sleep is not None:
-            sleep(self.sleep)
-        if self.exception is not None:
-            raise Exception(self.exception)
-        if self.output is not None:
-            self.job_source.set_output(self, self.output)
-
-
-class FileJobSource:
-
-    memory_limit = None
-
-    def __init__(self, root):
-        self.root = root
-        self.job_root = os.path.join(self.root, 'job')
-        self.output_root = os.path.join(self.root, 'output')
-        def ensure_dir(path):
-            try:
-                os.mkdir(path)
-            except OSError, e:
-                if e.errno != errno.EEXIST:
-                    raise
-        ensure_dir(self.job_root)
-        ensure_dir(self.output_root)
-
-    def _job_file(self, job_id, mode):
-        return open(os.path.join(self.job_root, str(job_id)), mode)
-
-    def _job_output_file(self, job_id, mode):
-        return open(os.path.join(self.output_root, str(job_id)), mode)
-
-    def get(self, job_id):
-        with self._job_file(job_id, 'r') as job_file:
-            job_data = json.load(job_file)
-            job_data['status'] = JobStatus.by_value[job_data['status']]
-            return FileJob(self, **job_data)
-
-    def set(self, job):
-        with self._job_file(job.job_id, 'w') as job_file:
-            job_info = {
-                'job_id': job.job_id,
-                'output': job.output,
-                'status': job.status.value,
-                'exception': job.exception,
-                'sleep': job.sleep,
-            }
-            json.dump(job_info, job_file)
-
-    def get_output(self, job):
-        try:
-            with self._job_output_file(job.job_id, 'r') as job_output_file:
-                    return job_output_file.read()
-        except IOError, e:
-            if e.errno == errno.ENOENT:
-                return None
-            raise
-
-    def set_output(self, job, output):
-        with self._job_output_file(job.job_id, 'w') as job_output_file:
-            job_output_file.write(output)
-
-
-class TestRunJob(TestCase):
-
-    @staticmethod
-    def makeFakeJobSource(job=None):
-        js = FakeJobSource()
-        if job is None:
-            job = FakeJob(10)
-        js.jobs[job.job_id] = job
-        return js
-
-    @staticmethod
-    def runJob(js):
-        task = RunJob()
-        task.job_source = js
-        task.run(10)
-
-    def test_run(self):
-        js = self.makeFakeJobSource()
-        self.assertTrue(js.jobs[10].unrun)
-        self.runJob(js)
-        self.assertFalse(js.jobs[10].unrun)
-
-    def test_memory_limit(self):
-
-        class MemoryCheckJob(FakeJob):
-
-            def run(self):
-                super(MemoryCheckJob, self).run()
-                self.current_memory_limit = getrlimit(RLIMIT_AS)[0]
-
-        start_limits = getrlimit(RLIMIT_AS)
-        js = FakeJobSource()
-        job = MemoryCheckJob(10)
-        js.jobs[10] = job
-        js.memory_limit = 1024 ** 3
-        task = RunJob()
-        task.job_source = js
-        task.run(10)
-        self.assertEqual(1024 ** 3, job.current_memory_limit)
-        self.assertEqual(start_limits, getrlimit(RLIMIT_AS))
-
-    def test_acquires_lease(self):
-        js = self.makeFakeJobSource()
-        self.assertFalse(js.jobs[10].lease_held)
-        self.runJob(js)
-        self.assertTrue(js.jobs[10].lease_held)
-
-    def test_skips_failed_acquisition(self):
-        js = self.makeFakeJobSource()
-        js.jobs[10].acquireLease()
-        self.runJob(js)
-        self.assertTrue(js.jobs[10].unrun)
-
-
-def get_root():
-    import lazr.jobrunner
-    root = os.path.join(os.path.dirname(lazr.jobrunner.__file__), '../../../')
-    return os.path.normpath(root)
-
-
[e-mail]
-def tempdir():
-    dirname = tempfile.mkdtemp()
-    try:
-        yield dirname
-    finally:
-        shutil.rmtree(dirname)
-
-
[e-mail]
-def celeryd(config_module, file_job_dir):
-    cmdname = os.path.join(get_root(), 'bin/celeryd')
-    environ = dict(os.environ)
-    environ['FILE_JOB_DIR'] = file_job_dir
-    proc = subprocess.Popen([cmdname, '--config', config_module], env=environ,
-                            stderr=subprocess.PIPE)
-    try:
-        yield proc
-    finally:
-        proc.terminate()
-        proc.wait()
-
-
-class RunFileJob(RunJob):
-
-    name = 'run_file_job'
-
-    def __init__(self):
-        file_job_dir = None
-
-    @property
-    def job_source(self):
-        return FileJobSource(self.file_job_dir)
-
-
-class TestCeleryD(TestCase):
-
-    def test_run_job(self):
-        with tempdir() as temp_dir:
-            js = FileJobSource(temp_dir)
-            job = FileJob(js, 10, 'my_output')
-            job.save()
-            result = RunFileJob.delay(10)
-            self.assertIs(None, js.get_output(job))
-            self.assertEqual(JobStatus.WAITING, job.status)
-            with celeryd('lazr.jobrunner.tests.config1', temp_dir):
-                result.wait(10)
-            job = js.get(job.job_id)
-            self.assertEqual('my_output', js.get_output(job))
-            self.assertEqual(JobStatus.COMPLETED, job.status)
-
-    def run_file_job(self, temp_dir, config='lazr.jobrunner.tests.config1',
-                     **kwargs):
-        js = FileJobSource(temp_dir)
-        job = FileJob(js, 10, **kwargs)
-        job.save()
-        result = RunFileJob.delay(10)
-        with celeryd(config, temp_dir) as proc:
-            try:
-                result.wait(10)
-            except SoftTimeLimitExceeded:
-                pass
-        job = js.get(job.job_id)
-        return job, proc
-
-    def test_run_job_emits_oopses(self):
-        with tempdir() as temp_dir:
-            job, proc = self.run_file_job(
-                temp_dir, exception='Catch me if you can!')
-            err = proc.stderr.read()
-            self.assertEqual(JobStatus.FAILED, job.status)
-            self.assertIs(None, job.job_source.get_output(job))
-            self.assertIn(
-                "OOPS while executing job 10: [] Exception(u'Catch me if you"
-                " can!',)", err)
-
-    def test_timeout_long(self):
-        """Raises exception when a job exceeds the configured time limit."""
-        with tempdir() as temp_dir:
-            job, proc = self.run_file_job(
-                temp_dir, config='lazr.jobrunner.tests.time_limit_config',
-                sleep=10)
-        self.assertEqual(JobStatus.FAILED, job.status)
-        err = proc.stderr.read()
-        self.assertIn(
-            'OOPS while executing job 10: [] SoftTimeLimitExceeded', err)
-
-
 class OOPSTestRepository:
     """Record OOPSes in memory."""
 

=== modified file 'src/lazr/jobrunner/tests/time_limit_config.py'
--- src/lazr/jobrunner/tests/time_limit_config.py	2012-03-14 14:22:04 +0000
+++ src/lazr/jobrunner/tests/time_limit_config.py	2012-03-22 18:19:18 +0000
@@ -1,6 +1,6 @@
 BROKER_VHOST = "/"
 CELERY_RESULT_BACKEND = "amqp"
-CELERY_IMPORTS = ("lazr.jobrunner.tests.test_jobrunner", )
+CELERY_IMPORTS = ("lazr.jobrunner.tests.test_celerytask", )
 CELERYD_CONCURRENCY = 1
 CELERYD_TASK_SOFT_TIME_LIMIT = 1
 import os

=== modified file 'src/lazr/jobrunner/version.txt'
--- src/lazr/jobrunner/version.txt	2012-03-15 14:59:09 +0000
+++ src/lazr/jobrunner/version.txt	2012-03-22 18:19:18 +0000
@@ -1,1 +1,1 @@
-0.1
+0.2