| #!/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 |
| |
| try: |
| # pylint: disable=E0611 |
| from pyinotify import pyinotify |
| except ImportError: |
| import pyinotify |
| |
| 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 _FailingWatchManager(pyinotify.WatchManager): |
| """Subclass of L{pyinotify.WatchManager} which always fails to register. |
| |
| """ |
| def add_watch(self, filename, mask): |
| assert mask == (pyinotify.EventsCodes.ALL_FLAGS["IN_MODIFY"] | |
| pyinotify.EventsCodes.ALL_FLAGS["IN_IGNORED"]) |
| |
| return { |
| filename: -1, |
| } |
| |
| |
| 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) |
| |
| def testNonExistentFile(self): |
| wfjc = jqueue._WaitForJobChangesHelper() |
| |
| filename = utils.PathJoin(self.tmpdir, "does-not-exist") |
| self.assertFalse(os.path.exists(filename)) |
| |
| result = wfjc(filename, self._LoadLostJob, ["status"], None, None, 1.0, |
| _waiter_cls=compat.partial(jqueue._JobChangesWaiter, |
| _waiter_cls=NotImplemented)) |
| self.assertTrue(result is None) |
| |
| def testInotifyError(self): |
| jobfile_waiter_cls = \ |
| compat.partial(jqueue._JobFileChangesWaiter, |
| _inotify_wm_cls=_FailingWatchManager) |
| |
| jobchange_waiter_cls = \ |
| compat.partial(jqueue._JobChangesWaiter, _waiter_cls=jobfile_waiter_cls) |
| |
| wfjc = jqueue._WaitForJobChangesHelper() |
| |
| # Test if failing to watch a job file (e.g. due to |
| # fs.inotify.max_user_watches being too low) raises errors.InotifyError |
| self.assertRaises(errors.InotifyError, wfjc, |
| self.filename, self._LoadWaitingJob, |
| ["status"], [constants.JOB_STATUS_WAITING], None, 1.0, |
| _waiter_cls=jobchange_waiter_cls) |
| |
| |
| 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.assertFalse(compat.any(hasattr(op.input, "priority") |
| 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.assertFalse(compat.any(hasattr(op.input, "priority") |
| for op in job.ops)) |
| 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.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_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) |
| |
| # Job is running, cancelling shouldn't be possible |
| (success, _) = job.Cancel() |
| self.assertFalse(success) |
| |
| opexec = _FakeExecOpCodeForProc(queue, _BeforeStart, _AfterStart) |
| |
| for idx in range(len(ops)): |
| self.assertRaises(IndexError, queue.GetNextUpdate) |
| result = jqueue._JobProcessor(queue, opexec, job)() |
| self.assertEqual(queue.GetNextUpdate(), (job, True)) |
| self.assertRaises(IndexError, queue.GetNextUpdate) |
| if idx == len(ops) - 1: |
| # Last opcode |
| self.assertEqual(result, jqueue._JobProcessor.FINISHED) |
| else: |
| 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.assertRaises(IndexError, queue.GetNextUpdate) |
| |
| self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_SUCCESS) |
| self.assertEqual(job.GetInfo(["status"]), [constants.JOB_STATUS_SUCCESS]) |
| self.assertEqual(job.GetInfo(["opresult"]), |
| [[op.input.result for op in job.ops]]) |
| self.assertEqual(job.GetInfo(["opstatus"]), |
| [len(job.ops) * [constants.OP_STATUS_SUCCESS]]) |
| self.assert_(compat.all(op.start_timestamp and op.end_timestamp |
| for op in job.ops)) |
| |
| self._GenericCheckJob(job) |
| |
| # Calling the processor on a finished job should be a no-op |
| self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(), |
| jqueue._JobProcessor.FINISHED) |
| self.assertRaises(IndexError, queue.GetNextUpdate) |
| |
| def testOpcodeError(self): |
| queue = _FakeQueueForProc() |
| |
| testdata = [ |
| (17077, 1, 0, 0), |
| (1782, 5, 2, 2), |
| (18179, 10, 9, 9), |
| (4744, 10, 3, 8), |
| (23816, 100, 39, 45), |
| ] |
| |
| for (job_id, opcount, failfrom, failto) in testdata: |
| # Prepare opcodes |
| ops = [opcodes.OpTestDummy(result="Res%s" % i, |
| fail=(failfrom <= i and |
| i <= failto)) |
| for i in range(opcount)] |
| |
| # Create job |
| job = self._CreateJob(queue, str(job_id), ops) |
| |
| opexec = _FakeExecOpCodeForProc(queue, None, None) |
| |
| for idx in range(len(ops)): |
| self.assertRaises(IndexError, queue.GetNextUpdate) |
| result = jqueue._JobProcessor(queue, opexec, job)() |
| # queued to waitlock |
| self.assertEqual(queue.GetNextUpdate(), (job, True)) |
| # waitlock to running |
| self.assertEqual(queue.GetNextUpdate(), (job, True)) |
| # Opcode result |
| self.assertEqual(queue.GetNextUpdate(), (job, True)) |
| self.assertRaises(IndexError, queue.GetNextUpdate) |
| |
| if idx in (failfrom, len(ops) - 1): |
| # Last opcode |
| self.assertEqual(result, jqueue._JobProcessor.FINISHED) |
| break |
| |
| self.assertEqual(result, jqueue._JobProcessor.DEFER) |
| |
| self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) |
| |
| self.assertRaises(IndexError, queue.GetNextUpdate) |
| |
| # Check job status |
| self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_ERROR) |
| self.assertEqual(job.GetInfo(["id"]), [job_id]) |
| self.assertEqual(job.GetInfo(["status"]), [constants.JOB_STATUS_ERROR]) |
| |
| # Check opcode status |
| data = zip(job.ops, |
| job.GetInfo(["opstatus"])[0], |
| job.GetInfo(["opresult"])[0]) |
| |
| for idx, (op, opstatus, opresult) in enumerate(data): |
| if idx < failfrom: |
| assert not op.input.fail |
| self.assertEqual(opstatus, constants.OP_STATUS_SUCCESS) |
| self.assertEqual(opresult, op.input.result) |
| elif idx <= failto: |
| assert op.input.fail |
| self.assertEqual(opstatus, constants.OP_STATUS_ERROR) |
| self.assertRaises(errors.OpExecError, errors.MaybeRaise, opresult) |
| else: |
| assert not op.input.fail |
| self.assertEqual(opstatus, constants.OP_STATUS_ERROR) |
| self.assertRaises(errors.OpExecError, errors.MaybeRaise, opresult) |
| |
| self.assert_(compat.all(op.start_timestamp and op.end_timestamp |
| for op in job.ops[:failfrom])) |
| |
| self._GenericCheckJob(job) |
| |
| # Calling the processor on a finished job should be a no-op |
| self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(), |
| jqueue._JobProcessor.FINISHED) |
| self.assertRaises(IndexError, queue.GetNextUpdate) |
| |
| def testCancelWhileInQueue(self): |
| queue = _FakeQueueForProc() |
| |
| ops = [opcodes.OpTestDummy(result="Res%s" % i, fail=False) |
| for i in range(5)] |
| |
| # Create job |
| job_id = 17045 |
| job = self._CreateJob(queue, job_id, ops) |
| |
| self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) |
| |
| # Mark as cancelled |
| (success, _) = job.Cancel() |
| self.assert_(success) |
| |
| self.assertRaises(IndexError, queue.GetNextUpdate) |
| |
| self.assertFalse(job.start_timestamp) |
| self.assertTrue(job.end_timestamp) |
| self.assert_(compat.all(op.status == constants.OP_STATUS_CANCELED |
| for op in job.ops)) |
| |
| # Serialize to check for differences |
| before_proc = job.Serialize() |
| |
| # Simulate processor called in workerpool |
| opexec = _FakeExecOpCodeForProc(queue, None, None) |
| self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(), |
| jqueue._JobProcessor.FINISHED) |
| |
| # Check result |
| self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_CANCELED) |
| self.assertEqual(job.GetInfo(["status"]), [constants.JOB_STATUS_CANCELED]) |
| self.assertFalse(job.start_timestamp) |
| self.assertTrue(job.end_timestamp) |
| self.assertFalse(compat.any(op.start_timestamp or op.end_timestamp |
| for op in job.ops)) |
| self.assertEqual(job.GetInfo(["opstatus", "opresult"]), |
| [[constants.OP_STATUS_CANCELED for _ in job.ops], |
| ["Job canceled by request" for _ in job.ops]]) |
| |
| # Must not have changed or written |
| self.assertEqual(before_proc, job.Serialize()) |
| self.assertRaises(IndexError, queue.GetNextUpdate) |
| |
| def testCancelWhileWaitlockInQueue(self): |
| queue = _FakeQueueForProc() |
| |
| ops = [opcodes.OpTestDummy(result="Res%s" % i, fail=False) |
| for i in range(5)] |
| |
| # Create job |
| job_id = 8645 |
| job = self._CreateJob(queue, job_id, ops) |
| |
| self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) |
| |
| job.ops[0].status = constants.OP_STATUS_WAITING |
| |
| assert len(job.ops) == 5 |
| |
| self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITING) |
| |
| # Mark as cancelling |
| (success, _) = job.Cancel() |
| self.assert_(success) |
| |
| self.assertRaises(IndexError, queue.GetNextUpdate) |
| |
| self.assert_(compat.all(op.status == constants.OP_STATUS_CANCELING |
| for op in job.ops)) |
| |
| opexec = _FakeExecOpCodeForProc(queue, None, None) |
| self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(), |
| jqueue._JobProcessor.FINISHED) |
| |
| # Check result |
| self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_CANCELED) |
| self.assertEqual(job.GetInfo(["status"]), [constants.JOB_STATUS_CANCELED]) |
| self.assertFalse(job.start_timestamp) |
| self.assert_(job.end_timestamp) |
| self.assertFalse(compat.any(op.start_timestamp or op.end_timestamp |
| for op in job.ops)) |
| self.assertEqual(job.GetInfo(["opstatus", "opresult"]), |
| [[constants.OP_STATUS_CANCELED for _ in job.ops], |
| ["Job canceled by request" for _ in job.ops]]) |
| |
| def testCancelWhileWaitlock(self): |
| queue = _FakeQueueForProc() |
| |
| ops = [opcodes.OpTestDummy(result="Res%s" % i, fail=False) |
| for i in range(5)] |
| |
| # Create job |
| job_id = 11009 |
| job = self._CreateJob(queue, job_id, ops) |
| |
| self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) |
| |
| 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) |
| |
| # Mark as cancelled |
| (success, _) = job.Cancel() |
| self.assert_(success) |
| |
| self.assert_(compat.all(op.status == constants.OP_STATUS_CANCELING |
| for op in job.ops)) |
| self.assertRaises(IndexError, queue.GetNextUpdate) |
| |
| 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) |
| |
| opexec = _FakeExecOpCodeForProc(queue, _BeforeStart, _AfterStart) |
| |
| self.assertRaises(IndexError, queue.GetNextUpdate) |
| self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(), |
| jqueue._JobProcessor.FINISHED) |
| self.assertEqual(queue.GetNextUpdate(), (job, True)) |
| self.assertRaises(IndexError, queue.GetNextUpdate) |
| |
| # Check result |
| self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_CANCELED) |
| self.assertEqual(job.GetInfo(["status"]), [constants.JOB_STATUS_CANCELED]) |
| self.assert_(job.start_timestamp) |
| self.assert_(job.end_timestamp) |
| self.assertFalse(compat.all(op.start_timestamp and op.end_timestamp |
| for op in job.ops)) |
| self.assertEqual(job.GetInfo(["opstatus", "opresult"]), |
| [[constants.OP_STATUS_CANCELED for _ in job.ops], |
| ["Job canceled by request" for _ in job.ops]]) |
| |
| def _TestCancelWhileSomething(self, cb): |
| queue = _FakeQueueForProc() |
| |
| ops = [opcodes.OpTestDummy(result="Res%s" % i, fail=False) |
| for i in range(5)] |
| |
| # Create job |
| job_id = 24314 |
| job = self._CreateJob(queue, job_id, ops) |
| |
| self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) |
| |
| def _BeforeStart(timeout, priority): |
| self.assertFalse(queue.IsAcquired()) |
| self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITING) |
| |
| # Mark as cancelled |
| (success, _) = job.Cancel() |
| self.assert_(success) |
| |
| self.assert_(compat.all(op.status == constants.OP_STATUS_CANCELING |
| for op in job.ops)) |
| |
| cb(queue) |
| |
| def _AfterStart(op, cbs): |
| self.fail("Should not reach this") |
| |
| opexec = _FakeExecOpCodeForProc(queue, _BeforeStart, _AfterStart) |
| |
| self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(), |
| jqueue._JobProcessor.FINISHED) |
| |
| # Check result |
| self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_CANCELED) |
| self.assertEqual(job.GetInfo(["status"]), [constants.JOB_STATUS_CANCELED]) |
| self.assert_(job.start_timestamp) |
| self.assert_(job.end_timestamp) |
| self.assertFalse(compat.all(op.start_timestamp and op.end_timestamp |
| for op in job.ops)) |
| self.assertEqual(job.GetInfo(["opstatus", "opresult"]), |
| [[constants.OP_STATUS_CANCELED for _ in job.ops], |
| ["Job canceled by request" for _ in job.ops]]) |
| |
| return queue |
| |
| def testCancelWhileWaitlockWithTimeout(self): |
| def fn(_): |
| # Fake an acquire attempt timing out |
| raise mcpu.LockAcquireTimeout() |
| |
| self._TestCancelWhileSomething(fn) |
| |
| def testCancelDuringQueueShutdown(self): |
| queue = self._TestCancelWhileSomething(lambda q: q.StopAcceptingJobs()) |
| self.assertFalse(queue.AcceptingJobsUnlocked()) |
| |
| def testCancelWhileRunning(self): |
| # Tests canceling a job with finished opcodes and more, unprocessed ones |
| queue = _FakeQueueForProc() |
| |
| ops = [opcodes.OpTestDummy(result="Res%s" % i, fail=False) |
| for i in range(3)] |
| |
| # Create job |
| job_id = 28492 |
| job = self._CreateJob(queue, job_id, ops) |
| |
| self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) |
| |
| opexec = _FakeExecOpCodeForProc(queue, None, None) |
| |
| # Run one opcode |
| self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(), |
| jqueue._JobProcessor.DEFER) |
| |
| # Job goes back to queued |
| self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) |
| self.assertEqual(job.GetInfo(["opstatus", "opresult"]), |
| [[constants.OP_STATUS_SUCCESS, |
| constants.OP_STATUS_QUEUED, |
| constants.OP_STATUS_QUEUED], |
| ["Res0", None, None]]) |
| |
| # Mark as cancelled |
| (success, _) = job.Cancel() |
| self.assert_(success) |
| |
| # Try processing another opcode (this will actually cancel the job) |
| self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(), |
| jqueue._JobProcessor.FINISHED) |
| |
| # Check result |
| self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_CANCELED) |
| self.assertEqual(job.GetInfo(["id"]), [job_id]) |
| self.assertEqual(job.GetInfo(["status"]), [constants.JOB_STATUS_CANCELED]) |
| self.assertEqual(job.GetInfo(["opstatus", "opresult"]), |
| [[constants.OP_STATUS_SUCCESS, |
| constants.OP_STATUS_CANCELED, |
| constants.OP_STATUS_CANCELED], |
| ["Res0", "Job canceled by request", |
| "Job canceled by request"]]) |
| |
| def _TestQueueShutdown(self, queue, opexec, job, runcount): |
| self.assertTrue(queue.AcceptingJobsUnlocked()) |
| |
| # Simulate shutdown |
| queue.StopAcceptingJobs() |
| |
| self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(), |
| jqueue._JobProcessor.DEFER) |
| |
| # Check result |
| self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) |
| self.assertEqual(job.GetInfo(["status"]), [constants.JOB_STATUS_QUEUED]) |
| self.assertFalse(job.cur_opctx) |
| self.assertTrue(job.start_timestamp) |
| self.assertFalse(job.end_timestamp) |
| self.assertEqual(job.start_timestamp, job.ops[0].start_timestamp) |
| self.assertTrue(compat.all(op.start_timestamp and op.end_timestamp |
| for op in job.ops[:runcount])) |
| self.assertFalse(job.ops[runcount].end_timestamp) |
| self.assertFalse(compat.any(op.start_timestamp or op.end_timestamp |
| for op in job.ops[(runcount + 1):])) |
| self.assertEqual(job.GetInfo(["opstatus", "opresult"]), |
| [(([constants.OP_STATUS_SUCCESS] * runcount) + |
| ([constants.OP_STATUS_QUEUED] * |
| (len(job.ops) - runcount))), |
| (["Res%s" % i for i in range(runcount)] + |
| ([None] * (len(job.ops) - runcount)))]) |
| |
| # Must have been written and replicated |
| self.assertEqual(queue.GetNextUpdate(), (job, True)) |
| self.assertRaises(IndexError, queue.GetNextUpdate) |
| |
| def testQueueShutdownWhileRunning(self): |
| # Tests shutting down the queue while a job is running |
| queue = _FakeQueueForProc() |
| |
| ops = [opcodes.OpTestDummy(result="Res%s" % i, fail=False) |
| for i in range(3)] |
| |
| # Create job |
| job_id = 2718211587 |
| job = self._CreateJob(queue, job_id, ops) |
| |
| self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) |
| |
| opexec = _FakeExecOpCodeForProc(queue, None, None) |
| |
| self.assertRaises(IndexError, queue.GetNextUpdate) |
| |
| # Run one opcode |
| self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(), |
| jqueue._JobProcessor.DEFER) |
| |
| # Job goes back to queued |
| self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) |
| self.assertEqual(job.GetInfo(["opstatus", "opresult"]), |
| [[constants.OP_STATUS_SUCCESS, |
| constants.OP_STATUS_QUEUED, |
| constants.OP_STATUS_QUEUED], |
| ["Res0", None, None]]) |
| self.assertFalse(job.cur_opctx) |
| |
| # Writes for waiting, running and result |
| for _ in range(3): |
| self.assertEqual(queue.GetNextUpdate(), (job, True)) |
| |
| # Run second opcode |
| self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(), |
| jqueue._JobProcessor.DEFER) |
| |
| # Job goes back to queued |
| self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) |
| self.assertEqual(job.GetInfo(["opstatus", "opresult"]), |
| [[constants.OP_STATUS_SUCCESS, |
| constants.OP_STATUS_SUCCESS, |
| constants.OP_STATUS_QUEUED], |
| ["Res0", "Res1", None]]) |
| self.assertFalse(job.cur_opctx) |
| |
| # Writes for waiting, running and result |
| for _ in range(3): |
| self.assertEqual(queue.GetNextUpdate(), (job, True)) |
| |
| self._TestQueueShutdown(queue, opexec, job, 2) |
| |
| def testQueueShutdownWithLockTimeout(self): |
| # Tests shutting down while a lock acquire times out |
| queue = _FakeQueueForProc() |
| |
| ops = [opcodes.OpTestDummy(result="Res%s" % i, fail=False) |
| for i in range(3)] |
| |
| # Create job |
| job_id = 1304231178 |
| job = self._CreateJob(queue, job_id, ops) |
| |
| self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) |
| |
| acquire_timeout = False |
| |
| def _BeforeStart(timeout, priority): |
| self.assertFalse(queue.IsAcquired()) |
| self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITING) |
| if acquire_timeout: |
| raise mcpu.LockAcquireTimeout() |
| |
| opexec = _FakeExecOpCodeForProc(queue, _BeforeStart, None) |
| |
| self.assertRaises(IndexError, queue.GetNextUpdate) |
| |
| # Run one opcode |
| self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(), |
| jqueue._JobProcessor.DEFER) |
| |
| # Job goes back to queued |
| self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) |
| self.assertEqual(job.GetInfo(["opstatus", "opresult"]), |
| [[constants.OP_STATUS_SUCCESS, |
| constants.OP_STATUS_QUEUED, |
| constants.OP_STATUS_QUEUED], |
| ["Res0", None, None]]) |
| self.assertFalse(job.cur_opctx) |
| |
| # Writes for waiting, running and result |
| for _ in range(3): |
| self.assertEqual(queue.GetNextUpdate(), (job, True)) |
| |
| # The next opcode should have expiring lock acquires |
| acquire_timeout = True |
| |
| self._TestQueueShutdown(queue, opexec, job, 1) |
| |
| def testQueueShutdownWhileInQueue(self): |
| # This should never happen in reality (no new jobs are started by the |
| # workerpool once a shutdown has been initiated), but it's better to test |
| # the job processor for this scenario |
| queue = _FakeQueueForProc() |
| |
| ops = [opcodes.OpTestDummy(result="Res%s" % i, fail=False) |
| for i in range(5)] |
| |
| # Create job |
| job_id = 2031 |
| job = self._CreateJob(queue, job_id, ops) |
| |
| self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) |
| self.assertRaises(IndexError, queue.GetNextUpdate) |
| |
| self.assertFalse(job.start_timestamp) |
| self.assertFalse(job.end_timestamp) |
| self.assertTrue(compat.all(op.status == constants.OP_STATUS_QUEUED |
| for op in job.ops)) |
| |
| opexec = _FakeExecOpCodeForProc(queue, None, None) |
| self._TestQueueShutdown(queue, opexec, job, 0) |
| |
| def testQueueShutdownWhileWaitlockInQueue(self): |
| queue = _FakeQueueForProc() |
| |
| ops = [opcodes.OpTestDummy(result="Res%s" % i, fail=False) |
| for i in range(5)] |
| |
| # Create job |
| job_id = 53125685 |
| job = self._CreateJob(queue, job_id, ops) |
| |
| self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) |
| |
| job.ops[0].status = constants.OP_STATUS_WAITING |
| |
| assert len(job.ops) == 5 |
| |
| self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITING) |
| |
| self.assertRaises(IndexError, queue.GetNextUpdate) |
| |
| opexec = _FakeExecOpCodeForProc(queue, None, None) |
| self._TestQueueShutdown(queue, opexec, job, 0) |
| |
| def testPartiallyRun(self): |
| # Tests calling the processor on a job that's been partially run before the |
| # program was restarted |
| queue = _FakeQueueForProc() |
| |
| opexec = _FakeExecOpCodeForProc(queue, None, None) |
| |
| for job_id, successcount in [(30697, 1), (2552, 4), (12489, 9)]: |
| ops = [opcodes.OpTestDummy(result="Res%s" % i, fail=False) |
| for i in range(10)] |
| |
| # Create job |
| job = self._CreateJob(queue, job_id, ops) |
| |
| self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) |
| |
| for _ in range(successcount): |
| self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(), |
| jqueue._JobProcessor.DEFER) |
| |
| self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) |
| self.assertEqual(job.GetInfo(["opstatus"]), |
| [[constants.OP_STATUS_SUCCESS |
| for _ in range(successcount)] + |
| [constants.OP_STATUS_QUEUED |
| for _ in range(len(ops) - successcount)]]) |
| |
| self.assert_(job.ops_iter) |
| |
| # Serialize and restore (simulates program restart) |
| newjob = jqueue._QueuedJob.Restore(queue, job.Serialize(), True, False) |
| self.assertFalse(newjob.ops_iter) |
| self._TestPartial(newjob, successcount) |
| |
| def _TestPartial(self, job, successcount): |
| self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) |
| self.assertEqual(job.start_timestamp, job.ops[0].start_timestamp) |
| |
| queue = _FakeQueueForProc() |
| opexec = _FakeExecOpCodeForProc(queue, None, None) |
| |
| for remaining in reversed(range(len(job.ops) - successcount)): |
| result = jqueue._JobProcessor(queue, opexec, job)() |
| self.assertEqual(queue.GetNextUpdate(), (job, True)) |
| self.assertEqual(queue.GetNextUpdate(), (job, True)) |
| self.assertEqual(queue.GetNextUpdate(), (job, True)) |
| self.assertRaises(IndexError, queue.GetNextUpdate) |
| |
| if remaining == 0: |
| # Last opcode |
| self.assertEqual(result, jqueue._JobProcessor.FINISHED) |
| break |
| |
| self.assertEqual(result, jqueue._JobProcessor.DEFER) |
| |
| self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) |
| |
| self.assertRaises(IndexError, queue.GetNextUpdate) |
| self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_SUCCESS) |
| self.assertEqual(job.GetInfo(["status"]), [constants.JOB_STATUS_SUCCESS]) |
| self.assertEqual(job.GetInfo(["opresult"]), |
| [[op.input.result for op in job.ops]]) |
| self.assertEqual(job.GetInfo(["opstatus"]), |
| [[constants.OP_STATUS_SUCCESS for _ in job.ops]]) |
| self.assert_(compat.all(op.start_timestamp and op.end_timestamp |
| for op in job.ops)) |
| |
| self._GenericCheckJob(job) |
| |
| # Calling the processor on a finished job should be a no-op |
| 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(), True, False) |
| # Calling the processor on a finished job should be a no-op |
| self.assertEqual(jqueue._JobProcessor(queue, opexec, job2)(), |
| jqueue._JobProcessor.FINISHED) |
| self.assertRaises(IndexError, queue.GetNextUpdate) |
| |
| def testProcessorOnRunningJob(self): |
| ops = [opcodes.OpTestDummy(result="result", fail=False)] |
| |
| queue = _FakeQueueForProc() |
| opexec = _FakeExecOpCodeForProc(queue, None, None) |
| |
| # Create job |
| job = self._CreateJob(queue, 9571, ops) |
| |
| self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) |
| |
| job.ops[0].status = constants.OP_STATUS_RUNNING |
| |
| assert len(job.ops) == 1 |
| |
| self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_RUNNING) |
| |
| # Calling on running job must fail |
| self.assertRaises(errors.ProgrammerError, |
| jqueue._JobProcessor(queue, opexec, job)) |
| |
| def testLogMessages(self): |
| # Tests the "Feedback" callback function |
| queue = _FakeQueueForProc() |
| |
| messages = { |
| 1: [ |
| (None, "Hello"), |
| (None, "World"), |
| (constants.ELOG_MESSAGE, "there"), |
| ], |
| 4: [ |
| (constants.ELOG_JQUEUE_TEST, (1, 2, 3)), |
| (constants.ELOG_JQUEUE_TEST, ("other", "type")), |
| ], |
| } |
| ops = [opcodes.OpTestDummy(result="Logtest%s" % i, fail=False, |
| messages=messages.get(i, [])) |
| for i in range(5)] |
| |
| # Create job |
| job = self._CreateJob(queue, 29386, 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) |
| |
| 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.assertRaises(AssertionError, cbs.Feedback, |
| "too", "many", "arguments") |
| |
| for (log_type, msg) in op.messages: |
| self.assertRaises(IndexError, queue.GetNextUpdate) |
| if log_type: |
| cbs.Feedback(log_type, msg) |
| else: |
| cbs.Feedback(msg) |
| # Check for job update without replication |
| self.assertEqual(queue.GetNextUpdate(), (job, False)) |
| self.assertRaises(IndexError, queue.GetNextUpdate) |
| |
| opexec = _FakeExecOpCodeForProc(queue, _BeforeStart, _AfterStart) |
| |
| for remaining in reversed(range(len(job.ops))): |
| self.assertRaises(IndexError, queue.GetNextUpdate) |
| result = jqueue._JobProcessor(queue, opexec, job)() |
| self.assertEqual(queue.GetNextUpdate(), (job, True)) |
| self.assertRaises(IndexError, queue.GetNextUpdate) |
| |
| if remaining == 0: |
| # Last opcode |
| self.assertEqual(result, jqueue._JobProcessor.FINISHED) |
| break |
| |
| self.assertEqual(result, jqueue._JobProcessor.DEFER) |
| |
| self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) |
| |
| self.assertRaises(IndexError, queue.GetNextUpdate) |
| |
| self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_SUCCESS) |
| self.assertEqual(job.GetInfo(["opresult"]), |
| [[op.input.result for op in job.ops]]) |
| |
| logmsgcount = sum(len(m) for m in messages.values()) |
| |
| self._CheckLogMessages(job, logmsgcount) |
| |
| # Serialize and restore (simulates program restart) |
| newjob = jqueue._QueuedJob.Restore(queue, job.Serialize(), True, False) |
| self._CheckLogMessages(newjob, logmsgcount) |
| |
| # Check each message |
| prevserial = -1 |
| for idx, oplog in enumerate(job.GetInfo(["oplog"])[0]): |
| for (serial, timestamp, log_type, msg) in oplog: |
| (exptype, expmsg) = messages.get(idx).pop(0) |
| if exptype: |
| self.assertEqual(log_type, exptype) |
| else: |
| self.assertEqual(log_type, constants.ELOG_MESSAGE) |
| self.assertEqual(expmsg, msg) |
| self.assert_(serial > prevserial) |
| prevserial = serial |
| |
| def _CheckLogMessages(self, job, count): |
| # Check serial |
| self.assertEqual(job.log_serial, count) |
| |
| # No filter |
| self.assertEqual(job.GetLogEntries(None), |
| [entry for entries in job.GetInfo(["oplog"])[0] if entries |
| for entry in entries]) |
| |
| # Filter with serial |
| assert count > 3 |
| self.assert_(job.GetLogEntries(3)) |
| self.assertEqual(job.GetLogEntries(3), |
| [entry for entries in job.GetInfo(["oplog"])[0] if entries |
| for entry in entries][3:]) |
| |
| # No log message after highest serial |
| self.assertFalse(job.GetLogEntries(count)) |
| self.assertFalse(job.GetLogEntries(count + 3)) |
| |
| def testSubmitManyJobs(self): |
| queue = _FakeQueueForProc() |
| |
| job_id = 15656 |
| ops = [ |
| opcodes.OpTestDummy(result="Res0", fail=False, |
| submit_jobs=[]), |
| opcodes.OpTestDummy(result="Res1", fail=False, |
| submit_jobs=[ |
| [opcodes.OpTestDummy(result="r1j0", fail=False)], |
| ]), |
| opcodes.OpTestDummy(result="Res2", fail=False, |
| submit_jobs=[ |
| [opcodes.OpTestDummy(result="r2j0o0", fail=False), |
| opcodes.OpTestDummy(result="r2j0o1", fail=False), |
| opcodes.OpTestDummy(result="r2j0o2", fail=False), |
| opcodes.OpTestDummy(result="r2j0o3", fail=False)], |
| [opcodes.OpTestDummy(result="r2j1", fail=False)], |
| [opcodes.OpTestDummy(result="r2j3o0", fail=False), |
| opcodes.OpTestDummy(result="r2j3o1", fail=False)], |
| ]), |
| ] |
| |
| # 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) |
| |
| # Job is running, cancelling shouldn't be possible |
| (success, _) = job.Cancel() |
| self.assertFalse(success) |
| |
| opexec = _FakeExecOpCodeForProc(queue, _BeforeStart, _AfterStart) |
| |
| for idx in range(len(ops)): |
| self.assertRaises(IndexError, queue.GetNextUpdate) |
| result = jqueue._JobProcessor(queue, opexec, job)() |
| self.assertEqual(queue.GetNextUpdate(), (job, True)) |
| self.assertRaises(IndexError, queue.GetNextUpdate) |
| if idx == len(ops) - 1: |
| # Last opcode |
| self.assertEqual(result, jqueue._JobProcessor.FINISHED) |
| else: |
| 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.assertRaises(IndexError, queue.GetNextUpdate) |
| |
| for idx, submitted_ops in enumerate(job_ops |
| for op in ops |
| for job_ops in op.submit_jobs): |
| self.assertEqual(queue.GetNextSubmittedJob(), |
| (1000 + idx, submitted_ops)) |
| self.assertRaises(IndexError, queue.GetNextSubmittedJob) |
| |
| self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_SUCCESS) |
| self.assertEqual(job.GetInfo(["status"]), [constants.JOB_STATUS_SUCCESS]) |
| self.assertEqual(job.GetInfo(["opresult"]), |
| [[[], [1000], [1001, 1002, 1003]]]) |
| self.assertEqual(job.GetInfo(["opstatus"]), |
| [len(job.ops) * [constants.OP_STATUS_SUCCESS]]) |
| |
| self._GenericCheckJob(job) |
| |
| # Calling the processor on a finished job should be a no-op |
| self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(), |
| jqueue._JobProcessor.FINISHED) |
| self.assertRaises(IndexError, queue.GetNextUpdate) |
| |
| def testJobDependency(self): |
| depmgr = _FakeDependencyManager() |
| queue = _FakeQueueForProc(depmgr=depmgr) |
| |
| self.assertEqual(queue.depmgr, depmgr) |
| |
| prev_job_id = 22113 |
| prev_job_id2 = 28102 |
| job_id = 29929 |
| ops = [ |
| opcodes.OpTestDummy(result="Res0", fail=False, |
| depends=[ |
| [prev_job_id2, None], |
| [prev_job_id, None], |
| ]), |
| opcodes.OpTestDummy(result="Res1", fail=False), |
| ] |
| |
| # Create job |
| job = self._CreateJob(queue, job_id, ops) |
| |
| def _BeforeStart(timeout, priority): |
| if attempt == 0 or attempt > 5: |
| # Job should only be updated when it wasn't waiting for another job |
| 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) |
| |
| # Job is running, cancelling shouldn't be possible |
| (success, _) = job.Cancel() |
| self.assertFalse(success) |
| |
| opexec = _FakeExecOpCodeForProc(queue, _BeforeStart, _AfterStart) |
| |
| self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) |
| |
| counter = itertools.count() |
| while True: |
| attempt = counter.next() |
| |
| self.assertRaises(IndexError, queue.GetNextUpdate) |
| self.assertRaises(IndexError, depmgr.GetNextNotification) |
| |
| if attempt < 2: |
| depmgr.AddCheckResult(job, prev_job_id2, None, |
| (jqueue._JobDependencyManager.WAIT, "wait2")) |
| elif attempt == 2: |
| depmgr.AddCheckResult(job, prev_job_id2, None, |
| (jqueue._JobDependencyManager.CONTINUE, "cont")) |
| # The processor will ask for the next dependency immediately |
| depmgr.AddCheckResult(job, prev_job_id, None, |
| (jqueue._JobDependencyManager.WAIT, "wait")) |
| elif attempt < 5: |
| depmgr.AddCheckResult(job, prev_job_id, None, |
| (jqueue._JobDependencyManager.WAIT, "wait")) |
| elif attempt == 5: |
| depmgr.AddCheckResult(job, prev_job_id, None, |
| (jqueue._JobDependencyManager.CONTINUE, "cont")) |
| if attempt == 2: |
| self.assertEqual(depmgr.CountPendingResults(), 2) |
| elif attempt > 5: |
| self.assertEqual(depmgr.CountPendingResults(), 0) |
| else: |
| self.assertEqual(depmgr.CountPendingResults(), 1) |
| |
| result = jqueue._JobProcessor(queue, opexec, job)() |
| if attempt == 0 or attempt >= 5: |
| # Job should only be updated if there was an actual change |
| self.assertEqual(queue.GetNextUpdate(), (job, True)) |
| self.assertRaises(IndexError, queue.GetNextUpdate) |
| self.assertFalse(depmgr.CountPendingResults()) |
| |
| if attempt < 5: |
| # Simulate waiting for other job |
| self.assertEqual(result, jqueue._JobProcessor.WAITDEP) |
| self.assertTrue(job.cur_opctx) |
| 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 == jqueue._JobProcessor.FINISHED: |
| # Last opcode |
| self.assertFalse(job.cur_opctx) |
| break |
| |
| self.assertRaises(IndexError, depmgr.GetNextNotification) |
| |
| 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.assertEqual(job.CalcStatus(), constants.JOB_STATUS_SUCCESS) |
| self.assertEqual(job.GetInfo(["status"]), [constants.JOB_STATUS_SUCCESS]) |
| self.assertEqual(job.GetInfo(["opresult"]), |
| [[op.input.result for op in job.ops]]) |
| self.assertEqual(job.GetInfo(["opstatus"]), |
| [len(job.ops) * [constants.OP_STATUS_SUCCESS]]) |
| self.assertTrue(compat.all(op.start_timestamp and op.end_timestamp |
| for op in job.ops)) |
| |
| self._GenericCheckJob(job) |
| |
| self.assertRaises(IndexError, queue.GetNextUpdate) |
| self.assertRaises(IndexError, depmgr.GetNextNotification) |
| self.assertFalse(depmgr.CountPendingResults()) |
| self.assertFalse(depmgr.CountWaitingJobs()) |
| |
| # Calling the processor on a finished job should be a no-op |
| self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(), |
| jqueue._JobProcessor.FINISHED) |
| self.assertRaises(IndexError, queue.GetNextUpdate) |
| |
| def testJobDependencyCancel(self): |
| depmgr = _FakeDependencyManager() |
| queue = _FakeQueueForProc(depmgr=depmgr) |
| |
| self.assertEqual(queue.depmgr, depmgr) |
| |
| prev_job_id = 13623 |
| job_id = 30876 |
| ops = [ |
| opcodes.OpTestDummy(result="Res0", fail=False), |
| opcodes.OpTestDummy(result="Res1", fail=False, |
| depends=[ |
| [prev_job_id, None], |
| ]), |
| opcodes.OpTestDummy(result="Res2", fail=False), |
| ] |
| |
| # Create job |
| job = self._CreateJob(queue, job_id, ops) |
| |
| def _BeforeStart(timeout, priority): |
| if attempt == 0 or attempt > 5: |
| # Job should only be updated when it wasn't waiting for another job |
| 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) |
| |
| # Job is running, cancelling shouldn't be possible |
| (success, _) = job.Cancel() |
| self.assertFalse(success) |
| |
| opexec = _FakeExecOpCodeForProc(queue, _BeforeStart, _AfterStart) |
| |
| self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) |
| |
| counter = itertools.count() |
| while True: |
| attempt = counter.next() |
| |
| self.assertRaises(IndexError, queue.GetNextUpdate) |
| self.assertRaises(IndexError, depmgr.GetNextNotification) |
| |
| if attempt == 0: |
| # This will handle the first opcode |
| pass |
| elif attempt < 4: |
| depmgr.AddCheckResult(job, prev_job_id, None, |
| (jqueue._JobDependencyManager.WAIT, "wait")) |
| elif attempt == 4: |
| # Other job was cancelled |
| depmgr.AddCheckResult(job, prev_job_id, None, |
| (jqueue._JobDependencyManager.CANCEL, "cancel")) |
| |
| if attempt == 0: |
| self.assertEqual(depmgr.CountPendingResults(), 0) |
| else: |
| self.assertEqual(depmgr.CountPendingResults(), 1) |
| |
| result = jqueue._JobProcessor(queue, opexec, job)() |
| if attempt <= 1 or attempt >= 4: |
| # Job should only be updated if there was an actual change |
| self.assertEqual(queue.GetNextUpdate(), (job, True)) |
| self.assertRaises(IndexError, queue.GetNextUpdate) |
| self.assertFalse(depmgr.CountPendingResults()) |
| |
| if attempt > 0 and attempt < 4: |
| # Simulate waiting for other job |
| self.assertEqual(result, jqueue._JobProcessor.WAITDEP) |
| self.assertTrue(job.cur_opctx) |
| 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 == jqueue._JobProcessor.FINISHED: |
| # Last opcode |
| self.assertFalse(job.cur_opctx) |
| break |
| |
| self.assertRaises(IndexError, depmgr.GetNextNotification) |
| |
| 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.assertEqual(job.CalcStatus(), constants.JOB_STATUS_CANCELED) |
| self.assertEqual(job.GetInfo(["status"]), [constants.JOB_STATUS_CANCELED]) |
| self.assertEqual(job.GetInfo(["opstatus", "opresult"]), |
| [[constants.OP_STATUS_SUCCESS, |
| constants.OP_STATUS_CANCELED, |
| constants.OP_STATUS_CANCELED], |
| ["Res0", "Job canceled by request", |
| "Job canceled by request"]]) |
| |
| self._GenericCheckJob(job) |
| |
| self.assertRaises(IndexError, queue.GetNextUpdate) |
| self.assertRaises(IndexError, depmgr.GetNextNotification) |
| self.assertFalse(depmgr.CountPendingResults()) |
| |
| # Calling the processor on a finished job should be a no-op |
| self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(), |
| jqueue._JobProcessor.FINISHED) |
| self.assertRaises(IndexError, queue.GetNextUpdate) |
| |
| def testJobDependencyWrongstatus(self): |
| depmgr = _FakeDependencyManager() |
| queue = _FakeQueueForProc(depmgr=depmgr) |
| |
| self.assertEqual(queue.depmgr, depmgr) |
| |
| prev_job_id = 9741 |
| job_id = 11763 |
| ops = [ |
| opcodes.OpTestDummy(result="Res0", fail=False), |
| opcodes.OpTestDummy(result="Res1", fail=False, |
| depends=[ |
| [prev_job_id, None], |
| ]), |
| opcodes.OpTestDummy(result="Res2", fail=False), |
| ] |
| |
| # Create job |
| job = self._CreateJob(queue, job_id, ops) |
| |
| def _BeforeStart(timeout, priority): |
| if attempt == 0 or attempt > 5: |
| # Job should only be updated when it wasn't waiting for another job |
| 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) |
| |
| # Job is running, cancelling shouldn't be possible |
| (success, _) = job.Cancel() |
| self.assertFalse(success) |
| |
| opexec = _FakeExecOpCodeForProc(queue, _BeforeStart, _AfterStart) |
| |
| self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) |
| |
| counter = itertools.count() |
| while True: |
| attempt = counter.next() |
| |
| self.assertRaises(IndexError, queue.GetNextUpdate) |
| self.assertRaises(IndexError, depmgr.GetNextNotification) |
| |
| if attempt == 0: |
| # This will handle the first opcode |
| pass |
| elif attempt < 4: |
| depmgr.AddCheckResult(job, prev_job_id, None, |
| (jqueue._JobDependencyManager.WAIT, "wait")) |
| elif attempt == 4: |
| # Other job failed |
| depmgr.AddCheckResult(job, prev_job_id, None, |
| (jqueue._JobDependencyManager.WRONGSTATUS, "w")) |
| |
| if attempt == 0: |
| self.assertEqual(depmgr.CountPendingResults(), 0) |
| else: |
| self.assertEqual(depmgr.CountPendingResults(), 1) |
| |
| result = jqueue._JobProcessor(queue, opexec, job)() |
| if attempt <= 1 or attempt >= 4: |
| # Job should only be updated if there was an actual change |
| self.assertEqual(queue.GetNextUpdate(), (job, True)) |
| self.assertRaises(IndexError, queue.GetNextUpdate) |
| self.assertFalse(depmgr.CountPendingResults()) |
| |
| if attempt > 0 and attempt < 4: |
| # Simulate waiting for other job |
| self.assertEqual(result, jqueue._JobProcessor.WAITDEP) |
| self.assertTrue(job.cur_opctx) |
| 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 == jqueue._JobProcessor.FINISHED: |
| # Last opcode |
| self.assertFalse(job.cur_opctx) |
| break |
| |
| self.assertRaises(IndexError, depmgr.GetNextNotification) |
| |
| 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.assertEqual(job.CalcStatus(), constants.JOB_STATUS_ERROR) |
| self.assertEqual(job.GetInfo(["status"]), [constants.JOB_STATUS_ERROR]) |
| self.assertEqual(job.GetInfo(["opstatus"]), |
| [[constants.OP_STATUS_SUCCESS, |
| constants.OP_STATUS_ERROR, |
| constants.OP_STATUS_ERROR]]), |
| |
| (opresult, ) = job.GetInfo(["opresult"]) |
| self.assertEqual(len(opresult), len(ops)) |
| self.assertEqual(opresult[0], "Res0") |
| self.assertTrue(errors.GetEncodedError(opresult[1])) |
| self.assertTrue(errors.GetEncodedError(opresult[2])) |
| |
| self._GenericCheckJob(job) |
| |
| self.assertRaises(IndexError, queue.GetNextUpdate) |
| self.assertRaises(IndexError, depmgr.GetNextNotification) |
| self.assertFalse(depmgr.CountPendingResults()) |
| |
| # Calling the processor on a finished job should be a no-op |
| 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.attempts = 0 |
| self.last_timeout = None |
| |
| def NextAttempt(self): |
| self.attempts += 1 |
| if self.timeouts: |
| timeout = self.timeouts.pop(0) |
| else: |
| timeout = None |
| self.last_timeout = timeout |
| return timeout |
| |
| |
| class TestJobProcessorTimeouts(unittest.TestCase, _JobProcessorTestUtils): |
| def setUp(self): |
| self.queue = _FakeQueueForProc() |
| self.job = None |
| self.curop = None |
| self.opcounter = None |
| self.timeout_strategy = None |
| self.retries = 0 |
| self.prev_tsop = None |
| self.prev_prio = None |
| self.prev_status = None |
| self.lock_acq_prio = None |
| self.gave_lock = None |
| self.done_lock_before_blocking = False |
| |
| def _BeforeStart(self, timeout, priority): |
| job = self.job |
| |
| # If status has changed, job must've been written |
| if self.prev_status != self.job.ops[self.curop].status: |
| self.assertEqual(self.queue.GetNextUpdate(), (job, True)) |
| self.assertRaises(IndexError, self.queue.GetNextUpdate) |
| |
| self.assertFalse(self.queue.IsAcquired()) |
| self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITING) |
| |
| ts = self.timeout_strategy |
| |
| self.assert_(timeout is None or isinstance(timeout, (int, float))) |
| self.assertEqual(timeout, ts.last_timeout) |
| self.assertEqual(priority, job.ops[self.curop].priority) |
| |
| self.gave_lock = True |
| self.lock_acq_prio = priority |
| |
| if (self.curop == 3 and |
| job.ops[self.curop].priority == constants.OP_PRIO_HIGHEST + 3): |
| # Give locks before running into blocking acquire |
| assert self.retries == 7 |
| self.retries = 0 |
| self.done_lock_before_blocking = True |
| return |
| |
| if self.retries > 0: |
| self.assert_(timeout is not None) |
| self.retries -= 1 |
| self.gave_lock = False |
| raise mcpu.LockAcquireTimeout() |
| |
| if job.ops[self.curop].priority == constants.OP_PRIO_HIGHEST: |
| assert self.retries == 0, "Didn't exhaust all retries at highest priority" |
| assert not ts.timeouts |
| self.assert_(timeout is None) |
| |
| def _AfterStart(self, op, cbs): |
| job = self.job |
| |
| # Setting to "running" requires an update |
| self.assertEqual(self.queue.GetNextUpdate(), (job, True)) |
| self.assertRaises(IndexError, self.queue.GetNextUpdate) |
| |
| self.assertFalse(self.queue.IsAcquired()) |
| self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_RUNNING) |
| |
| # Job is running, cancelling shouldn't be possible |
| (success, _) = job.Cancel() |
| self.assertFalse(success) |
| |
| def _NextOpcode(self): |
| self.curop = self.opcounter.next() |
| self.prev_prio = self.job.ops[self.curop].priority |
| self.prev_status = self.job.ops[self.curop].status |
| |
| def _NewTimeoutStrategy(self): |
| job = self.job |
| |
| self.assertEqual(self.retries, 0) |
| |
| if self.prev_tsop == self.curop: |
| # Still on the same opcode, priority must've been increased |
| self.assertEqual(self.prev_prio, job.ops[self.curop].priority + 1) |
| |
| if self.curop == 1: |
| # Normal retry |
| timeouts = range(10, 31, 10) |
| self.retries = len(timeouts) - 1 |
| |
| elif self.curop == 2: |
| # Let this run into a blocking acquire |
| timeouts = range(11, 61, 12) |
| self.retries = len(timeouts) |
| |
| elif self.curop == 3: |
| # Wait for priority to increase, but give lock before blocking acquire |
| timeouts = range(12, 100, 14) |
| self.retries = len(timeouts) |
| |
| self.assertFalse(self.done_lock_before_blocking) |
| |
| elif self.curop == 4: |
| self.assert_(self.done_lock_before_blocking) |
| |
| # Timeouts, but no need to retry |
| timeouts = range(10, 31, 10) |
| self.retries = 0 |
| |
| elif self.curop == 5: |
| # Normal retry |
| timeouts = range(19, 100, 11) |
| self.retries = len(timeouts) |
| |
| else: |
| timeouts = [] |
| self.retries = 0 |
| |
| assert len(job.ops) == 10 |
| assert self.retries <= len(timeouts) |
| |
| ts = _FakeTimeoutStrategy(timeouts) |
| |
| self.timeout_strategy = ts |
| self.prev_tsop = self.curop |
| self.prev_prio = job.ops[self.curop].priority |
| |
| return ts |
| |
| def testTimeout(self): |
| ops = [opcodes.OpTestDummy(result="Res%s" % i, fail=False) |
| for i in range(10)] |
| |
| # Create job |
| job_id = 15801 |
| job = self._CreateJob(self.queue, job_id, ops) |
| self.job = job |
| |
| self.opcounter = itertools.count(0) |
| |
| opexec = _FakeExecOpCodeForProc(self.queue, self._BeforeStart, |
| self._AfterStart) |
| tsf = self._NewTimeoutStrategy |
| |
| self.assertFalse(self.done_lock_before_blocking) |
| |
| while True: |
| proc = jqueue._JobProcessor(self.queue, opexec, job, |
| _timeout_strategy_factory=tsf) |
| |
| self.assertRaises(IndexError, self.queue.GetNextUpdate) |
| |
| if self.curop is not None: |
| self.prev_status = self.job.ops[self.curop].status |
| |
| self.lock_acq_prio = None |
| |
| result = proc(_nextop_fn=self._NextOpcode) |
| assert self.curop is not None |
| |
| # Input priority should never be set or modified |
| self.assertFalse(compat.any(hasattr(op.input, "priority") |
| for op in job.ops)) |
| |
| 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.assertRaises(IndexError, self.queue.GetNextUpdate) |
| self.assertEqual(self.lock_acq_prio, job.ops[self.curop].priority) |
| self.assert_(job.ops[self.curop].exec_timestamp) |
| |
| if result == jqueue._JobProcessor.FINISHED: |
| self.assertFalse(job.cur_opctx) |
| break |
| |
| self.assertEqual(result, jqueue._JobProcessor.DEFER) |
| |
| if self.curop == 0: |
| self.assertEqual(job.ops[self.curop].start_timestamp, |
| job.start_timestamp) |
| |
| if self.gave_lock: |
| # Opcode finished, but job not yet done |
| self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) |
| else: |
| # Did not get locks |
| self.assert_(job.cur_opctx) |
| 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_WAITING) |
| |
| # If priority has changed since acquiring locks, the job must've been |
| # updated |
| if self.lock_acq_prio != job.ops[self.curop].priority: |
| self.assertEqual(self.queue.GetNextUpdate(), (job, True)) |
| |
| self.assertRaises(IndexError, self.queue.GetNextUpdate) |
| |
| self.assert_(job.start_timestamp) |
| self.assertFalse(job.end_timestamp) |
| |
| self.assertEqual(self.curop, len(job.ops) - 1) |
| self.assertEqual(self.job, job) |
| self.assertEqual(self.opcounter.next(), len(job.ops)) |
| self.assert_(self.done_lock_before_blocking) |
| |
| self.assertRaises(IndexError, self.queue.GetNextUpdate) |
| self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_SUCCESS) |
| self.assertEqual(job.GetInfo(["status"]), [constants.JOB_STATUS_SUCCESS]) |
| self.assertEqual(job.GetInfo(["opresult"]), |
| [[op.input.result for op in job.ops]]) |
| self.assertEqual(job.GetInfo(["opstatus"]), |
| [len(job.ops) * [constants.OP_STATUS_SUCCESS]]) |
| self.assert_(compat.all(op.start_timestamp and op.end_timestamp |
| for op in job.ops)) |
| |
| # Calling the processor on a finished job should be a no-op |
| self.assertEqual(jqueue._JobProcessor(self.queue, opexec, job)(), |
| jqueue._JobProcessor.FINISHED) |
| self.assertRaises(IndexError, self.queue.GetNextUpdate) |
| |
| |
| class TestJobProcessorChangePriority(unittest.TestCase, _JobProcessorTestUtils): |
| def setUp(self): |
| self.queue = _FakeQueueForProc() |
| self.opexecprio = [] |
| |
| def _BeforeStart(self, timeout, priority): |
| self.assertFalse(self.queue.IsAcquired()) |
| self.opexecprio.append(priority) |
| |
| def testChangePriorityWhileRunning(self): |
| # Tests changing the priority on a job while it has finished opcodes |
| # (successful) and more, unprocessed ones |
| ops = [opcodes.OpTestDummy(result="Res%s" % i, fail=False) |
| for i in range(3)] |
| |
| # Create job |
| job_id = 3499 |
| job = self._CreateJob(self.queue, job_id, ops) |
| |
| self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) |
| |
| opexec = _FakeExecOpCodeForProc(self.queue, self._BeforeStart, None) |
| |
| # Run first opcode |
| self.assertEqual(jqueue._JobProcessor(self.queue, opexec, job)(), |
| jqueue._JobProcessor.DEFER) |
| |
| # Job goes back to queued |
| self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) |
| self.assertEqual(job.CalcPriority(), constants.OP_PRIO_DEFAULT) |
| self.assertEqual(job.GetInfo(["opstatus", "opresult"]), |
| [[constants.OP_STATUS_SUCCESS, |
| constants.OP_STATUS_QUEUED, |
| constants.OP_STATUS_QUEUED], |
| ["Res0", None, None]]) |
| |
| self.assertEqual(self.opexecprio.pop(0), constants.OP_PRIO_DEFAULT) |
| self.assertRaises(IndexError, self.opexecprio.pop, 0) |
| |
| # Change priority |
| self.assertEqual(job.ChangePriority(-10), |
| (True, |
| ("Priorities of pending opcodes for job 3499 have" |
| " been changed to -10"))) |
| self.assertEqual(job.CalcPriority(), -10) |
| |
| # Process second opcode |
| self.assertEqual(jqueue._JobProcessor(self.queue, opexec, job)(), |
| jqueue._JobProcessor.DEFER) |
| |
| self.assertEqual(self.opexecprio.pop(0), -10) |
| self.assertRaises(IndexError, self.opexecprio.pop, 0) |
| |
| # Check status |
| self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED) |
| self.assertEqual(job.CalcPriority(), -10) |
| self.assertEqual(job.GetInfo(["id"]), [job_id]) |
| self.assertEqual(job.GetInfo(["status"]), [constants.JOB_STATUS_QUEUED]) |
| self.assertEqual(job.GetInfo(["opstatus", "opresult"]), |
| [[constants.OP_STATUS_SUCCESS, |
| constants.OP_STATUS_SUCCESS, |
| constants.OP_STATUS_QUEUED], |
| ["Res0", "Res1", None]]) |
| |
| # Change priority once more |
| self.assertEqual(job.ChangePriority(5), |
| (True, |
| ("Priorities of pending opcodes for job 3499 have" |
| " been changed to 5"))) |
| self.assertEqual(job.CalcPriority(), 5) |
| |
| # Process third opcode |
| self.assertEqual(jqueue._JobProcessor(self.queue, opexec, job)(), |
| jqueue._JobProcessor.FINISHED) |
| |
| self.assertEqual(self.opexecprio.pop(0), 5) |
| self.assertRaises(IndexError, self.opexecprio.pop, 0) |
| |
| # Check status |
| self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_SUCCESS) |
| self.assertEqual(job.CalcPriority(), constants.OP_PRIO_DEFAULT) |
| self.assertEqual(job.GetInfo(["id"]), [job_id]) |
| self.assertEqual(job.GetInfo(["status"]), [constants.JOB_STATUS_SUCCESS]) |
| self.assertEqual(job.GetInfo(["opstatus", "opresult"]), |
| [[constants.OP_STATUS_SUCCESS, |
| constants.OP_STATUS_SUCCESS, |
| constants.OP_STATUS_SUCCESS], |
| ["Res0", "Res1", "Res2"]]) |
| self.assertEqual(map(operator.attrgetter("priority"), job.ops), |
| [constants.OP_PRIO_DEFAULT, -10, 5]) |
| |
| |
| 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 = [] |
| self.jdm = jqueue._JobDependencyManager(self._GetStatus, self._Enqueue) |
| |
| def _GetStatus(self, job_id): |
| (exp_job_id, result) = self._status.pop(0) |
| self.assertEqual(exp_job_id, job_id) |
| 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 = _IdOnlyFakeJob(17697) |
| job_id = str(28625) |
| |
| self._status.append((job_id, constants.JOB_STATUS_RUNNING)) |
| (result, _) = self.jdm.CheckAndRegister(job, job_id, []) |
| 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, { |
| 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.assertEqual(result, self.jdm.CANCEL) |
| self.assertFalse(self._status) |
| self.assertFalse(self._queue) |
| self.assertFalse(self.jdm.JobWaiting(job)) |
| self.assertFalse(self.jdm.GetLockInfo([query.LQ_PENDING])) |
| |
| def testNotFinalizedThenQueued(self): |
| # This can happen on a queue shutdown |
| job = _IdOnlyFakeJob(1320) |
| job_id = str(22971) |
| |
| for i in range(5): |
| if i > 2: |
| self._status.append((job_id, constants.JOB_STATUS_QUEUED)) |
| else: |
| self._status.append((job_id, constants.JOB_STATUS_RUNNING)) |
| (result, _) = self.jdm.CheckAndRegister(job, job_id, []) |
| 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, { |
| job_id: set([job]), |
| }) |
| self.assertEqual(self.jdm.GetLockInfo([query.LQ_PENDING]), [ |
| ("job/22971", None, None, [("job", [job.id])]) |
| ]) |
| |
| def testRequireCancel(self): |
| job = _IdOnlyFakeJob(5278) |
| job_id = str(9610) |
| dep_status = [constants.JOB_STATUS_CANCELED] |
| |
| 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._queue) |
| self.assertTrue(self.jdm.JobWaiting(job)) |
| 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.assertEqual(result, self.jdm.CONTINUE) |
| 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 = _IdOnlyFakeJob(21459) |
| job_id = str(25519) |
| dep_status = [constants.JOB_STATUS_ERROR] |
| |
| 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._queue) |
| self.assertTrue(self.jdm.JobWaiting(job)) |
| self.assertEqual(self.jdm._waiters, { |
| job_id: set([job]), |
| }) |
| |
| self._status.append((job_id, constants.JOB_STATUS_ERROR)) |
| (result, _) = self.jdm.CheckAndRegister(job, job_id, dep_status) |
| self.assertEqual(result, self.jdm.CONTINUE) |
| 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 = _IdOnlyFakeJob(21343) |
| job_id = str(14609) |
| |
| 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._queue) |
| self.assertTrue(self.jdm.JobWaiting(job)) |
| 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.assertEqual(result, self.jdm.CONTINUE) |
| 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 = _IdOnlyFakeJob(8227) |
| job_id = str(4113) |
| |
| self._status.append((job_id, constants.JOB_STATUS_RUNNING)) |
| (result, _) = self.jdm.CheckAndRegister(job, job_id, []) |
| 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, { |
| job_id: set([job]), |
| }) |
| |
| self.jdm.NotifyWaiters(job_id) |
| self.assertFalse(self._status) |
| self.assertFalse(self.jdm._waiters) |
| self.assertFalse(self.jdm.JobWaiting(job)) |
| self.assertEqual(self._queue, [set([job])]) |
| |
| def testWrongStatus(self): |
| job = _IdOnlyFakeJob(10102) |
| job_id = str(1271) |
| |
| 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, { |
| job_id: set([job]), |
| }) |
| |
| self._status.append((job_id, constants.JOB_STATUS_ERROR)) |
| (result, _) = self.jdm.CheckAndRegister(job, job_id, |
| [constants.JOB_STATUS_SUCCESS]) |
| self.assertEqual(result, self.jdm.WRONGSTATUS) |
| self.assertFalse(self._status) |
| self.assertFalse(self._queue) |
| self.assertFalse(self.jdm.JobWaiting(job)) |
| |
| def testCorrectStatus(self): |
| job = _IdOnlyFakeJob(24273) |
| job_id = str(23885) |
| |
| 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, { |
| job_id: set([job]), |
| }) |
| |
| 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)) |
| |
| def testFinalizedRightAway(self): |
| job = _IdOnlyFakeJob(224) |
| job_id = str(3081) |
| |
| 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)) |
| self.assertEqual(self.jdm._waiters, { |
| job_id: set(), |
| }) |
| |
| # Force cleanup |
| self.jdm.NotifyWaiters("0") |
| self.assertFalse(self.jdm._waiters) |
| 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 = _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 = _IdOnlyFakeJob(30540) |
| job_id = str(23769) |
| |
| def _FakeStatus(_): |
| raise errors.JobLost("#msg#") |
| |
| jdm = jqueue._JobDependencyManager(_FakeStatus, None) |
| (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__": |
| testutils.GanetiTestProgram() |
