#!/usr/bin/python
#
-# Copyright (C) 2010, 2011 Google Inc.
+# 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
import shutil
import errno
import itertools
+import random
from ganeti import constants
from ganeti import utils
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 = []
shutil.rmtree(self.tmpdir)
def _LoadWaitingJob(self):
- return _FakeJob(2614226563, constants.JOB_STATUS_WAITLOCK)
+ return _FakeJob(2614226563, constants.JOB_STATUS_WAITING)
def _LoadLostJob(self):
return None
# No change
self.assertEqual(wfjc(self.filename, self._LoadWaitingJob, ["status"],
- [constants.JOB_STATUS_WAITLOCK], None, 0.1),
+ [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_WAITLOCK], []))
+ ([constants.JOB_STATUS_WAITING], []))
def testLostJob(self):
wfjc = jqueue._WaitForJobChangesHelper()
class TestQueuedJob(unittest.TestCase):
def test(self):
self.assertRaises(errors.GenericError, jqueue._QueuedJob,
- None, 1, [])
+ None, 1, [], False)
def testDefaults(self):
job_id = 4260
]
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.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.OpExecError, job.GetInfo,
+ 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)
+ job1 = jqueue._QueuedJob(None, job_id, ops, True)
_Check(job1)
- job2 = jqueue._QueuedJob.Restore(None, job1.Serialize())
+ 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 = [
self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED)
self.assert_(repr(job).startswith("<"))
- job = jqueue._QueuedJob(None, job_id, ops)
+ 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))
for op in ops))
def _Waitlock1(ops):
- ops[0].status = constants.OP_STATUS_WAITLOCK
+ 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_WAITLOCK
+ ops[2].status = constants.OP_STATUS_WAITING
def _Running(ops):
ops[0].status = constants.OP_STATUS_SUCCESS
tests = {
constants.JOB_STATUS_QUEUED: [_Queued],
- constants.JOB_STATUS_WAITLOCK: [_Waitlock1, _Waitlock2],
+ constants.JOB_STATUS_WAITING: [_Waitlock1, _Waitlock2],
constants.JOB_STATUS_RUNNING: [_Running],
constants.JOB_STATUS_CANCELING: [_Canceling1, _Canceling2],
constants.JOB_STATUS_CANCELED: [_Canceled],
def _NewJob():
job = jqueue._QueuedJob(None, 1,
- [opcodes.OpTestDelay() for _ in range(10)])
+ [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))
class _JobProcessorTestUtils:
def _CreateJob(self, queue, job_id, ops):
- job = jqueue._QueuedJob(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.assertEqual(queue.GetNextUpdate(), (job, True))
self.assertRaises(IndexError, queue.GetNextUpdate)
self.assertFalse(queue.IsAcquired())
- self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITLOCK)
+ self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITING)
self.assertFalse(job.cur_opctx)
def _AfterStart(op, cbs):
self.assertRaises(IndexError, queue.GetNextUpdate)
if idx == len(ops) - 1:
# Last opcode
- self.assert_(result)
+ self.assertEqual(result, jqueue._JobProcessor.FINISHED)
else:
- self.assertFalse(result)
+ self.assertEqual(result, jqueue._JobProcessor.DEFER)
self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED)
self.assert_(job.start_timestamp)
self._GenericCheckJob(job)
# Calling the processor on a finished job should be a no-op
- self.assertTrue(jqueue._JobProcessor(queue, opexec, job)())
+ self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(),
+ jqueue._JobProcessor.FINISHED)
self.assertRaises(IndexError, queue.GetNextUpdate)
def testOpcodeError(self):
for i in range(opcount)]
# Create job
- job = self._CreateJob(queue, job_id, ops)
+ job = self._CreateJob(queue, str(job_id), ops)
opexec = _FakeExecOpCodeForProc(queue, None, None)
if idx in (failfrom, len(ops) - 1):
# Last opcode
- self.assert_(result)
+ self.assertEqual(result, jqueue._JobProcessor.FINISHED)
break
- self.assertFalse(result)
+ self.assertEqual(result, jqueue._JobProcessor.DEFER)
self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED)
self._GenericCheckJob(job)
# Calling the processor on a finished job should be a no-op
- self.assertTrue(jqueue._JobProcessor(queue, opexec, job)())
+ self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(),
+ jqueue._JobProcessor.FINISHED)
self.assertRaises(IndexError, queue.GetNextUpdate)
def testCancelWhileInQueue(self):
# Simulate processor called in workerpool
opexec = _FakeExecOpCodeForProc(queue, None, None)
- self.assert_(jqueue._JobProcessor(queue, opexec, job)())
+ self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(),
+ jqueue._JobProcessor.FINISHED)
# Check result
self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_CANCELED)
self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED)
- job.ops[0].status = constants.OP_STATUS_WAITLOCK
+ job.ops[0].status = constants.OP_STATUS_WAITING
assert len(job.ops) == 5
- self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITLOCK)
+ self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITING)
# Mark as cancelling
(success, _) = job.Cancel()
for op in job.ops))
opexec = _FakeExecOpCodeForProc(queue, None, None)
- self.assert_(jqueue._JobProcessor(queue, opexec, job)())
+ self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(),
+ jqueue._JobProcessor.FINISHED)
# Check result
self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_CANCELED)
self.assertEqual(queue.GetNextUpdate(), (job, True))
self.assertRaises(IndexError, queue.GetNextUpdate)
self.assertFalse(queue.IsAcquired())
- self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITLOCK)
+ self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITING)
# Mark as cancelled
(success, _) = job.Cancel()
opexec = _FakeExecOpCodeForProc(queue, _BeforeStart, _AfterStart)
self.assertRaises(IndexError, queue.GetNextUpdate)
- self.assert_(jqueue._JobProcessor(queue, opexec, job)())
+ self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(),
+ jqueue._JobProcessor.FINISHED)
self.assertEqual(queue.GetNextUpdate(), (job, True))
self.assertRaises(IndexError, queue.GetNextUpdate)
def _BeforeStart(timeout, priority):
self.assertFalse(queue.IsAcquired())
- self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITLOCK)
+ self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITING)
# Mark as cancelled
(success, _) = job.Cancel()
opexec = _FakeExecOpCodeForProc(queue, _BeforeStart, _AfterStart)
- self.assert_(jqueue._JobProcessor(queue, opexec, job)())
+ self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(),
+ jqueue._JobProcessor.FINISHED)
# Check result
self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_CANCELED)
opexec = _FakeExecOpCodeForProc(queue, None, None)
# Run one opcode
- self.assertFalse(jqueue._JobProcessor(queue, opexec, job)())
+ self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(),
+ jqueue._JobProcessor.DEFER)
# Job goes back to queued
self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED)
self.assert_(success)
# Try processing another opcode (this will actually cancel the job)
- self.assert_(jqueue._JobProcessor(queue, opexec, job)())
+ self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(),
+ jqueue._JobProcessor.FINISHED)
# Check result
self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_CANCELED)
self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED)
for _ in range(successcount):
- self.assertFalse(jqueue._JobProcessor(queue, opexec, job)())
+ self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(),
+ jqueue._JobProcessor.DEFER)
self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED)
self.assertEqual(job.GetInfo(["opstatus"]),
self.assert_(job.ops_iter)
# Serialize and restore (simulates program restart)
- newjob = jqueue._QueuedJob.Restore(queue, job.Serialize())
+ newjob = jqueue._QueuedJob.Restore(queue, job.Serialize(), True, False)
self.assertFalse(newjob.ops_iter)
self._TestPartial(newjob, successcount)
if remaining == 0:
# Last opcode
- self.assert_(result)
+ self.assertEqual(result, jqueue._JobProcessor.FINISHED)
break
- self.assertFalse(result)
+ self.assertEqual(result, jqueue._JobProcessor.DEFER)
self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED)
self._GenericCheckJob(job)
# Calling the processor on a finished job should be a no-op
- self.assertTrue(jqueue._JobProcessor(queue, opexec, job)())
+ self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(),
+ jqueue._JobProcessor.FINISHED)
self.assertRaises(IndexError, queue.GetNextUpdate)
# ... also after being restored
- job2 = jqueue._QueuedJob.Restore(queue, job.Serialize())
+ job2 = jqueue._QueuedJob.Restore(queue, job.Serialize(), True, False)
# Calling the processor on a finished job should be a no-op
- self.assertTrue(jqueue._JobProcessor(queue, opexec, job2)())
+ self.assertEqual(jqueue._JobProcessor(queue, opexec, job2)(),
+ jqueue._JobProcessor.FINISHED)
self.assertRaises(IndexError, queue.GetNextUpdate)
def testProcessorOnRunningJob(self):
self.assertEqual(queue.GetNextUpdate(), (job, True))
self.assertRaises(IndexError, queue.GetNextUpdate)
self.assertFalse(queue.IsAcquired())
- self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITLOCK)
+ self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITING)
def _AfterStart(op, cbs):
self.assertEqual(queue.GetNextUpdate(), (job, True))
if remaining == 0:
# Last opcode
- self.assert_(result)
+ self.assertEqual(result, jqueue._JobProcessor.FINISHED)
break
- self.assertFalse(result)
+ self.assertEqual(result, jqueue._JobProcessor.DEFER)
self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED)
self._CheckLogMessages(job, logmsgcount)
# Serialize and restore (simulates program restart)
- newjob = jqueue._QueuedJob.Restore(queue, job.Serialize())
+ newjob = jqueue._QueuedJob.Restore(queue, job.Serialize(), True, False)
self._CheckLogMessages(newjob, logmsgcount)
# Check each message
self.assertEqual(queue.GetNextUpdate(), (job, True))
self.assertRaises(IndexError, queue.GetNextUpdate)
self.assertFalse(queue.IsAcquired())
- self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITLOCK)
+ self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITING)
self.assertFalse(job.cur_opctx)
def _AfterStart(op, cbs):
self.assertRaises(IndexError, queue.GetNextUpdate)
if idx == len(ops) - 1:
# Last opcode
- self.assert_(result)
+ self.assertEqual(result, jqueue._JobProcessor.FINISHED)
else:
- self.assertFalse(result)
+ self.assertEqual(result, jqueue._JobProcessor.DEFER)
self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED)
self.assert_(job.start_timestamp)
self._GenericCheckJob(job)
# Calling the processor on a finished job should be a no-op
- self.assertTrue(jqueue._JobProcessor(queue, opexec, job)())
+ self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(),
+ jqueue._JobProcessor.FINISHED)
self.assertRaises(IndexError, queue.GetNextUpdate)
def testJobDependency(self):
self.assertEqual(queue.GetNextUpdate(), (job, True))
self.assertRaises(IndexError, queue.GetNextUpdate)
self.assertFalse(queue.IsAcquired())
- self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITLOCK)
+ self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITING)
self.assertFalse(job.cur_opctx)
def _AfterStart(op, cbs):
if attempt < 5:
# Simulate waiting for other job
- self.assertTrue(result)
+ self.assertEqual(result, jqueue._JobProcessor.WAITDEP)
self.assertTrue(job.cur_opctx)
- self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITLOCK)
+ self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITING)
self.assertRaises(IndexError, depmgr.GetNextNotification)
self.assert_(job.start_timestamp)
self.assertFalse(job.end_timestamp)
continue
- if result:
+ if result == jqueue._JobProcessor.FINISHED:
# Last opcode
self.assertFalse(job.cur_opctx)
- self.assertEqual(queue.depmgr.GetNextNotification(), job_id)
break
self.assertRaises(IndexError, depmgr.GetNextNotification)
- self.assertFalse(result)
+ self.assertEqual(result, jqueue._JobProcessor.DEFER)
self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED)
self.assert_(job.start_timestamp)
self.assertFalse(job.end_timestamp)
self.assertFalse(depmgr.CountWaitingJobs())
# Calling the processor on a finished job should be a no-op
- self.assertTrue(jqueue._JobProcessor(queue, opexec, job)())
+ self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(),
+ jqueue._JobProcessor.FINISHED)
self.assertRaises(IndexError, queue.GetNextUpdate)
def testJobDependencyCancel(self):
self.assertEqual(queue.GetNextUpdate(), (job, True))
self.assertRaises(IndexError, queue.GetNextUpdate)
self.assertFalse(queue.IsAcquired())
- self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITLOCK)
+ self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITING)
self.assertFalse(job.cur_opctx)
def _AfterStart(op, cbs):
if attempt > 0 and attempt < 4:
# Simulate waiting for other job
- self.assertTrue(result)
+ self.assertEqual(result, jqueue._JobProcessor.WAITDEP)
self.assertTrue(job.cur_opctx)
- self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITLOCK)
+ self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITING)
self.assertRaises(IndexError, depmgr.GetNextNotification)
self.assert_(job.start_timestamp)
self.assertFalse(job.end_timestamp)
continue
- if result:
+ if result == jqueue._JobProcessor.FINISHED:
# Last opcode
self.assertFalse(job.cur_opctx)
- self.assertEqual(queue.depmgr.GetNextNotification(), job_id)
break
self.assertRaises(IndexError, depmgr.GetNextNotification)
- self.assertFalse(result)
+ self.assertEqual(result, jqueue._JobProcessor.DEFER)
self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED)
self.assert_(job.start_timestamp)
self.assertFalse(job.end_timestamp)
self.assertFalse(depmgr.CountPendingResults())
# Calling the processor on a finished job should be a no-op
- self.assertTrue(jqueue._JobProcessor(queue, opexec, job)())
+ self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(),
+ jqueue._JobProcessor.FINISHED)
self.assertRaises(IndexError, queue.GetNextUpdate)
def testJobDependencyWrongstatus(self):
self.assertEqual(queue.GetNextUpdate(), (job, True))
self.assertRaises(IndexError, queue.GetNextUpdate)
self.assertFalse(queue.IsAcquired())
- self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITLOCK)
+ self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITING)
self.assertFalse(job.cur_opctx)
def _AfterStart(op, cbs):
if attempt > 0 and attempt < 4:
# Simulate waiting for other job
- self.assertTrue(result)
+ self.assertEqual(result, jqueue._JobProcessor.WAITDEP)
self.assertTrue(job.cur_opctx)
- self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITLOCK)
+ self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITING)
self.assertRaises(IndexError, depmgr.GetNextNotification)
self.assert_(job.start_timestamp)
self.assertFalse(job.end_timestamp)
continue
- if result:
+ if result == jqueue._JobProcessor.FINISHED:
# Last opcode
self.assertFalse(job.cur_opctx)
- self.assertEqual(queue.depmgr.GetNextNotification(), job_id)
break
self.assertRaises(IndexError, depmgr.GetNextNotification)
- self.assertFalse(result)
+ self.assertEqual(result, jqueue._JobProcessor.DEFER)
self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED)
self.assert_(job.start_timestamp)
self.assertFalse(job.end_timestamp)
self.assertFalse(depmgr.CountPendingResults())
# Calling the processor on a finished job should be a no-op
- self.assertTrue(jqueue._JobProcessor(queue, opexec, job)())
+ self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(),
+ jqueue._JobProcessor.FINISHED)
self.assertRaises(IndexError, queue.GetNextUpdate)
+class TestEvaluateJobProcessorResult(unittest.TestCase):
+ def testFinished(self):
+ depmgr = _FakeDependencyManager()
+ job = _IdOnlyFakeJob(30953)
+ jqueue._EvaluateJobProcessorResult(depmgr, job,
+ jqueue._JobProcessor.FINISHED)
+ self.assertEqual(depmgr.GetNextNotification(), job.id)
+ self.assertRaises(IndexError, depmgr.GetNextNotification)
+
+ def testDefer(self):
+ depmgr = _FakeDependencyManager()
+ job = _IdOnlyFakeJob(11326, priority=5463)
+ try:
+ jqueue._EvaluateJobProcessorResult(depmgr, job,
+ jqueue._JobProcessor.DEFER)
+ except workerpool.DeferTask, err:
+ self.assertEqual(err.priority, 5463)
+ else:
+ self.fail("Didn't raise exception")
+ self.assertRaises(IndexError, depmgr.GetNextNotification)
+
+ def testWaitdep(self):
+ depmgr = _FakeDependencyManager()
+ job = _IdOnlyFakeJob(21317)
+ jqueue._EvaluateJobProcessorResult(depmgr, job,
+ jqueue._JobProcessor.WAITDEP)
+ self.assertRaises(IndexError, depmgr.GetNextNotification)
+
+ def testOther(self):
+ depmgr = _FakeDependencyManager()
+ job = _IdOnlyFakeJob(5813)
+ self.assertRaises(errors.ProgrammerError,
+ jqueue._EvaluateJobProcessorResult,
+ depmgr, job, "Other result")
+ self.assertRaises(IndexError, depmgr.GetNextNotification)
+
+
class _FakeTimeoutStrategy:
def __init__(self, timeouts):
self.timeouts = timeouts
self.assertRaises(IndexError, self.queue.GetNextUpdate)
self.assertFalse(self.queue.IsAcquired())
- self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITLOCK)
+ self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITING)
ts = self.timeout_strategy
result = proc(_nextop_fn=self._NextOpcode)
assert self.curop is not None
- if result or self.gave_lock:
+ if result == jqueue._JobProcessor.FINISHED or self.gave_lock:
# Got lock and/or job is done, result must've been written
self.assertFalse(job.cur_opctx)
self.assertEqual(self.queue.GetNextUpdate(), (job, True))
self.assertEqual(self.lock_acq_prio, job.ops[self.curop].priority)
self.assert_(job.ops[self.curop].exec_timestamp)
- if result:
+ if result == jqueue._JobProcessor.FINISHED:
self.assertFalse(job.cur_opctx)
break
- self.assertFalse(result)
+ self.assertEqual(result, jqueue._JobProcessor.DEFER)
if self.curop == 0:
self.assertEqual(job.ops[self.curop].start_timestamp,
self.assertEqual(job.cur_opctx._timeout_strategy._fn,
self.timeout_strategy.NextAttempt)
self.assertFalse(job.ops[self.curop].exec_timestamp)
- self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITLOCK)
+ self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITING)
# If priority has changed since acquiring locks, the job must've been
# updated
for op in job.ops))
# Calling the processor on a finished job should be a no-op
- self.assertTrue(jqueue._JobProcessor(self.queue, opexec, job)())
+ self.assertEqual(jqueue._JobProcessor(self.queue, opexec, job)(),
+ jqueue._JobProcessor.FINISHED)
self.assertRaises(IndexError, self.queue.GetNextUpdate)
-class TestJobDependencyManager(unittest.TestCase):
- class _FakeJob:
- def __init__(self, job_id):
- self.id = str(job_id)
+class _IdOnlyFakeJob:
+ def __init__(self, job_id, priority=NotImplemented):
+ self.id = str(job_id)
+ self._priority = priority
+
+ def CalcPriority(self):
+ return self._priority
+
+class TestJobDependencyManager(unittest.TestCase):
def setUp(self):
self._status = []
self._queue = []
return result
def _Enqueue(self, jobs):
+ self.assertFalse(self.jdm._lock.is_owned(),
+ msg=("Must not own manager lock while re-adding jobs"
+ " (potential deadlock)"))
self._queue.append(jobs)
def testNotFinalizedThenCancel(self):
- job = self._FakeJob(17697)
+ job = _IdOnlyFakeJob(17697)
job_id = str(28625)
self._status.append((job_id, constants.JOB_STATUS_RUNNING))
self.assertEqual(self.jdm._waiters, {
job_id: set([job]),
})
+ self.assertEqual(self.jdm.GetLockInfo([query.LQ_PENDING]), [
+ ("job/28625", None, None, [("job", [job.id])])
+ ])
self._status.append((job_id, constants.JOB_STATUS_CANCELED))
(result, _) = self.jdm.CheckAndRegister(job, job_id, [])
self.assertFalse(self._status)
self.assertFalse(self._queue)
self.assertFalse(self.jdm.JobWaiting(job))
+ self.assertFalse(self.jdm.GetLockInfo([query.LQ_PENDING]))
def testRequireCancel(self):
- job = self._FakeJob(5278)
+ job = _IdOnlyFakeJob(5278)
job_id = str(9610)
dep_status = [constants.JOB_STATUS_CANCELED]
- self._status.append((job_id, constants.JOB_STATUS_WAITLOCK))
+ self._status.append((job_id, constants.JOB_STATUS_WAITING))
(result, _) = self.jdm.CheckAndRegister(job, job_id, dep_status)
self.assertEqual(result, self.jdm.WAIT)
self.assertFalse(self._status)
self.assertEqual(self.jdm._waiters, {
job_id: set([job]),
})
+ self.assertEqual(self.jdm.GetLockInfo([query.LQ_PENDING]), [
+ ("job/9610", None, None, [("job", [job.id])])
+ ])
self._status.append((job_id, constants.JOB_STATUS_CANCELED))
(result, _) = self.jdm.CheckAndRegister(job, job_id, dep_status)
self.assertFalse(self._status)
self.assertFalse(self._queue)
self.assertFalse(self.jdm.JobWaiting(job))
+ self.assertFalse(self.jdm.GetLockInfo([query.LQ_PENDING]))
def testRequireError(self):
- job = self._FakeJob(21459)
+ job = _IdOnlyFakeJob(21459)
job_id = str(25519)
dep_status = [constants.JOB_STATUS_ERROR]
- self._status.append((job_id, constants.JOB_STATUS_WAITLOCK))
+ self._status.append((job_id, constants.JOB_STATUS_WAITING))
(result, _) = self.jdm.CheckAndRegister(job, job_id, dep_status)
self.assertEqual(result, self.jdm.WAIT)
self.assertFalse(self._status)
self.assertFalse(self._status)
self.assertFalse(self._queue)
self.assertFalse(self.jdm.JobWaiting(job))
+ self.assertFalse(self.jdm.GetLockInfo([query.LQ_PENDING]))
def testRequireMultiple(self):
dep_status = list(constants.JOBS_FINALIZED)
for end_status in dep_status:
- job = self._FakeJob(21343)
+ job = _IdOnlyFakeJob(21343)
job_id = str(14609)
- self._status.append((job_id, constants.JOB_STATUS_WAITLOCK))
+ self._status.append((job_id, constants.JOB_STATUS_WAITING))
(result, _) = self.jdm.CheckAndRegister(job, job_id, dep_status)
self.assertEqual(result, self.jdm.WAIT)
self.assertFalse(self._status)
self.assertEqual(self.jdm._waiters, {
job_id: set([job]),
})
+ self.assertEqual(self.jdm.GetLockInfo([query.LQ_PENDING]), [
+ ("job/14609", None, None, [("job", [job.id])])
+ ])
self._status.append((job_id, end_status))
(result, _) = self.jdm.CheckAndRegister(job, job_id, dep_status)
self.assertFalse(self._status)
self.assertFalse(self._queue)
self.assertFalse(self.jdm.JobWaiting(job))
+ self.assertFalse(self.jdm.GetLockInfo([query.LQ_PENDING]))
def testNotify(self):
- job = self._FakeJob(8227)
+ job = _IdOnlyFakeJob(8227)
job_id = str(4113)
self._status.append((job_id, constants.JOB_STATUS_RUNNING))
self.assertEqual(self._queue, [set([job])])
def testWrongStatus(self):
- job = self._FakeJob(10102)
+ job = _IdOnlyFakeJob(10102)
job_id = str(1271)
self._status.append((job_id, constants.JOB_STATUS_QUEUED))
self.assertFalse(self.jdm.JobWaiting(job))
def testCorrectStatus(self):
- job = self._FakeJob(24273)
+ job = _IdOnlyFakeJob(24273)
job_id = str(23885)
self._status.append((job_id, constants.JOB_STATUS_QUEUED))
self.assertFalse(self.jdm.JobWaiting(job))
def testFinalizedRightAway(self):
- job = self._FakeJob(224)
+ job = _IdOnlyFakeJob(224)
job_id = str(3081)
self._status.append((job_id, constants.JOB_STATUS_SUCCESS))
self.assertFalse(self._status)
self.assertFalse(self._queue)
+ def testMultipleWaiting(self):
+ # Use a deterministic random generator
+ rnd = random.Random(21402)
+
+ job_ids = map(str, rnd.sample(range(1, 10000), 150))
+
+ waiters = dict((job_ids.pop(),
+ set(map(_IdOnlyFakeJob,
+ [job_ids.pop()
+ for _ in range(rnd.randint(1, 20))])))
+ for _ in range(10))
+
+ # Ensure there are no duplicate job IDs
+ assert not utils.FindDuplicates(waiters.keys() +
+ [job.id
+ for jobs in waiters.values()
+ for job in jobs])
+
+ # Register all jobs as waiters
+ for job_id, job in [(job_id, job)
+ for (job_id, jobs) in waiters.items()
+ for job in jobs]:
+ self._status.append((job_id, constants.JOB_STATUS_QUEUED))
+ (result, _) = self.jdm.CheckAndRegister(job, job_id,
+ [constants.JOB_STATUS_SUCCESS])
+ self.assertEqual(result, self.jdm.WAIT)
+ self.assertFalse(self._status)
+ self.assertFalse(self._queue)
+ self.assertTrue(self.jdm.JobWaiting(job))
+
+ self.assertEqual(self.jdm._waiters, waiters)
+
+ def _MakeSet((name, mode, owner_names, pending)):
+ return (name, mode, owner_names,
+ [(pendmode, set(pend)) for (pendmode, pend) in pending])
+
+ def _CheckLockInfo():
+ info = self.jdm.GetLockInfo([query.LQ_PENDING])
+ self.assertEqual(sorted(map(_MakeSet, info)), sorted([
+ ("job/%s" % job_id, None, None,
+ [("job", set([job.id for job in jobs]))])
+ for job_id, jobs in waiters.items()
+ if jobs
+ ]))
+
+ _CheckLockInfo()
+
+ # Notify in random order
+ for job_id in rnd.sample(waiters, len(waiters)):
+ # Remove from pending waiter list
+ jobs = waiters.pop(job_id)
+ for job in jobs:
+ self._status.append((job_id, constants.JOB_STATUS_SUCCESS))
+ (result, _) = self.jdm.CheckAndRegister(job, job_id,
+ [constants.JOB_STATUS_SUCCESS])
+ self.assertEqual(result, self.jdm.CONTINUE)
+ self.assertFalse(self._status)
+ self.assertFalse(self._queue)
+ self.assertFalse(self.jdm.JobWaiting(job))
+
+ _CheckLockInfo()
+
+ self.assertFalse(self.jdm.GetLockInfo([query.LQ_PENDING]))
+
+ assert not waiters
+
def testSelfDependency(self):
- job = self._FakeJob(18937)
+ job = _IdOnlyFakeJob(18937)
self._status.append((job.id, constants.JOB_STATUS_SUCCESS))
(result, _) = self.jdm.CheckAndRegister(job, job.id, [])
self.assertEqual(result, self.jdm.ERROR)
def testJobDisappears(self):
- job = self._FakeJob(30540)
+ job = _IdOnlyFakeJob(30540)
job_id = str(23769)
def _FakeStatus(_):
(result, _) = jdm.CheckAndRegister(job, job_id, [])
self.assertEqual(result, self.jdm.ERROR)
self.assertFalse(jdm.JobWaiting(job))
+ self.assertFalse(jdm.GetLockInfo([query.LQ_PENDING]))
if __name__ == "__main__":