Synnefo » snf-ganeti » ganeti-local

#!/usr/bin/python

#

# Copyright (C) 2010, 2011, 2012 Google Inc.

#

# This program is free software; you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation; either version 2 of the License, or

# (at your option) any later version.

#

# This program 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

# General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program; if not, write to the Free Software

# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA

# 02110-1301, USA.

"""Script for testing ganeti.jqueue"""

import os

import sys

import unittest

import tempfile

import shutil

import errno

import itertools

import random

import operator

from ganeti import constants

from ganeti import utils

from ganeti import errors

from ganeti import jqueue

from ganeti import opcodes

from ganeti import compat

from ganeti import mcpu

from ganeti import query

from ganeti import workerpool

import testutils

class _FakeJob:

  def __init__(self, job_id, status):

    self.id = job_id

    self.writable = False

    self._status = status

    self._log = []

  def SetStatus(self, status):

    self._status = status

  def AddLogEntry(self, msg):

    self._log.append((len(self._log), msg))

  def CalcStatus(self):

    return self._status

  def GetInfo(self, fields):

    result = []

    for name in fields:

      if name == "status":

        result.append(self._status)

      else:

        raise Exception("Unknown field")

    return result

  def GetLogEntries(self, newer_than):

    assert newer_than is None or newer_than >= 0

    if newer_than is None:

      return self._log

    return self._log[newer_than:]

class TestJobChangesChecker(unittest.TestCase):

  def testStatus(self):

    job = _FakeJob(9094, constants.JOB_STATUS_QUEUED)

    checker = jqueue._JobChangesChecker(["status"], None, None)

    self.assertEqual(checker(job), ([constants.JOB_STATUS_QUEUED], []))

    job.SetStatus(constants.JOB_STATUS_RUNNING)

    self.assertEqual(checker(job), ([constants.JOB_STATUS_RUNNING], []))

    job.SetStatus(constants.JOB_STATUS_SUCCESS)

    self.assertEqual(checker(job), ([constants.JOB_STATUS_SUCCESS], []))

    # job.id is used by checker

    self.assertEqual(job.id, 9094)

  def testStatusWithPrev(self):

    job = _FakeJob(12807, constants.JOB_STATUS_QUEUED)

    checker = jqueue._JobChangesChecker(["status"],

                                        [constants.JOB_STATUS_QUEUED], None)

    self.assert_(checker(job) is None)

    job.SetStatus(constants.JOB_STATUS_RUNNING)

    self.assertEqual(checker(job), ([constants.JOB_STATUS_RUNNING], []))

  def testFinalStatus(self):

    for status in constants.JOBS_FINALIZED:

      job = _FakeJob(2178711, status)

      checker = jqueue._JobChangesChecker(["status"], [status], None)

      # There won't be any changes in this status, hence it should signal

      # a change immediately

      self.assertEqual(checker(job), ([status], []))

  def testLog(self):

    job = _FakeJob(9094, constants.JOB_STATUS_RUNNING)

    checker = jqueue._JobChangesChecker(["status"], None, None)

    self.assertEqual(checker(job), ([constants.JOB_STATUS_RUNNING], []))

    job.AddLogEntry("Hello World")

    (job_info, log_entries) = checker(job)

    self.assertEqual(job_info, [constants.JOB_STATUS_RUNNING])

    self.assertEqual(log_entries, [[0, "Hello World"]])

    checker2 = jqueue._JobChangesChecker(["status"], job_info, len(log_entries))

    self.assert_(checker2(job) is None)

    job.AddLogEntry("Foo Bar")

    job.SetStatus(constants.JOB_STATUS_ERROR)

    (job_info, log_entries) = checker2(job)

    self.assertEqual(job_info, [constants.JOB_STATUS_ERROR])

    self.assertEqual(log_entries, [[1, "Foo Bar"]])

    checker3 = jqueue._JobChangesChecker(["status"], None, None)

    (job_info, log_entries) = checker3(job)

    self.assertEqual(job_info, [constants.JOB_STATUS_ERROR])

    self.assertEqual(log_entries, [[0, "Hello World"], [1, "Foo Bar"]])

class TestJobChangesWaiter(unittest.TestCase):

  def setUp(self):

    self.tmpdir = tempfile.mkdtemp()

    self.filename = utils.PathJoin(self.tmpdir, "job-1")

    utils.WriteFile(self.filename, data="")

  def tearDown(self):

    shutil.rmtree(self.tmpdir)

  def _EnsureNotifierClosed(self, notifier):

    try:

      os.fstat(notifier._fd)

    except EnvironmentError, err:

      self.assertEqual(err.errno, errno.EBADF)

    else:

      self.fail("File descriptor wasn't closed")

  def testClose(self):

    for wait in [False, True]:

      waiter = jqueue._JobFileChangesWaiter(self.filename)

      try:

        if wait:

          waiter.Wait(0.001)

      finally:

        waiter.Close()

      # Ensure file descriptor was closed

      self._EnsureNotifierClosed(waiter._notifier)

  def testChangingFile(self):

    waiter = jqueue._JobFileChangesWaiter(self.filename)

    try:

      self.assertFalse(waiter.Wait(0.1))

      utils.WriteFile(self.filename, data="changed")

      self.assert_(waiter.Wait(60))

    finally:

      waiter.Close()

    self._EnsureNotifierClosed(waiter._notifier)

  def testChangingFile2(self):

    waiter = jqueue._JobChangesWaiter(self.filename)

    try:

      self.assertFalse(waiter._filewaiter)

      self.assert_(waiter.Wait(0.1))

      self.assert_(waiter._filewaiter)

      # File waiter is now used, but there have been no changes

      self.assertFalse(waiter.Wait(0.1))

      utils.WriteFile(self.filename, data="changed")

      self.assert_(waiter.Wait(60))

    finally:

      waiter.Close()

    self._EnsureNotifierClosed(waiter._filewaiter._notifier)

class TestWaitForJobChangesHelper(unittest.TestCase):

  def setUp(self):

    self.tmpdir = tempfile.mkdtemp()

    self.filename = utils.PathJoin(self.tmpdir, "job-2614226563")

    utils.WriteFile(self.filename, data="")

  def tearDown(self):

    shutil.rmtree(self.tmpdir)

  def _LoadWaitingJob(self):

    return _FakeJob(2614226563, constants.JOB_STATUS_WAITING)

  def _LoadLostJob(self):

    return None

  def testNoChanges(self):

    wfjc = jqueue._WaitForJobChangesHelper()

    # No change

    self.assertEqual(wfjc(self.filename, self._LoadWaitingJob, ["status"],

                          [constants.JOB_STATUS_WAITING], None, 0.1),

                     constants.JOB_NOTCHANGED)

    # No previous information

    self.assertEqual(wfjc(self.filename, self._LoadWaitingJob,

                          ["status"], None, None, 1.0),

                     ([constants.JOB_STATUS_WAITING], []))

  def testLostJob(self):

    wfjc = jqueue._WaitForJobChangesHelper()

    self.assert_(wfjc(self.filename, self._LoadLostJob,

                      ["status"], None, None, 1.0) is None)

class TestEncodeOpError(unittest.TestCase):

  def test(self):

    encerr = jqueue._EncodeOpError(errors.LockError("Test 1"))

    self.assert_(isinstance(encerr, tuple))

    self.assertRaises(errors.LockError, errors.MaybeRaise, encerr)

    encerr = jqueue._EncodeOpError(errors.GenericError("Test 2"))

    self.assert_(isinstance(encerr, tuple))

    self.assertRaises(errors.GenericError, errors.MaybeRaise, encerr)

    encerr = jqueue._EncodeOpError(NotImplementedError("Foo"))

    self.assert_(isinstance(encerr, tuple))

    self.assertRaises(errors.OpExecError, errors.MaybeRaise, encerr)

    encerr = jqueue._EncodeOpError("Hello World")

    self.assert_(isinstance(encerr, tuple))

    self.assertRaises(errors.OpExecError, errors.MaybeRaise, encerr)

class TestQueuedOpCode(unittest.TestCase):

  def testDefaults(self):

    def _Check(op):

      self.assertFalse(hasattr(op.input, "dry_run"))

      self.assertEqual(op.priority, constants.OP_PRIO_DEFAULT)

      self.assertFalse(op.log)

      self.assert_(op.start_timestamp is None)

      self.assert_(op.exec_timestamp is None)

      self.assert_(op.end_timestamp is None)

      self.assert_(op.result is None)

      self.assertEqual(op.status, constants.OP_STATUS_QUEUED)

    op1 = jqueue._QueuedOpCode(opcodes.OpTestDelay())

    _Check(op1)

    op2 = jqueue._QueuedOpCode.Restore(op1.Serialize())

    _Check(op2)

    self.assertEqual(op1.Serialize(), op2.Serialize())

  def testPriority(self):

    def _Check(op):

      assert constants.OP_PRIO_DEFAULT != constants.OP_PRIO_HIGH, \

             "Default priority equals high priority; test can't work"

      self.assertEqual(op.priority, constants.OP_PRIO_HIGH)

      self.assertEqual(op.status, constants.OP_STATUS_QUEUED)

    inpop = opcodes.OpTagsGet(priority=constants.OP_PRIO_HIGH)

    op1 = jqueue._QueuedOpCode(inpop)

    _Check(op1)

    op2 = jqueue._QueuedOpCode.Restore(op1.Serialize())

    _Check(op2)

    self.assertEqual(op1.Serialize(), op2.Serialize())

class TestQueuedJob(unittest.TestCase):

  def testNoOpCodes(self):

    self.assertRaises(errors.GenericError, jqueue._QueuedJob,

                      None, 1, [], False)

  def testDefaults(self):

    job_id = 4260

    ops = [

      opcodes.OpTagsGet(),

      opcodes.OpTestDelay(),

      ]

    def _Check(job):

      self.assertTrue(job.writable)

      self.assertEqual(job.id, job_id)

      self.assertEqual(job.log_serial, 0)

      self.assert_(job.received_timestamp)

      self.assert_(job.start_timestamp is None)

      self.assert_(job.end_timestamp is None)

      self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED)

      self.assertEqual(job.CalcPriority(), constants.OP_PRIO_DEFAULT)

      self.assert_(repr(job).startswith("<"))

      self.assertEqual(len(job.ops), len(ops))

      self.assert_(compat.all(inp.__getstate__() == op.input.__getstate__()

                              for (inp, op) in zip(ops, job.ops)))

      self.assertRaises(errors.OpPrereqError, job.GetInfo,

                        ["unknown-field"])

      self.assertEqual(job.GetInfo(["summary"]),

                       [[op.input.Summary() for op in job.ops]])

      self.assertFalse(job.archived)

    job1 = jqueue._QueuedJob(None, job_id, ops, True)

    _Check(job1)

    job2 = jqueue._QueuedJob.Restore(None, job1.Serialize(), True, False)

    _Check(job2)

    self.assertEqual(job1.Serialize(), job2.Serialize())

  def testWritable(self):

    job = jqueue._QueuedJob(None, 1, [opcodes.OpTestDelay()], False)

    self.assertFalse(job.writable)

    job = jqueue._QueuedJob(None, 1, [opcodes.OpTestDelay()], True)

    self.assertTrue(job.writable)

  def testArchived(self):

    job = jqueue._QueuedJob(None, 1, [opcodes.OpTestDelay()], False)

    self.assertFalse(job.archived)

    newjob = jqueue._QueuedJob.Restore(None, job.Serialize(), True, True)

    self.assertTrue(newjob.archived)

    newjob2 = jqueue._QueuedJob.Restore(None, newjob.Serialize(), True, False)

    self.assertFalse(newjob2.archived)

  def testPriority(self):

    job_id = 4283

    ops = [

      opcodes.OpTagsGet(priority=constants.OP_PRIO_DEFAULT),

      opcodes.OpTestDelay(),

      ]

    def _Check(job):

      self.assertEqual(job.id, job_id)

      self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED)

      self.assert_(repr(job).startswith("<"))

    job = jqueue._QueuedJob(None, job_id, ops, True)

    _Check(job)

    self.assert_(compat.all(op.priority == constants.OP_PRIO_DEFAULT

                            for op in job.ops))

    self.assertEqual(job.CalcPriority(), constants.OP_PRIO_DEFAULT)

    # Increase first

    job.ops[0].priority -= 1

    _Check(job)

    self.assertEqual(job.CalcPriority(), constants.OP_PRIO_DEFAULT - 1)

    # Mark opcode as finished

    job.ops[0].status = constants.OP_STATUS_SUCCESS

    _Check(job)

    self.assertEqual(job.CalcPriority(), constants.OP_PRIO_DEFAULT)

    # Increase second

    job.ops[1].priority -= 10

    self.assertEqual(job.CalcPriority(), constants.OP_PRIO_DEFAULT - 10)

    # Test increasing first

    job.ops[0].status = constants.OP_STATUS_RUNNING

    job.ops[0].priority -= 19

    self.assertEqual(job.CalcPriority(), constants.OP_PRIO_DEFAULT - 20)

  def _JobForPriority(self, job_id):

    ops = [

      opcodes.OpTagsGet(),

      opcodes.OpTestDelay(),

      opcodes.OpTagsGet(),

      opcodes.OpTestDelay(),

      ]

    job = jqueue._QueuedJob(None, job_id, ops, True)

    self.assertTrue(compat.all(op.priority == constants.OP_PRIO_DEFAULT

                               for op in job.ops))

    self.assertEqual(job.CalcPriority(), constants.OP_PRIO_DEFAULT)

    self.assertFalse(compat.any(hasattr(op.input, "priority")

                                for op in job.ops))

    return job

  def testChangePriorityAllQueued(self):

    job = self._JobForPriority(24984)

    self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED)

    self.assertTrue(compat.all(op.status == constants.OP_STATUS_QUEUED

                               for op in job.ops))

    result = job.ChangePriority(-10)

    self.assertEqual(job.CalcPriority(), -10)

    self.assertTrue(compat.all(op.priority == -10 for op in job.ops))

    self.assertTrue(compat.all(op.input.priority == -10 for op in job.ops))

    self.assertEqual(result,

                     (True, ("Priorities of pending opcodes for job 24984 have"

                             " been changed to -10")))

  def testChangePriorityAllFinished(self):

    job = self._JobForPriority(16405)

    for (idx, op) in enumerate(job.ops):

      if idx > 2:

        op.status = constants.OP_STATUS_ERROR

      else:

        op.status = constants.OP_STATUS_SUCCESS

    self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_ERROR)

    self.assertEqual(job.CalcPriority(), constants.OP_PRIO_DEFAULT)

    result = job.ChangePriority(-10)

    self.assertEqual(job.CalcPriority(), constants.OP_PRIO_DEFAULT)

    self.assertTrue(compat.all(op.priority == constants.OP_PRIO_DEFAULT

                               for op in job.ops))

    self.assertFalse(compat.any(hasattr(op.input, "priority")

                                for op in job.ops))

    self.assertEqual(map(operator.attrgetter("status"), job.ops), [

      constants.OP_STATUS_SUCCESS,

      constants.OP_STATUS_SUCCESS,

      constants.OP_STATUS_SUCCESS,

      constants.OP_STATUS_ERROR,

      ])

    self.assertEqual(result, (False, "Job 16405 is finished"))

  def testChangePriorityCancelling(self):

    job = self._JobForPriority(31572)

    for (idx, op) in enumerate(job.ops):

      if idx > 1:

        op.status = constants.OP_STATUS_CANCELING

      else:

        op.status = constants.OP_STATUS_SUCCESS

    self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_CANCELING)

    self.assertEqual(job.CalcPriority(), constants.OP_PRIO_DEFAULT)

    result = job.ChangePriority(5)

    self.assertEqual(job.CalcPriority(), constants.OP_PRIO_DEFAULT)

    self.assertTrue(compat.all(op.priority == constants.OP_PRIO_DEFAULT

                               for op in job.ops))

    self.assertFalse(compat.any(hasattr(op.input, "priority")

                                for op in job.ops))

    self.assertEqual(map(operator.attrgetter("status"), job.ops), [

      constants.OP_STATUS_SUCCESS,

      constants.OP_STATUS_SUCCESS,

      constants.OP_STATUS_CANCELING,

      constants.OP_STATUS_CANCELING,

      ])

    self.assertEqual(result, (False, "Job 31572 is cancelling"))

  def testChangePriorityFirstRunning(self):

    job = self._JobForPriority(1716215889)

    for (idx, op) in enumerate(job.ops):

      if idx == 0:

        op.status = constants.OP_STATUS_RUNNING

      else:

        op.status = constants.OP_STATUS_QUEUED

    self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_RUNNING)

    self.assertEqual(job.CalcPriority(), constants.OP_PRIO_DEFAULT)

    result = job.ChangePriority(7)

    self.assertEqual(job.CalcPriority(), constants.OP_PRIO_DEFAULT)

    self.assertEqual(map(operator.attrgetter("priority"), job.ops),

                     [constants.OP_PRIO_DEFAULT, 7, 7, 7])

    self.assertEqual([getattr(op.input, "priority", None) for op in job.ops],

                     [None, 7, 7, 7])

    self.assertEqual(map(operator.attrgetter("status"), job.ops), [

      constants.OP_STATUS_RUNNING,

      constants.OP_STATUS_QUEUED,

      constants.OP_STATUS_QUEUED,

      constants.OP_STATUS_QUEUED,

      ])

    self.assertEqual(result,

                     (True, ("Priorities of pending opcodes for job"

                             " 1716215889 have been changed to 7")))

  def testChangePriorityLastRunning(self):

    job = self._JobForPriority(1308)

    for (idx, op) in enumerate(job.ops):

      if idx == (len(job.ops) - 1):

        op.status = constants.OP_STATUS_RUNNING

      else:

        op.status = constants.OP_STATUS_SUCCESS

    self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_RUNNING)

    self.assertEqual(job.CalcPriority(), constants.OP_PRIO_DEFAULT)

    result = job.ChangePriority(-3)

    self.assertEqual(job.CalcPriority(), constants.OP_PRIO_DEFAULT)

    self.assertTrue(compat.all(op.priority == constants.OP_PRIO_DEFAULT

                               for op in job.ops))

    self.assertFalse(compat.any(hasattr(op.input, "priority")

                                for op in job.ops))

    self.assertEqual(map(operator.attrgetter("status"), job.ops), [

      constants.OP_STATUS_SUCCESS,

      constants.OP_STATUS_SUCCESS,

      constants.OP_STATUS_SUCCESS,

      constants.OP_STATUS_RUNNING,

      ])

    self.assertEqual(result, (False, "Job 1308 had no pending opcodes"))

  def testChangePrioritySecondOpcodeRunning(self):

    job = self._JobForPriority(27701)

    self.assertEqual(len(job.ops), 4)

    job.ops[0].status = constants.OP_STATUS_SUCCESS

    job.ops[1].status = constants.OP_STATUS_RUNNING

    job.ops[2].status = constants.OP_STATUS_QUEUED

    job.ops[3].status = constants.OP_STATUS_QUEUED

    self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_RUNNING)

    result = job.ChangePriority(-19)

    self.assertEqual(job.CalcPriority(), -19)

    self.assertEqual(map(operator.attrgetter("priority"), job.ops),

                     [constants.OP_PRIO_DEFAULT, constants.OP_PRIO_DEFAULT,

                      -19, -19])

    self.assertEqual([getattr(op.input, "priority", None) for op in job.ops],

                     [None, None, -19, -19])

    self.assertEqual(map(operator.attrgetter("status"), job.ops), [

      constants.OP_STATUS_SUCCESS,

      constants.OP_STATUS_RUNNING,

      constants.OP_STATUS_QUEUED,

      constants.OP_STATUS_QUEUED,

      ])

    self.assertEqual(result,

                     (True, ("Priorities of pending opcodes for job"

                             " 27701 have been changed to -19")))

  def testChangePriorityWithInconsistentJob(self):

    job = self._JobForPriority(30097)

    self.assertEqual(len(job.ops), 4)

    # This job is invalid (as it has two opcodes marked as running) and make

    # the call fail because an unprocessed opcode precedes a running one (which

    # should never happen in reality)

    job.ops[0].status = constants.OP_STATUS_SUCCESS

    job.ops[1].status = constants.OP_STATUS_RUNNING

    job.ops[2].status = constants.OP_STATUS_QUEUED

    job.ops[3].status = constants.OP_STATUS_RUNNING

    self.assertRaises(AssertionError, job.ChangePriority, 19)

  def testCalcStatus(self):

    def _Queued(ops):

      # The default status is "queued"

      self.assert_(compat.all(op.status == constants.OP_STATUS_QUEUED

                              for op in ops))

    def _Waitlock1(ops):

      ops[0].status = constants.OP_STATUS_WAITING

    def _Waitlock2(ops):

      ops[0].status = constants.OP_STATUS_SUCCESS

      ops[1].status = constants.OP_STATUS_SUCCESS

      ops[2].status = constants.OP_STATUS_WAITING

    def _Running(ops):

      ops[0].status = constants.OP_STATUS_SUCCESS

      ops[1].status = constants.OP_STATUS_RUNNING

      for op in ops[2:]:

        op.status = constants.OP_STATUS_QUEUED

    def _Canceling1(ops):

      ops[0].status = constants.OP_STATUS_SUCCESS

      ops[1].status = constants.OP_STATUS_SUCCESS

      for op in ops[2:]:

        op.status = constants.OP_STATUS_CANCELING

    def _Canceling2(ops):

      for op in ops:

        op.status = constants.OP_STATUS_CANCELING

    def _Canceled(ops):

      for op in ops:

        op.status = constants.OP_STATUS_CANCELED

    def _Error1(ops):

      for idx, op in enumerate(ops):

        if idx > 3:

          op.status = constants.OP_STATUS_ERROR

        else:

          op.status = constants.OP_STATUS_SUCCESS

    def _Error2(ops):

      for op in ops:

        op.status = constants.OP_STATUS_ERROR

    def _Success(ops):

      for op in ops:

        op.status = constants.OP_STATUS_SUCCESS

    tests = {

      constants.JOB_STATUS_QUEUED: [_Queued],

      constants.JOB_STATUS_WAITING: [_Waitlock1, _Waitlock2],

      constants.JOB_STATUS_RUNNING: [_Running],

      constants.JOB_STATUS_CANCELING: [_Canceling1, _Canceling2],

      constants.JOB_STATUS_CANCELED: [_Canceled],

      constants.JOB_STATUS_ERROR: [_Error1, _Error2],

      constants.JOB_STATUS_SUCCESS: [_Success],

      }

    def _NewJob():

      job = jqueue._QueuedJob(None, 1,

                              [opcodes.OpTestDelay() for _ in range(10)],

                              True)

      self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED)

      self.assert_(compat.all(op.status == constants.OP_STATUS_QUEUED

                              for op in job.ops))

      return job

    for status in constants.JOB_STATUS_ALL:

      sttests = tests[status]

      assert sttests

      for fn in sttests:

        job = _NewJob()

        fn(job.ops)

        self.assertEqual(job.CalcStatus(), status)

class _FakeDependencyManager:

  def __init__(self):

    self._checks = []

    self._notifications = []

    self._waiting = set()

  def AddCheckResult(self, job, dep_job_id, dep_status, result):

    self._checks.append((job, dep_job_id, dep_status, result))

  def CountPendingResults(self):

    return len(self._checks)

  def CountWaitingJobs(self):

    return len(self._waiting)

  def GetNextNotification(self):

    return self._notifications.pop(0)

  def JobWaiting(self, job):

    return job in self._waiting

  def CheckAndRegister(self, job, dep_job_id, dep_status):

    (exp_job, exp_dep_job_id, exp_dep_status, result) = self._checks.pop(0)

    assert exp_job == job

    assert exp_dep_job_id == dep_job_id

    assert exp_dep_status == dep_status

    (result_status, _) = result

    if result_status == jqueue._JobDependencyManager.WAIT:

      self._waiting.add(job)

    elif result_status == jqueue._JobDependencyManager.CONTINUE:

      self._waiting.remove(job)

    return result

  def NotifyWaiters(self, job_id):

    self._notifications.append(job_id)

class _DisabledFakeDependencyManager:

  def JobWaiting(self, _):

    return False

  def CheckAndRegister(self, *args):

    assert False, "Should not be called"

  def NotifyWaiters(self, _):

    pass

class _FakeQueueForProc:

  def __init__(self, depmgr=None):

    self._acquired = False

    self._updates = []

    self._submitted = []

    self._accepting_jobs = True

    self._submit_count = itertools.count(1000)

    if depmgr:

      self.depmgr = depmgr

    else:

      self.depmgr = _DisabledFakeDependencyManager()

  def IsAcquired(self):

    return self._acquired

  def GetNextUpdate(self):

    return self._updates.pop(0)

  def GetNextSubmittedJob(self):

    return self._submitted.pop(0)

  def acquire(self, shared=0):

    assert shared == 1

    self._acquired = True

  def release(self):

    assert self._acquired

    self._acquired = False

  def UpdateJobUnlocked(self, job, replicate=True):

    assert self._acquired, "Lock not acquired while updating job"

    self._updates.append((job, bool(replicate)))

  def SubmitManyJobs(self, jobs):

    assert not self._acquired, "Lock acquired while submitting jobs"

    job_ids = [self._submit_count.next() for _ in jobs]

    self._submitted.extend(zip(job_ids, jobs))

    return job_ids

  def StopAcceptingJobs(self):

    self._accepting_jobs = False

  def AcceptingJobsUnlocked(self):

    return self._accepting_jobs

class _FakeExecOpCodeForProc:

  def __init__(self, queue, before_start, after_start):

    self._queue = queue

    self._before_start = before_start

    self._after_start = after_start

  def __call__(self, op, cbs, timeout=None):

    assert isinstance(op, opcodes.OpTestDummy)

    assert not self._queue.IsAcquired(), \

           "Queue lock not released when executing opcode"

    if self._before_start:

      self._before_start(timeout, cbs.CurrentPriority())

    cbs.NotifyStart()

    if self._after_start:

      self._after_start(op, cbs)

    # Check again after the callbacks

    assert not self._queue.IsAcquired()

    if op.fail:

      raise errors.OpExecError("Error requested (%s)" % op.result)

    if hasattr(op, "submit_jobs") and op.submit_jobs is not None:

      return cbs.SubmitManyJobs(op.submit_jobs)

    return op.result

class _JobProcessorTestUtils:

  def _CreateJob(self, queue, job_id, ops):

    job = jqueue._QueuedJob(queue, job_id, ops, True)

    self.assertFalse(job.start_timestamp)

    self.assertFalse(job.end_timestamp)

    self.assertEqual(len(ops), len(job.ops))

    self.assert_(compat.all(op.input == inp

                            for (op, inp) in zip(job.ops, ops)))

    self.assertEqual(job.GetInfo(["ops"]), [[op.__getstate__() for op in ops]])

    return job

class TestJobProcessor(unittest.TestCase, _JobProcessorTestUtils):

  def _GenericCheckJob(self, job):

    assert compat.all(isinstance(op.input, opcodes.OpTestDummy)

                      for op in job.ops)

    self.assertEqual(job.GetInfo(["opstart", "opexec", "opend"]),

                     [[op.start_timestamp for op in job.ops],

                      [op.exec_timestamp for op in job.ops],

                      [op.end_timestamp for op in job.ops]])

    self.assertEqual(job.GetInfo(["received_ts", "start_ts", "end_ts"]),

                     [job.received_timestamp,

                      job.start_timestamp,

                      job.end_timestamp])

    self.assert_(job.start_timestamp)

    self.assert_(job.end_timestamp)

    self.assertEqual(job.start_timestamp, job.ops[0].start_timestamp)

  def testSuccess(self):

    queue = _FakeQueueForProc()

    for (job_id, opcount) in [(25351, 1), (6637, 3),

                              (24644, 10), (32207, 100)]:

      ops = [opcodes.OpTestDummy(result="Res%s" % i, fail=False)

             for i in range(opcount)]

      # Create job

      job = self._CreateJob(queue, job_id, ops)

      def _BeforeStart(timeout, priority):

        self.assertEqual(queue.GetNextUpdate(), (job, True))

        self.assertRaises(IndexError, queue.GetNextUpdate)

        self.assertFalse(queue.IsAcquired())

        self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITING)

        self.assertFalse(job.cur_opctx)

      def _AfterStart(op, cbs):

        self.assertEqual(queue.GetNextUpdate(), (job, True))

        self.assertRaises(IndexError, queue.GetNextUpdate)

        self.assertFalse(queue.IsAcquired())

        self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_RUNNING)

        self.assertFalse(job.cur_opctx)