← Back to branch summary

~launchpad-pqm/launchpad/devel

« back to all changes in this revision

Viewing changes to lib/lp/services/job/tests/test_runner.py

  • Committer: Launchpad Patch Queue Manager
  • Date: 2011-09-01 20:20:29 UTC
  • mfrom: (13843.1.1 rollback-13831)
  • Revision ID: launchpad@pqm.canonical.com-20110901202029-hhe1cuwc21qgwt60
[r=abentley][rollback=13831] Roll back r13831

Show diffs side-by-side

added added

removed removed

Lines of Context:
lib/lp/services/job/tests/test_runner.py
23
23
from lp.services.job.interfaces.job import (
24
24
    IRunnableJob,
25
25
    JobStatus,
26
 
    LeaseHeld,
27
26
    SuspendJobException,
28
27
    )
29
28
from lp.services.job.model.job import Job
501
500
        self.x = '*' * (10 ** 6)
502
501
 
503
502
 
504
 
class NoJobs(StaticJobSource):
505
 
 
506
 
    done = False
507
 
 
508
 
    jobs = []
509
 
 
510
 
 
511
 
class LeaseHeldJob(StaticJobSource):
512
 
 
513
 
    implements(IRunnableJob)
514
 
 
515
 
    jobs = [()]
516
 
 
517
 
    done = False
518
 
 
519
 
    def __init__(self, id):
520
 
        self.job = Job()
521
 
        self.id = id
522
 
 
523
 
    def acquireLease(self):
524
 
        raise LeaseHeld()
525
 
 
526
 
 
527
503
class TestTwistedJobRunner(ZopeTestInSubProcess, TestCaseWithFactory):
528
504
 
529
505
    layer = ZopelessDatabaseLayer
632
608
        oops = self.getOopsReport(runner, 0)
633
609
        self.assertEqual('MemoryError', oops.type)
634
610
 
635
 
    def test_no_jobs(self):
636
 
        logger = BufferLogger()
637
 
        logger.setLevel(logging.INFO)
638
 
        runner = TwistedJobRunner.runFromSource(
639
 
            NoJobs, 'branchscanner', logger)
640
 
        self.assertEqual(
641
 
            (0, 0), (len(runner.completed_jobs), len(runner.incomplete_jobs)))
642
 
 
643
 
    def test_lease_held_handled(self):
644
 
        """Jobs that raise LeaseHeld are handled correctly."""
645
 
        logger = BufferLogger()
646
 
        logger.setLevel(logging.DEBUG)
647
 
        runner = TwistedJobRunner.runFromSource(
648
 
            LeaseHeldJob, 'branchscanner', logger)
649
 
        self.assertIn('Could not acquire lease', logger.getLogBuffer())
650
 
        self.assertEqual(
651
 
            (0, 1), (len(runner.completed_jobs), len(runner.incomplete_jobs)))
652
 
 
653
611
 
654
612
class TestJobCronScript(ZopeTestInSubProcess, TestCaseWithFactory):
655
613