Statistics
| Branch: | Tag: | Revision:

root / test / ganeti.jqueue_unittest.py @ df5a5730

History | View | Annotate | Download (75.8 kB)

1
#!/usr/bin/python
2
#
3

    
4
# Copyright (C) 2010, 2011 Google Inc.
5
#
6
# This program is free software; you can redistribute it and/or modify
7
# it under the terms of the GNU General Public License as published by
8
# the Free Software Foundation; either version 2 of the License, or
9
# (at your option) any later version.
10
#
11
# This program is distributed in the hope that it will be useful, but
12
# WITHOUT ANY WARRANTY; without even the implied warranty of
13
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
14
# General Public License for more details.
15
#
16
# You should have received a copy of the GNU General Public License
17
# along with this program; if not, write to the Free Software
18
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
19
# 02110-1301, USA.
20

    
21

    
22
"""Script for testing ganeti.jqueue"""
23

    
24
import os
25
import sys
26
import unittest
27
import tempfile
28
import shutil
29
import errno
30
import itertools
31
import random
32

    
33
from ganeti import constants
34
from ganeti import utils
35
from ganeti import errors
36
from ganeti import jqueue
37
from ganeti import opcodes
38
from ganeti import compat
39
from ganeti import mcpu
40
from ganeti import query
41
from ganeti import workerpool
42

    
43
import testutils
44

    
45

    
46
class _FakeJob:
47
  def __init__(self, job_id, status):
48
    self.id = job_id
49
    self.writable = False
50
    self._status = status
51
    self._log = []
52

    
53
  def SetStatus(self, status):
54
    self._status = status
55

    
56
  def AddLogEntry(self, msg):
57
    self._log.append((len(self._log), msg))
58

    
59
  def CalcStatus(self):
60
    return self._status
61

    
62
  def GetInfo(self, fields):
63
    result = []
64

    
65
    for name in fields:
66
      if name == "status":
67
        result.append(self._status)
68
      else:
69
        raise Exception("Unknown field")
70

    
71
    return result
72

    
73
  def GetLogEntries(self, newer_than):
74
    assert newer_than is None or newer_than >= 0
75

    
76
    if newer_than is None:
77
      return self._log
78

    
79
    return self._log[newer_than:]
80

    
81

    
82
class TestJobChangesChecker(unittest.TestCase):
83
  def testStatus(self):
84
    job = _FakeJob(9094, constants.JOB_STATUS_QUEUED)
85
    checker = jqueue._JobChangesChecker(["status"], None, None)
86
    self.assertEqual(checker(job), ([constants.JOB_STATUS_QUEUED], []))
87

    
88
    job.SetStatus(constants.JOB_STATUS_RUNNING)
89
    self.assertEqual(checker(job), ([constants.JOB_STATUS_RUNNING], []))
90

    
91
    job.SetStatus(constants.JOB_STATUS_SUCCESS)
92
    self.assertEqual(checker(job), ([constants.JOB_STATUS_SUCCESS], []))
93

    
94
    # job.id is used by checker
95
    self.assertEqual(job.id, 9094)
96

    
97
  def testStatusWithPrev(self):
98
    job = _FakeJob(12807, constants.JOB_STATUS_QUEUED)
99
    checker = jqueue._JobChangesChecker(["status"],
100
                                        [constants.JOB_STATUS_QUEUED], None)
101
    self.assert_(checker(job) is None)
102

    
103
    job.SetStatus(constants.JOB_STATUS_RUNNING)
104
    self.assertEqual(checker(job), ([constants.JOB_STATUS_RUNNING], []))
105

    
106
  def testFinalStatus(self):
107
    for status in constants.JOBS_FINALIZED:
108
      job = _FakeJob(2178711, status)
109
      checker = jqueue._JobChangesChecker(["status"], [status], None)
110
      # There won't be any changes in this status, hence it should signal
111
      # a change immediately
112
      self.assertEqual(checker(job), ([status], []))
113

    
114
  def testLog(self):
115
    job = _FakeJob(9094, constants.JOB_STATUS_RUNNING)
116
    checker = jqueue._JobChangesChecker(["status"], None, None)
117
    self.assertEqual(checker(job), ([constants.JOB_STATUS_RUNNING], []))
118

    
119
    job.AddLogEntry("Hello World")
120
    (job_info, log_entries) = checker(job)
121
    self.assertEqual(job_info, [constants.JOB_STATUS_RUNNING])
122
    self.assertEqual(log_entries, [[0, "Hello World"]])
123

    
124
    checker2 = jqueue._JobChangesChecker(["status"], job_info, len(log_entries))
125
    self.assert_(checker2(job) is None)
126

    
127
    job.AddLogEntry("Foo Bar")
128
    job.SetStatus(constants.JOB_STATUS_ERROR)
129

    
130
    (job_info, log_entries) = checker2(job)
131
    self.assertEqual(job_info, [constants.JOB_STATUS_ERROR])
132
    self.assertEqual(log_entries, [[1, "Foo Bar"]])
133

    
134
    checker3 = jqueue._JobChangesChecker(["status"], None, None)
135
    (job_info, log_entries) = checker3(job)
136
    self.assertEqual(job_info, [constants.JOB_STATUS_ERROR])
137
    self.assertEqual(log_entries, [[0, "Hello World"], [1, "Foo Bar"]])
138

    
139

    
140
class TestJobChangesWaiter(unittest.TestCase):
141
  def setUp(self):
142
    self.tmpdir = tempfile.mkdtemp()
143
    self.filename = utils.PathJoin(self.tmpdir, "job-1")
144
    utils.WriteFile(self.filename, data="")
145

    
146
  def tearDown(self):
147
    shutil.rmtree(self.tmpdir)
148

    
149
  def _EnsureNotifierClosed(self, notifier):
150
    try:
151
      os.fstat(notifier._fd)
152
    except EnvironmentError, err:
153
      self.assertEqual(err.errno, errno.EBADF)
154
    else:
155
      self.fail("File descriptor wasn't closed")
156

    
157
  def testClose(self):
158
    for wait in [False, True]:
159
      waiter = jqueue._JobFileChangesWaiter(self.filename)
160
      try:
161
        if wait:
162
          waiter.Wait(0.001)
163
      finally:
164
        waiter.Close()
165

    
166
      # Ensure file descriptor was closed
167
      self._EnsureNotifierClosed(waiter._notifier)
168

    
169
  def testChangingFile(self):
170
    waiter = jqueue._JobFileChangesWaiter(self.filename)
171
    try:
172
      self.assertFalse(waiter.Wait(0.1))
173
      utils.WriteFile(self.filename, data="changed")
174
      self.assert_(waiter.Wait(60))
175
    finally:
176
      waiter.Close()
177

    
178
    self._EnsureNotifierClosed(waiter._notifier)
179

    
180
  def testChangingFile2(self):
181
    waiter = jqueue._JobChangesWaiter(self.filename)
182
    try:
183
      self.assertFalse(waiter._filewaiter)
184
      self.assert_(waiter.Wait(0.1))
185
      self.assert_(waiter._filewaiter)
186

    
187
      # File waiter is now used, but there have been no changes
188
      self.assertFalse(waiter.Wait(0.1))
189
      utils.WriteFile(self.filename, data="changed")
190
      self.assert_(waiter.Wait(60))
191
    finally:
192
      waiter.Close()
193

    
194
    self._EnsureNotifierClosed(waiter._filewaiter._notifier)
195

    
196

    
197
class TestWaitForJobChangesHelper(unittest.TestCase):
198
  def setUp(self):
199
    self.tmpdir = tempfile.mkdtemp()
200
    self.filename = utils.PathJoin(self.tmpdir, "job-2614226563")
201
    utils.WriteFile(self.filename, data="")
202

    
203
  def tearDown(self):
204
    shutil.rmtree(self.tmpdir)
205

    
206
  def _LoadWaitingJob(self):
207
    return _FakeJob(2614226563, constants.JOB_STATUS_WAITING)
208

    
209
  def _LoadLostJob(self):
210
    return None
211

    
212
  def testNoChanges(self):
213
    wfjc = jqueue._WaitForJobChangesHelper()
214

    
215
    # No change
216
    self.assertEqual(wfjc(self.filename, self._LoadWaitingJob, ["status"],
217
                          [constants.JOB_STATUS_WAITING], None, 0.1),
218
                     constants.JOB_NOTCHANGED)
219

    
220
    # No previous information
221
    self.assertEqual(wfjc(self.filename, self._LoadWaitingJob,
222
                          ["status"], None, None, 1.0),
223
                     ([constants.JOB_STATUS_WAITING], []))
224

    
225
  def testLostJob(self):
226
    wfjc = jqueue._WaitForJobChangesHelper()
227
    self.assert_(wfjc(self.filename, self._LoadLostJob,
228
                      ["status"], None, None, 1.0) is None)
229

    
230

    
231
class TestEncodeOpError(unittest.TestCase):
232
  def test(self):
233
    encerr = jqueue._EncodeOpError(errors.LockError("Test 1"))
234
    self.assert_(isinstance(encerr, tuple))
235
    self.assertRaises(errors.LockError, errors.MaybeRaise, encerr)
236

    
237
    encerr = jqueue._EncodeOpError(errors.GenericError("Test 2"))
238
    self.assert_(isinstance(encerr, tuple))
239
    self.assertRaises(errors.GenericError, errors.MaybeRaise, encerr)
240

    
241
    encerr = jqueue._EncodeOpError(NotImplementedError("Foo"))
242
    self.assert_(isinstance(encerr, tuple))
243
    self.assertRaises(errors.OpExecError, errors.MaybeRaise, encerr)
244

    
245
    encerr = jqueue._EncodeOpError("Hello World")
246
    self.assert_(isinstance(encerr, tuple))
247
    self.assertRaises(errors.OpExecError, errors.MaybeRaise, encerr)
248

    
249

    
250
class TestQueuedOpCode(unittest.TestCase):
251
  def testDefaults(self):
252
    def _Check(op):
253
      self.assertFalse(hasattr(op.input, "dry_run"))
254
      self.assertEqual(op.priority, constants.OP_PRIO_DEFAULT)
255
      self.assertFalse(op.log)
256
      self.assert_(op.start_timestamp is None)
257
      self.assert_(op.exec_timestamp is None)
258
      self.assert_(op.end_timestamp is None)
259
      self.assert_(op.result is None)
260
      self.assertEqual(op.status, constants.OP_STATUS_QUEUED)
261

    
262
    op1 = jqueue._QueuedOpCode(opcodes.OpTestDelay())
263
    _Check(op1)
264
    op2 = jqueue._QueuedOpCode.Restore(op1.Serialize())
265
    _Check(op2)
266
    self.assertEqual(op1.Serialize(), op2.Serialize())
267

    
268
  def testPriority(self):
269
    def _Check(op):
270
      assert constants.OP_PRIO_DEFAULT != constants.OP_PRIO_HIGH, \
271
             "Default priority equals high priority; test can't work"
272
      self.assertEqual(op.priority, constants.OP_PRIO_HIGH)
273
      self.assertEqual(op.status, constants.OP_STATUS_QUEUED)
274

    
275
    inpop = opcodes.OpTagsGet(priority=constants.OP_PRIO_HIGH)
276
    op1 = jqueue._QueuedOpCode(inpop)
277
    _Check(op1)
278
    op2 = jqueue._QueuedOpCode.Restore(op1.Serialize())
279
    _Check(op2)
280
    self.assertEqual(op1.Serialize(), op2.Serialize())
281

    
282

    
283
class TestQueuedJob(unittest.TestCase):
284
  def test(self):
285
    self.assertRaises(errors.GenericError, jqueue._QueuedJob,
286
                      None, 1, [], False)
287

    
288
  def testDefaults(self):
289
    job_id = 4260
290
    ops = [
291
      opcodes.OpTagsGet(),
292
      opcodes.OpTestDelay(),
293
      ]
294

    
295
    def _Check(job):
296
      self.assertTrue(job.writable)
297
      self.assertEqual(job.id, job_id)
298
      self.assertEqual(job.log_serial, 0)
299
      self.assert_(job.received_timestamp)
300
      self.assert_(job.start_timestamp is None)
301
      self.assert_(job.end_timestamp is None)
302
      self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED)
303
      self.assertEqual(job.CalcPriority(), constants.OP_PRIO_DEFAULT)
304
      self.assert_(repr(job).startswith("<"))
305
      self.assertEqual(len(job.ops), len(ops))
306
      self.assert_(compat.all(inp.__getstate__() == op.input.__getstate__()
307
                              for (inp, op) in zip(ops, job.ops)))
308
      self.assertRaises(errors.OpExecError, job.GetInfo,
309
                        ["unknown-field"])
310
      self.assertEqual(job.GetInfo(["summary"]),
311
                       [[op.input.Summary() for op in job.ops]])
312

    
313
    job1 = jqueue._QueuedJob(None, job_id, ops, True)
314
    _Check(job1)
315
    job2 = jqueue._QueuedJob.Restore(None, job1.Serialize(), True)
316
    _Check(job2)
317
    self.assertEqual(job1.Serialize(), job2.Serialize())
318

    
319
  def testWritable(self):
320
    job = jqueue._QueuedJob(None, 1, [opcodes.OpTestDelay()], False)
321
    self.assertFalse(job.writable)
322

    
323
    job = jqueue._QueuedJob(None, 1, [opcodes.OpTestDelay()], True)
324
    self.assertTrue(job.writable)
325

    
326
  def testPriority(self):
327
    job_id = 4283
328
    ops = [
329
      opcodes.OpTagsGet(priority=constants.OP_PRIO_DEFAULT),
330
      opcodes.OpTestDelay(),
331
      ]
332

    
333
    def _Check(job):
334
      self.assertEqual(job.id, job_id)
335
      self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED)
336
      self.assert_(repr(job).startswith("<"))
337

    
338
    job = jqueue._QueuedJob(None, job_id, ops, True)
339
    _Check(job)
340
    self.assert_(compat.all(op.priority == constants.OP_PRIO_DEFAULT
341
                            for op in job.ops))
342
    self.assertEqual(job.CalcPriority(), constants.OP_PRIO_DEFAULT)
343

    
344
    # Increase first
345
    job.ops[0].priority -= 1
346
    _Check(job)
347
    self.assertEqual(job.CalcPriority(), constants.OP_PRIO_DEFAULT - 1)
348

    
349
    # Mark opcode as finished
350
    job.ops[0].status = constants.OP_STATUS_SUCCESS
351
    _Check(job)
352
    self.assertEqual(job.CalcPriority(), constants.OP_PRIO_DEFAULT)
353

    
354
    # Increase second
355
    job.ops[1].priority -= 10
356
    self.assertEqual(job.CalcPriority(), constants.OP_PRIO_DEFAULT - 10)
357

    
358
    # Test increasing first
359
    job.ops[0].status = constants.OP_STATUS_RUNNING
360
    job.ops[0].priority -= 19
361
    self.assertEqual(job.CalcPriority(), constants.OP_PRIO_DEFAULT - 20)
362

    
363
  def testCalcStatus(self):
364
    def _Queued(ops):
365
      # The default status is "queued"
366
      self.assert_(compat.all(op.status == constants.OP_STATUS_QUEUED
367
                              for op in ops))
368

    
369
    def _Waitlock1(ops):
370
      ops[0].status = constants.OP_STATUS_WAITING
371

    
372
    def _Waitlock2(ops):
373
      ops[0].status = constants.OP_STATUS_SUCCESS
374
      ops[1].status = constants.OP_STATUS_SUCCESS
375
      ops[2].status = constants.OP_STATUS_WAITING
376

    
377
    def _Running(ops):
378
      ops[0].status = constants.OP_STATUS_SUCCESS
379
      ops[1].status = constants.OP_STATUS_RUNNING
380
      for op in ops[2:]:
381
        op.status = constants.OP_STATUS_QUEUED
382

    
383
    def _Canceling1(ops):
384
      ops[0].status = constants.OP_STATUS_SUCCESS
385
      ops[1].status = constants.OP_STATUS_SUCCESS
386
      for op in ops[2:]:
387
        op.status = constants.OP_STATUS_CANCELING
388

    
389
    def _Canceling2(ops):
390
      for op in ops:
391
        op.status = constants.OP_STATUS_CANCELING
392

    
393
    def _Canceled(ops):
394
      for op in ops:
395
        op.status = constants.OP_STATUS_CANCELED
396

    
397
    def _Error1(ops):
398
      for idx, op in enumerate(ops):
399
        if idx > 3:
400
          op.status = constants.OP_STATUS_ERROR
401
        else:
402
          op.status = constants.OP_STATUS_SUCCESS
403

    
404
    def _Error2(ops):
405
      for op in ops:
406
        op.status = constants.OP_STATUS_ERROR
407

    
408
    def _Success(ops):
409
      for op in ops:
410
        op.status = constants.OP_STATUS_SUCCESS
411

    
412
    tests = {
413
      constants.JOB_STATUS_QUEUED: [_Queued],
414
      constants.JOB_STATUS_WAITING: [_Waitlock1, _Waitlock2],
415
      constants.JOB_STATUS_RUNNING: [_Running],
416
      constants.JOB_STATUS_CANCELING: [_Canceling1, _Canceling2],
417
      constants.JOB_STATUS_CANCELED: [_Canceled],
418
      constants.JOB_STATUS_ERROR: [_Error1, _Error2],
419
      constants.JOB_STATUS_SUCCESS: [_Success],
420
      }
421

    
422
    def _NewJob():
423
      job = jqueue._QueuedJob(None, 1,
424
                              [opcodes.OpTestDelay() for _ in range(10)],
425
                              True)
426
      self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED)
427
      self.assert_(compat.all(op.status == constants.OP_STATUS_QUEUED
428
                              for op in job.ops))
429
      return job
430

    
431
    for status in constants.JOB_STATUS_ALL:
432
      sttests = tests[status]
433
      assert sttests
434
      for fn in sttests:
435
        job = _NewJob()
436
        fn(job.ops)
437
        self.assertEqual(job.CalcStatus(), status)
438

    
439

    
440
class _FakeDependencyManager:
441
  def __init__(self):
442
    self._checks = []
443
    self._notifications = []
444
    self._waiting = set()
445

    
446
  def AddCheckResult(self, job, dep_job_id, dep_status, result):
447
    self._checks.append((job, dep_job_id, dep_status, result))
448

    
449
  def CountPendingResults(self):
450
    return len(self._checks)
451

    
452
  def CountWaitingJobs(self):
453
    return len(self._waiting)
454

    
455
  def GetNextNotification(self):
456
    return self._notifications.pop(0)
457

    
458
  def JobWaiting(self, job):
459
    return job in self._waiting
460

    
461
  def CheckAndRegister(self, job, dep_job_id, dep_status):
462
    (exp_job, exp_dep_job_id, exp_dep_status, result) = self._checks.pop(0)
463

    
464
    assert exp_job == job
465
    assert exp_dep_job_id == dep_job_id
466
    assert exp_dep_status == dep_status
467

    
468
    (result_status, _) = result
469

    
470
    if result_status == jqueue._JobDependencyManager.WAIT:
471
      self._waiting.add(job)
472
    elif result_status == jqueue._JobDependencyManager.CONTINUE:
473
      self._waiting.remove(job)
474

    
475
    return result
476

    
477
  def NotifyWaiters(self, job_id):
478
    self._notifications.append(job_id)
479

    
480

    
481
class _DisabledFakeDependencyManager:
482
  def JobWaiting(self, _):
483
    return False
484

    
485
  def CheckAndRegister(self, *args):
486
    assert False, "Should not be called"
487

    
488
  def NotifyWaiters(self, _):
489
    pass
490

    
491

    
492
class _FakeQueueForProc:
493
  def __init__(self, depmgr=None):
494
    self._acquired = False
495
    self._updates = []
496
    self._submitted = []
497

    
498
    self._submit_count = itertools.count(1000)
499

    
500
    if depmgr:
501
      self.depmgr = depmgr
502
    else:
503
      self.depmgr = _DisabledFakeDependencyManager()
504

    
505
  def IsAcquired(self):
506
    return self._acquired
507

    
508
  def GetNextUpdate(self):
509
    return self._updates.pop(0)
510

    
511
  def GetNextSubmittedJob(self):
512
    return self._submitted.pop(0)
513

    
514
  def acquire(self, shared=0):
515
    assert shared == 1
516
    self._acquired = True
517

    
518
  def release(self):
519
    assert self._acquired
520
    self._acquired = False
521

    
522
  def UpdateJobUnlocked(self, job, replicate=True):
523
    assert self._acquired, "Lock not acquired while updating job"
524
    self._updates.append((job, bool(replicate)))
525

    
526
  def SubmitManyJobs(self, jobs):
527
    assert not self._acquired, "Lock acquired while submitting jobs"
528
    job_ids = [self._submit_count.next() for _ in jobs]
529
    self._submitted.extend(zip(job_ids, jobs))
530
    return job_ids
531

    
532

    
533
class _FakeExecOpCodeForProc:
534
  def __init__(self, queue, before_start, after_start):
535
    self._queue = queue
536
    self._before_start = before_start
537
    self._after_start = after_start
538

    
539
  def __call__(self, op, cbs, timeout=None, priority=None):
540
    assert isinstance(op, opcodes.OpTestDummy)
541
    assert not self._queue.IsAcquired(), \
542
           "Queue lock not released when executing opcode"
543

    
544
    if self._before_start:
545
      self._before_start(timeout, priority)
546

    
547
    cbs.NotifyStart()
548

    
549
    if self._after_start:
550
      self._after_start(op, cbs)
551

    
552
    # Check again after the callbacks
553
    assert not self._queue.IsAcquired()
554

    
555
    if op.fail:
556
      raise errors.OpExecError("Error requested (%s)" % op.result)
557

    
558
    if hasattr(op, "submit_jobs") and op.submit_jobs is not None:
559
      return cbs.SubmitManyJobs(op.submit_jobs)
560

    
561
    return op.result
562

    
563

    
564
class _JobProcessorTestUtils:
565
  def _CreateJob(self, queue, job_id, ops):
566
    job = jqueue._QueuedJob(queue, job_id, ops, True)
567
    self.assertFalse(job.start_timestamp)
568
    self.assertFalse(job.end_timestamp)
569
    self.assertEqual(len(ops), len(job.ops))
570
    self.assert_(compat.all(op.input == inp
571
                            for (op, inp) in zip(job.ops, ops)))
572
    self.assertEqual(job.GetInfo(["ops"]), [[op.__getstate__() for op in ops]])
573
    return job
574

    
575

    
576
class TestJobProcessor(unittest.TestCase, _JobProcessorTestUtils):
577
  def _GenericCheckJob(self, job):
578
    assert compat.all(isinstance(op.input, opcodes.OpTestDummy)
579
                      for op in job.ops)
580

    
581
    self.assertEqual(job.GetInfo(["opstart", "opexec", "opend"]),
582
                     [[op.start_timestamp for op in job.ops],
583
                      [op.exec_timestamp for op in job.ops],
584
                      [op.end_timestamp for op in job.ops]])
585
    self.assertEqual(job.GetInfo(["received_ts", "start_ts", "end_ts"]),
586
                     [job.received_timestamp,
587
                      job.start_timestamp,
588
                      job.end_timestamp])
589
    self.assert_(job.start_timestamp)
590
    self.assert_(job.end_timestamp)
591
    self.assertEqual(job.start_timestamp, job.ops[0].start_timestamp)
592

    
593
  def testSuccess(self):
594
    queue = _FakeQueueForProc()
595

    
596
    for (job_id, opcount) in [(25351, 1), (6637, 3),
597
                              (24644, 10), (32207, 100)]:
598
      ops = [opcodes.OpTestDummy(result="Res%s" % i, fail=False)
599
             for i in range(opcount)]
600

    
601
      # Create job
602
      job = self._CreateJob(queue, job_id, ops)
603

    
604
      def _BeforeStart(timeout, priority):
605
        self.assertEqual(queue.GetNextUpdate(), (job, True))
606
        self.assertRaises(IndexError, queue.GetNextUpdate)
607
        self.assertFalse(queue.IsAcquired())
608
        self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITING)
609
        self.assertFalse(job.cur_opctx)
610

    
611
      def _AfterStart(op, cbs):
612
        self.assertEqual(queue.GetNextUpdate(), (job, True))
613
        self.assertRaises(IndexError, queue.GetNextUpdate)
614

    
615
        self.assertFalse(queue.IsAcquired())
616
        self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_RUNNING)
617
        self.assertFalse(job.cur_opctx)
618

    
619
        # Job is running, cancelling shouldn't be possible
620
        (success, _) = job.Cancel()
621
        self.assertFalse(success)
622

    
623
      opexec = _FakeExecOpCodeForProc(queue, _BeforeStart, _AfterStart)
624

    
625
      for idx in range(len(ops)):
626
        self.assertRaises(IndexError, queue.GetNextUpdate)
627
        result = jqueue._JobProcessor(queue, opexec, job)()
628
        self.assertEqual(queue.GetNextUpdate(), (job, True))
629
        self.assertRaises(IndexError, queue.GetNextUpdate)
630
        if idx == len(ops) - 1:
631
          # Last opcode
632
          self.assertEqual(result, jqueue._JobProcessor.FINISHED)
633
        else:
634
          self.assertEqual(result, jqueue._JobProcessor.DEFER)
635

    
636
          self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED)
637
          self.assert_(job.start_timestamp)
638
          self.assertFalse(job.end_timestamp)
639

    
640
      self.assertRaises(IndexError, queue.GetNextUpdate)
641

    
642
      self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_SUCCESS)
643
      self.assertEqual(job.GetInfo(["status"]), [constants.JOB_STATUS_SUCCESS])
644
      self.assertEqual(job.GetInfo(["opresult"]),
645
                       [[op.input.result for op in job.ops]])
646
      self.assertEqual(job.GetInfo(["opstatus"]),
647
                       [len(job.ops) * [constants.OP_STATUS_SUCCESS]])
648
      self.assert_(compat.all(op.start_timestamp and op.end_timestamp
649
                              for op in job.ops))
650

    
651
      self._GenericCheckJob(job)
652

    
653
      # Calling the processor on a finished job should be a no-op
654
      self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(),
655
                       jqueue._JobProcessor.FINISHED)
656
      self.assertRaises(IndexError, queue.GetNextUpdate)
657

    
658
  def testOpcodeError(self):
659
    queue = _FakeQueueForProc()
660

    
661
    testdata = [
662
      (17077, 1, 0, 0),
663
      (1782, 5, 2, 2),
664
      (18179, 10, 9, 9),
665
      (4744, 10, 3, 8),
666
      (23816, 100, 39, 45),
667
      ]
668

    
669
    for (job_id, opcount, failfrom, failto) in testdata:
670
      # Prepare opcodes
671
      ops = [opcodes.OpTestDummy(result="Res%s" % i,
672
                                 fail=(failfrom <= i and
673
                                       i <= failto))
674
             for i in range(opcount)]
675

    
676
      # Create job
677
      job = self._CreateJob(queue, job_id, ops)
678

    
679
      opexec = _FakeExecOpCodeForProc(queue, None, None)
680

    
681
      for idx in range(len(ops)):
682
        self.assertRaises(IndexError, queue.GetNextUpdate)
683
        result = jqueue._JobProcessor(queue, opexec, job)()
684
        # queued to waitlock
685
        self.assertEqual(queue.GetNextUpdate(), (job, True))
686
        # waitlock to running
687
        self.assertEqual(queue.GetNextUpdate(), (job, True))
688
        # Opcode result
689
        self.assertEqual(queue.GetNextUpdate(), (job, True))
690
        self.assertRaises(IndexError, queue.GetNextUpdate)
691

    
692
        if idx in (failfrom, len(ops) - 1):
693
          # Last opcode
694
          self.assertEqual(result, jqueue._JobProcessor.FINISHED)
695
          break
696

    
697
        self.assertEqual(result, jqueue._JobProcessor.DEFER)
698

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

    
701
      self.assertRaises(IndexError, queue.GetNextUpdate)
702

    
703
      # Check job status
704
      self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_ERROR)
705
      self.assertEqual(job.GetInfo(["id"]), [job_id])
706
      self.assertEqual(job.GetInfo(["status"]), [constants.JOB_STATUS_ERROR])
707

    
708
      # Check opcode status
709
      data = zip(job.ops,
710
                 job.GetInfo(["opstatus"])[0],
711
                 job.GetInfo(["opresult"])[0])
712

    
713
      for idx, (op, opstatus, opresult) in enumerate(data):
714
        if idx < failfrom:
715
          assert not op.input.fail
716
          self.assertEqual(opstatus, constants.OP_STATUS_SUCCESS)
717
          self.assertEqual(opresult, op.input.result)
718
        elif idx <= failto:
719
          assert op.input.fail
720
          self.assertEqual(opstatus, constants.OP_STATUS_ERROR)
721
          self.assertRaises(errors.OpExecError, errors.MaybeRaise, opresult)
722
        else:
723
          assert not op.input.fail
724
          self.assertEqual(opstatus, constants.OP_STATUS_ERROR)
725
          self.assertRaises(errors.OpExecError, errors.MaybeRaise, opresult)
726

    
727
      self.assert_(compat.all(op.start_timestamp and op.end_timestamp
728
                              for op in job.ops[:failfrom]))
729

    
730
      self._GenericCheckJob(job)
731

    
732
      # Calling the processor on a finished job should be a no-op
733
      self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(),
734
                       jqueue._JobProcessor.FINISHED)
735
      self.assertRaises(IndexError, queue.GetNextUpdate)
736

    
737
  def testCancelWhileInQueue(self):
738
    queue = _FakeQueueForProc()
739

    
740
    ops = [opcodes.OpTestDummy(result="Res%s" % i, fail=False)
741
           for i in range(5)]
742

    
743
    # Create job
744
    job_id = 17045
745
    job = self._CreateJob(queue, job_id, ops)
746

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

    
749
    # Mark as cancelled
750
    (success, _) = job.Cancel()
751
    self.assert_(success)
752

    
753
    self.assertRaises(IndexError, queue.GetNextUpdate)
754

    
755
    self.assertFalse(job.start_timestamp)
756
    self.assertTrue(job.end_timestamp)
757
    self.assert_(compat.all(op.status == constants.OP_STATUS_CANCELED
758
                            for op in job.ops))
759

    
760
    # Serialize to check for differences
761
    before_proc = job.Serialize()
762

    
763
    # Simulate processor called in workerpool
764
    opexec = _FakeExecOpCodeForProc(queue, None, None)
765
    self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(),
766
                     jqueue._JobProcessor.FINISHED)
767

    
768
    # Check result
769
    self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_CANCELED)
770
    self.assertEqual(job.GetInfo(["status"]), [constants.JOB_STATUS_CANCELED])
771
    self.assertFalse(job.start_timestamp)
772
    self.assertTrue(job.end_timestamp)
773
    self.assertFalse(compat.any(op.start_timestamp or op.end_timestamp
774
                                for op in job.ops))
775
    self.assertEqual(job.GetInfo(["opstatus", "opresult"]),
776
                     [[constants.OP_STATUS_CANCELED for _ in job.ops],
777
                      ["Job canceled by request" for _ in job.ops]])
778

    
779
    # Must not have changed or written
780
    self.assertEqual(before_proc, job.Serialize())
781
    self.assertRaises(IndexError, queue.GetNextUpdate)
782

    
783
  def testCancelWhileWaitlockInQueue(self):
784
    queue = _FakeQueueForProc()
785

    
786
    ops = [opcodes.OpTestDummy(result="Res%s" % i, fail=False)
787
           for i in range(5)]
788

    
789
    # Create job
790
    job_id = 8645
791
    job = self._CreateJob(queue, job_id, ops)
792

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

    
795
    job.ops[0].status = constants.OP_STATUS_WAITING
796

    
797
    assert len(job.ops) == 5
798

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

    
801
    # Mark as cancelling
802
    (success, _) = job.Cancel()
803
    self.assert_(success)
804

    
805
    self.assertRaises(IndexError, queue.GetNextUpdate)
806

    
807
    self.assert_(compat.all(op.status == constants.OP_STATUS_CANCELING
808
                            for op in job.ops))
809

    
810
    opexec = _FakeExecOpCodeForProc(queue, None, None)
811
    self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(),
812
                     jqueue._JobProcessor.FINISHED)
813

    
814
    # Check result
815
    self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_CANCELED)
816
    self.assertEqual(job.GetInfo(["status"]), [constants.JOB_STATUS_CANCELED])
817
    self.assertFalse(job.start_timestamp)
818
    self.assert_(job.end_timestamp)
819
    self.assertFalse(compat.any(op.start_timestamp or op.end_timestamp
820
                                for op in job.ops))
821
    self.assertEqual(job.GetInfo(["opstatus", "opresult"]),
822
                     [[constants.OP_STATUS_CANCELED for _ in job.ops],
823
                      ["Job canceled by request" for _ in job.ops]])
824

    
825
  def testCancelWhileWaitlock(self):
826
    queue = _FakeQueueForProc()
827

    
828
    ops = [opcodes.OpTestDummy(result="Res%s" % i, fail=False)
829
           for i in range(5)]
830

    
831
    # Create job
832
    job_id = 11009
833
    job = self._CreateJob(queue, job_id, ops)
834

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

    
837
    def _BeforeStart(timeout, priority):
838
      self.assertEqual(queue.GetNextUpdate(), (job, True))
839
      self.assertRaises(IndexError, queue.GetNextUpdate)
840
      self.assertFalse(queue.IsAcquired())
841
      self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITING)
842

    
843
      # Mark as cancelled
844
      (success, _) = job.Cancel()
845
      self.assert_(success)
846

    
847
      self.assert_(compat.all(op.status == constants.OP_STATUS_CANCELING
848
                              for op in job.ops))
849
      self.assertRaises(IndexError, queue.GetNextUpdate)
850

    
851
    def _AfterStart(op, cbs):
852
      self.assertEqual(queue.GetNextUpdate(), (job, True))
853
      self.assertRaises(IndexError, queue.GetNextUpdate)
854
      self.assertFalse(queue.IsAcquired())
855
      self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_RUNNING)
856

    
857
    opexec = _FakeExecOpCodeForProc(queue, _BeforeStart, _AfterStart)
858

    
859
    self.assertRaises(IndexError, queue.GetNextUpdate)
860
    self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(),
861
                     jqueue._JobProcessor.FINISHED)
862
    self.assertEqual(queue.GetNextUpdate(), (job, True))
863
    self.assertRaises(IndexError, queue.GetNextUpdate)
864

    
865
    # Check result
866
    self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_CANCELED)
867
    self.assertEqual(job.GetInfo(["status"]), [constants.JOB_STATUS_CANCELED])
868
    self.assert_(job.start_timestamp)
869
    self.assert_(job.end_timestamp)
870
    self.assertFalse(compat.all(op.start_timestamp and op.end_timestamp
871
                                for op in job.ops))
872
    self.assertEqual(job.GetInfo(["opstatus", "opresult"]),
873
                     [[constants.OP_STATUS_CANCELED for _ in job.ops],
874
                      ["Job canceled by request" for _ in job.ops]])
875

    
876
  def testCancelWhileWaitlockWithTimeout(self):
877
    queue = _FakeQueueForProc()
878

    
879
    ops = [opcodes.OpTestDummy(result="Res%s" % i, fail=False)
880
           for i in range(5)]
881

    
882
    # Create job
883
    job_id = 24314
884
    job = self._CreateJob(queue, job_id, ops)
885

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

    
888
    def _BeforeStart(timeout, priority):
889
      self.assertFalse(queue.IsAcquired())
890
      self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITING)
891

    
892
      # Mark as cancelled
893
      (success, _) = job.Cancel()
894
      self.assert_(success)
895

    
896
      self.assert_(compat.all(op.status == constants.OP_STATUS_CANCELING
897
                              for op in job.ops))
898

    
899
      # Fake an acquire attempt timing out
900
      raise mcpu.LockAcquireTimeout()
901

    
902
    def _AfterStart(op, cbs):
903
      self.fail("Should not reach this")
904

    
905
    opexec = _FakeExecOpCodeForProc(queue, _BeforeStart, _AfterStart)
906

    
907
    self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(),
908
                     jqueue._JobProcessor.FINISHED)
909

    
910
    # Check result
911
    self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_CANCELED)
912
    self.assertEqual(job.GetInfo(["status"]), [constants.JOB_STATUS_CANCELED])
913
    self.assert_(job.start_timestamp)
914
    self.assert_(job.end_timestamp)
915
    self.assertFalse(compat.all(op.start_timestamp and op.end_timestamp
916
                                for op in job.ops))
917
    self.assertEqual(job.GetInfo(["opstatus", "opresult"]),
918
                     [[constants.OP_STATUS_CANCELED for _ in job.ops],
919
                      ["Job canceled by request" for _ in job.ops]])
920

    
921
  def testCancelWhileRunning(self):
922
    # Tests canceling a job with finished opcodes and more, unprocessed ones
923
    queue = _FakeQueueForProc()
924

    
925
    ops = [opcodes.OpTestDummy(result="Res%s" % i, fail=False)
926
           for i in range(3)]
927

    
928
    # Create job
929
    job_id = 28492
930
    job = self._CreateJob(queue, job_id, ops)
931

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

    
934
    opexec = _FakeExecOpCodeForProc(queue, None, None)
935

    
936
    # Run one opcode
937
    self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(),
938
                     jqueue._JobProcessor.DEFER)
939

    
940
    # Job goes back to queued
941
    self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED)
942
    self.assertEqual(job.GetInfo(["opstatus", "opresult"]),
943
                     [[constants.OP_STATUS_SUCCESS,
944
                       constants.OP_STATUS_QUEUED,
945
                       constants.OP_STATUS_QUEUED],
946
                      ["Res0", None, None]])
947

    
948
    # Mark as cancelled
949
    (success, _) = job.Cancel()
950
    self.assert_(success)
951

    
952
    # Try processing another opcode (this will actually cancel the job)
953
    self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(),
954
                     jqueue._JobProcessor.FINISHED)
955

    
956
    # Check result
957
    self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_CANCELED)
958
    self.assertEqual(job.GetInfo(["id"]), [job_id])
959
    self.assertEqual(job.GetInfo(["status"]), [constants.JOB_STATUS_CANCELED])
960
    self.assertEqual(job.GetInfo(["opstatus", "opresult"]),
961
                     [[constants.OP_STATUS_SUCCESS,
962
                       constants.OP_STATUS_CANCELED,
963
                       constants.OP_STATUS_CANCELED],
964
                      ["Res0", "Job canceled by request",
965
                       "Job canceled by request"]])
966

    
967
  def testPartiallyRun(self):
968
    # Tests calling the processor on a job that's been partially run before the
969
    # program was restarted
970
    queue = _FakeQueueForProc()
971

    
972
    opexec = _FakeExecOpCodeForProc(queue, None, None)
973

    
974
    for job_id, successcount in [(30697, 1), (2552, 4), (12489, 9)]:
975
      ops = [opcodes.OpTestDummy(result="Res%s" % i, fail=False)
976
             for i in range(10)]
977

    
978
      # Create job
979
      job = self._CreateJob(queue, job_id, ops)
980

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

    
983
      for _ in range(successcount):
984
        self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(),
985
                         jqueue._JobProcessor.DEFER)
986

    
987
      self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED)
988
      self.assertEqual(job.GetInfo(["opstatus"]),
989
                       [[constants.OP_STATUS_SUCCESS
990
                         for _ in range(successcount)] +
991
                        [constants.OP_STATUS_QUEUED
992
                         for _ in range(len(ops) - successcount)]])
993

    
994
      self.assert_(job.ops_iter)
995

    
996
      # Serialize and restore (simulates program restart)
997
      newjob = jqueue._QueuedJob.Restore(queue, job.Serialize(), True)
998
      self.assertFalse(newjob.ops_iter)
999
      self._TestPartial(newjob, successcount)
1000

    
1001
  def _TestPartial(self, job, successcount):
1002
    self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED)
1003
    self.assertEqual(job.start_timestamp, job.ops[0].start_timestamp)
1004

    
1005
    queue = _FakeQueueForProc()
1006
    opexec = _FakeExecOpCodeForProc(queue, None, None)
1007

    
1008
    for remaining in reversed(range(len(job.ops) - successcount)):
1009
      result = jqueue._JobProcessor(queue, opexec, job)()
1010
      self.assertEqual(queue.GetNextUpdate(), (job, True))
1011
      self.assertEqual(queue.GetNextUpdate(), (job, True))
1012
      self.assertEqual(queue.GetNextUpdate(), (job, True))
1013
      self.assertRaises(IndexError, queue.GetNextUpdate)
1014

    
1015
      if remaining == 0:
1016
        # Last opcode
1017
        self.assertEqual(result, jqueue._JobProcessor.FINISHED)
1018
        break
1019

    
1020
      self.assertEqual(result, jqueue._JobProcessor.DEFER)
1021

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

    
1024
    self.assertRaises(IndexError, queue.GetNextUpdate)
1025
    self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_SUCCESS)
1026
    self.assertEqual(job.GetInfo(["status"]), [constants.JOB_STATUS_SUCCESS])
1027
    self.assertEqual(job.GetInfo(["opresult"]),
1028
                     [[op.input.result for op in job.ops]])
1029
    self.assertEqual(job.GetInfo(["opstatus"]),
1030
                     [[constants.OP_STATUS_SUCCESS for _ in job.ops]])
1031
    self.assert_(compat.all(op.start_timestamp and op.end_timestamp
1032
                            for op in job.ops))
1033

    
1034
    self._GenericCheckJob(job)
1035

    
1036
    # Calling the processor on a finished job should be a no-op
1037
    self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(),
1038
                     jqueue._JobProcessor.FINISHED)
1039
    self.assertRaises(IndexError, queue.GetNextUpdate)
1040

    
1041
    # ... also after being restored
1042
    job2 = jqueue._QueuedJob.Restore(queue, job.Serialize(), True)
1043
    # Calling the processor on a finished job should be a no-op
1044
    self.assertEqual(jqueue._JobProcessor(queue, opexec, job2)(),
1045
                     jqueue._JobProcessor.FINISHED)
1046
    self.assertRaises(IndexError, queue.GetNextUpdate)
1047

    
1048
  def testProcessorOnRunningJob(self):
1049
    ops = [opcodes.OpTestDummy(result="result", fail=False)]
1050

    
1051
    queue = _FakeQueueForProc()
1052
    opexec = _FakeExecOpCodeForProc(queue, None, None)
1053

    
1054
    # Create job
1055
    job = self._CreateJob(queue, 9571, ops)
1056

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

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

    
1061
    assert len(job.ops) == 1
1062

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

    
1065
    # Calling on running job must fail
1066
    self.assertRaises(errors.ProgrammerError,
1067
                      jqueue._JobProcessor(queue, opexec, job))
1068

    
1069
  def testLogMessages(self):
1070
    # Tests the "Feedback" callback function
1071
    queue = _FakeQueueForProc()
1072

    
1073
    messages = {
1074
      1: [
1075
        (None, "Hello"),
1076
        (None, "World"),
1077
        (constants.ELOG_MESSAGE, "there"),
1078
        ],
1079
      4: [
1080
        (constants.ELOG_JQUEUE_TEST, (1, 2, 3)),
1081
        (constants.ELOG_JQUEUE_TEST, ("other", "type")),
1082
        ],
1083
      }
1084
    ops = [opcodes.OpTestDummy(result="Logtest%s" % i, fail=False,
1085
                               messages=messages.get(i, []))
1086
           for i in range(5)]
1087

    
1088
    # Create job
1089
    job = self._CreateJob(queue, 29386, ops)
1090

    
1091
    def _BeforeStart(timeout, priority):
1092
      self.assertEqual(queue.GetNextUpdate(), (job, True))
1093
      self.assertRaises(IndexError, queue.GetNextUpdate)
1094
      self.assertFalse(queue.IsAcquired())
1095
      self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITING)
1096

    
1097
    def _AfterStart(op, cbs):
1098
      self.assertEqual(queue.GetNextUpdate(), (job, True))
1099
      self.assertRaises(IndexError, queue.GetNextUpdate)
1100
      self.assertFalse(queue.IsAcquired())
1101
      self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_RUNNING)
1102

    
1103
      self.assertRaises(AssertionError, cbs.Feedback,
1104
                        "too", "many", "arguments")
1105

    
1106
      for (log_type, msg) in op.messages:
1107
        self.assertRaises(IndexError, queue.GetNextUpdate)
1108
        if log_type:
1109
          cbs.Feedback(log_type, msg)
1110
        else:
1111
          cbs.Feedback(msg)
1112
        # Check for job update without replication
1113
        self.assertEqual(queue.GetNextUpdate(), (job, False))
1114
        self.assertRaises(IndexError, queue.GetNextUpdate)
1115

    
1116
    opexec = _FakeExecOpCodeForProc(queue, _BeforeStart, _AfterStart)
1117

    
1118
    for remaining in reversed(range(len(job.ops))):
1119
      self.assertRaises(IndexError, queue.GetNextUpdate)
1120
      result = jqueue._JobProcessor(queue, opexec, job)()
1121
      self.assertEqual(queue.GetNextUpdate(), (job, True))
1122
      self.assertRaises(IndexError, queue.GetNextUpdate)
1123

    
1124
      if remaining == 0:
1125
        # Last opcode
1126
        self.assertEqual(result, jqueue._JobProcessor.FINISHED)
1127
        break
1128

    
1129
      self.assertEqual(result, jqueue._JobProcessor.DEFER)
1130

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

    
1133
    self.assertRaises(IndexError, queue.GetNextUpdate)
1134

    
1135
    self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_SUCCESS)
1136
    self.assertEqual(job.GetInfo(["opresult"]),
1137
                     [[op.input.result for op in job.ops]])
1138

    
1139
    logmsgcount = sum(len(m) for m in messages.values())
1140

    
1141
    self._CheckLogMessages(job, logmsgcount)
1142

    
1143
    # Serialize and restore (simulates program restart)
1144
    newjob = jqueue._QueuedJob.Restore(queue, job.Serialize(), True)
1145
    self._CheckLogMessages(newjob, logmsgcount)
1146

    
1147
    # Check each message
1148
    prevserial = -1
1149
    for idx, oplog in enumerate(job.GetInfo(["oplog"])[0]):
1150
      for (serial, timestamp, log_type, msg) in oplog:
1151
        (exptype, expmsg) = messages.get(idx).pop(0)
1152
        if exptype:
1153
          self.assertEqual(log_type, exptype)
1154
        else:
1155
          self.assertEqual(log_type, constants.ELOG_MESSAGE)
1156
        self.assertEqual(expmsg, msg)
1157
        self.assert_(serial > prevserial)
1158
        prevserial = serial
1159

    
1160
  def _CheckLogMessages(self, job, count):
1161
    # Check serial
1162
    self.assertEqual(job.log_serial, count)
1163

    
1164
    # No filter
1165
    self.assertEqual(job.GetLogEntries(None),
1166
                     [entry for entries in job.GetInfo(["oplog"])[0] if entries
1167
                      for entry in entries])
1168

    
1169
    # Filter with serial
1170
    assert count > 3
1171
    self.assert_(job.GetLogEntries(3))
1172
    self.assertEqual(job.GetLogEntries(3),
1173
                     [entry for entries in job.GetInfo(["oplog"])[0] if entries
1174
                      for entry in entries][3:])
1175

    
1176
    # No log message after highest serial
1177
    self.assertFalse(job.GetLogEntries(count))
1178
    self.assertFalse(job.GetLogEntries(count + 3))
1179

    
1180
  def testSubmitManyJobs(self):
1181
    queue = _FakeQueueForProc()
1182

    
1183
    job_id = 15656
1184
    ops = [
1185
      opcodes.OpTestDummy(result="Res0", fail=False,
1186
                          submit_jobs=[]),
1187
      opcodes.OpTestDummy(result="Res1", fail=False,
1188
                          submit_jobs=[
1189
                            [opcodes.OpTestDummy(result="r1j0", fail=False)],
1190
                            ]),
1191
      opcodes.OpTestDummy(result="Res2", fail=False,
1192
                          submit_jobs=[
1193
                            [opcodes.OpTestDummy(result="r2j0o0", fail=False),
1194
                             opcodes.OpTestDummy(result="r2j0o1", fail=False),
1195
                             opcodes.OpTestDummy(result="r2j0o2", fail=False),
1196
                             opcodes.OpTestDummy(result="r2j0o3", fail=False)],
1197
                            [opcodes.OpTestDummy(result="r2j1", fail=False)],
1198
                            [opcodes.OpTestDummy(result="r2j3o0", fail=False),
1199
                             opcodes.OpTestDummy(result="r2j3o1", fail=False)],
1200
                            ]),
1201
      ]
1202

    
1203
    # Create job
1204
    job = self._CreateJob(queue, job_id, ops)
1205

    
1206
    def _BeforeStart(timeout, priority):
1207
      self.assertEqual(queue.GetNextUpdate(), (job, True))
1208
      self.assertRaises(IndexError, queue.GetNextUpdate)
1209
      self.assertFalse(queue.IsAcquired())
1210
      self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITING)
1211
      self.assertFalse(job.cur_opctx)
1212

    
1213
    def _AfterStart(op, cbs):
1214
      self.assertEqual(queue.GetNextUpdate(), (job, True))
1215
      self.assertRaises(IndexError, queue.GetNextUpdate)
1216

    
1217
      self.assertFalse(queue.IsAcquired())
1218
      self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_RUNNING)
1219
      self.assertFalse(job.cur_opctx)
1220

    
1221
      # Job is running, cancelling shouldn't be possible
1222
      (success, _) = job.Cancel()
1223
      self.assertFalse(success)
1224

    
1225
    opexec = _FakeExecOpCodeForProc(queue, _BeforeStart, _AfterStart)
1226

    
1227
    for idx in range(len(ops)):
1228
      self.assertRaises(IndexError, queue.GetNextUpdate)
1229
      result = jqueue._JobProcessor(queue, opexec, job)()
1230
      self.assertEqual(queue.GetNextUpdate(), (job, True))
1231
      self.assertRaises(IndexError, queue.GetNextUpdate)
1232
      if idx == len(ops) - 1:
1233
        # Last opcode
1234
        self.assertEqual(result, jqueue._JobProcessor.FINISHED)
1235
      else:
1236
        self.assertEqual(result, jqueue._JobProcessor.DEFER)
1237

    
1238
        self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED)
1239
        self.assert_(job.start_timestamp)
1240
        self.assertFalse(job.end_timestamp)
1241

    
1242
    self.assertRaises(IndexError, queue.GetNextUpdate)
1243

    
1244
    for idx, submitted_ops in enumerate(job_ops
1245
                                        for op in ops
1246
                                        for job_ops in op.submit_jobs):
1247
      self.assertEqual(queue.GetNextSubmittedJob(),
1248
                       (1000 + idx, submitted_ops))
1249
    self.assertRaises(IndexError, queue.GetNextSubmittedJob)
1250

    
1251
    self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_SUCCESS)
1252
    self.assertEqual(job.GetInfo(["status"]), [constants.JOB_STATUS_SUCCESS])
1253
    self.assertEqual(job.GetInfo(["opresult"]),
1254
                     [[[], [1000], [1001, 1002, 1003]]])
1255
    self.assertEqual(job.GetInfo(["opstatus"]),
1256
                     [len(job.ops) * [constants.OP_STATUS_SUCCESS]])
1257

    
1258
    self._GenericCheckJob(job)
1259

    
1260
    # Calling the processor on a finished job should be a no-op
1261
    self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(),
1262
                     jqueue._JobProcessor.FINISHED)
1263
    self.assertRaises(IndexError, queue.GetNextUpdate)
1264

    
1265
  def testJobDependency(self):
1266
    depmgr = _FakeDependencyManager()
1267
    queue = _FakeQueueForProc(depmgr=depmgr)
1268

    
1269
    self.assertEqual(queue.depmgr, depmgr)
1270

    
1271
    prev_job_id = 22113
1272
    prev_job_id2 = 28102
1273
    job_id = 29929
1274
    ops = [
1275
      opcodes.OpTestDummy(result="Res0", fail=False,
1276
                          depends=[
1277
                            [prev_job_id2, None],
1278
                            [prev_job_id, None],
1279
                            ]),
1280
      opcodes.OpTestDummy(result="Res1", fail=False),
1281
      ]
1282

    
1283
    # Create job
1284
    job = self._CreateJob(queue, job_id, ops)
1285

    
1286
    def _BeforeStart(timeout, priority):
1287
      if attempt == 0 or attempt > 5:
1288
        # Job should only be updated when it wasn't waiting for another job
1289
        self.assertEqual(queue.GetNextUpdate(), (job, True))
1290
      self.assertRaises(IndexError, queue.GetNextUpdate)
1291
      self.assertFalse(queue.IsAcquired())
1292
      self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITING)
1293
      self.assertFalse(job.cur_opctx)
1294

    
1295
    def _AfterStart(op, cbs):
1296
      self.assertEqual(queue.GetNextUpdate(), (job, True))
1297
      self.assertRaises(IndexError, queue.GetNextUpdate)
1298

    
1299
      self.assertFalse(queue.IsAcquired())
1300
      self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_RUNNING)
1301
      self.assertFalse(job.cur_opctx)
1302

    
1303
      # Job is running, cancelling shouldn't be possible
1304
      (success, _) = job.Cancel()
1305
      self.assertFalse(success)
1306

    
1307
    opexec = _FakeExecOpCodeForProc(queue, _BeforeStart, _AfterStart)
1308

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

    
1311
    counter = itertools.count()
1312
    while True:
1313
      attempt = counter.next()
1314

    
1315
      self.assertRaises(IndexError, queue.GetNextUpdate)
1316
      self.assertRaises(IndexError, depmgr.GetNextNotification)
1317

    
1318
      if attempt < 2:
1319
        depmgr.AddCheckResult(job, prev_job_id2, None,
1320
                              (jqueue._JobDependencyManager.WAIT, "wait2"))
1321
      elif attempt == 2:
1322
        depmgr.AddCheckResult(job, prev_job_id2, None,
1323
                              (jqueue._JobDependencyManager.CONTINUE, "cont"))
1324
        # The processor will ask for the next dependency immediately
1325
        depmgr.AddCheckResult(job, prev_job_id, None,
1326
                              (jqueue._JobDependencyManager.WAIT, "wait"))
1327
      elif attempt < 5:
1328
        depmgr.AddCheckResult(job, prev_job_id, None,
1329
                              (jqueue._JobDependencyManager.WAIT, "wait"))
1330
      elif attempt == 5:
1331
        depmgr.AddCheckResult(job, prev_job_id, None,
1332
                              (jqueue._JobDependencyManager.CONTINUE, "cont"))
1333
      if attempt == 2:
1334
        self.assertEqual(depmgr.CountPendingResults(), 2)
1335
      elif attempt > 5:
1336
        self.assertEqual(depmgr.CountPendingResults(), 0)
1337
      else:
1338
        self.assertEqual(depmgr.CountPendingResults(), 1)
1339

    
1340
      result = jqueue._JobProcessor(queue, opexec, job)()
1341
      if attempt == 0 or attempt >= 5:
1342
        # Job should only be updated if there was an actual change
1343
        self.assertEqual(queue.GetNextUpdate(), (job, True))
1344
      self.assertRaises(IndexError, queue.GetNextUpdate)
1345
      self.assertFalse(depmgr.CountPendingResults())
1346

    
1347
      if attempt < 5:
1348
        # Simulate waiting for other job
1349
        self.assertEqual(result, jqueue._JobProcessor.WAITDEP)
1350
        self.assertTrue(job.cur_opctx)
1351
        self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITING)
1352
        self.assertRaises(IndexError, depmgr.GetNextNotification)
1353
        self.assert_(job.start_timestamp)
1354
        self.assertFalse(job.end_timestamp)
1355
        continue
1356

    
1357
      if result == jqueue._JobProcessor.FINISHED:
1358
        # Last opcode
1359
        self.assertFalse(job.cur_opctx)
1360
        break
1361

    
1362
      self.assertRaises(IndexError, depmgr.GetNextNotification)
1363

    
1364
      self.assertEqual(result, jqueue._JobProcessor.DEFER)
1365
      self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED)
1366
      self.assert_(job.start_timestamp)
1367
      self.assertFalse(job.end_timestamp)
1368

    
1369
    self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_SUCCESS)
1370
    self.assertEqual(job.GetInfo(["status"]), [constants.JOB_STATUS_SUCCESS])
1371
    self.assertEqual(job.GetInfo(["opresult"]),
1372
                     [[op.input.result for op in job.ops]])
1373
    self.assertEqual(job.GetInfo(["opstatus"]),
1374
                     [len(job.ops) * [constants.OP_STATUS_SUCCESS]])
1375
    self.assertTrue(compat.all(op.start_timestamp and op.end_timestamp
1376
                               for op in job.ops))
1377

    
1378
    self._GenericCheckJob(job)
1379

    
1380
    self.assertRaises(IndexError, queue.GetNextUpdate)
1381
    self.assertRaises(IndexError, depmgr.GetNextNotification)
1382
    self.assertFalse(depmgr.CountPendingResults())
1383
    self.assertFalse(depmgr.CountWaitingJobs())
1384

    
1385
    # Calling the processor on a finished job should be a no-op
1386
    self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(),
1387
                     jqueue._JobProcessor.FINISHED)
1388
    self.assertRaises(IndexError, queue.GetNextUpdate)
1389

    
1390
  def testJobDependencyCancel(self):
1391
    depmgr = _FakeDependencyManager()
1392
    queue = _FakeQueueForProc(depmgr=depmgr)
1393

    
1394
    self.assertEqual(queue.depmgr, depmgr)
1395

    
1396
    prev_job_id = 13623
1397
    job_id = 30876
1398
    ops = [
1399
      opcodes.OpTestDummy(result="Res0", fail=False),
1400
      opcodes.OpTestDummy(result="Res1", fail=False,
1401
                          depends=[
1402
                            [prev_job_id, None],
1403
                            ]),
1404
      opcodes.OpTestDummy(result="Res2", fail=False),
1405
      ]
1406

    
1407
    # Create job
1408
    job = self._CreateJob(queue, job_id, ops)
1409

    
1410
    def _BeforeStart(timeout, priority):
1411
      if attempt == 0 or attempt > 5:
1412
        # Job should only be updated when it wasn't waiting for another job
1413
        self.assertEqual(queue.GetNextUpdate(), (job, True))
1414
      self.assertRaises(IndexError, queue.GetNextUpdate)
1415
      self.assertFalse(queue.IsAcquired())
1416
      self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITING)
1417
      self.assertFalse(job.cur_opctx)
1418

    
1419
    def _AfterStart(op, cbs):
1420
      self.assertEqual(queue.GetNextUpdate(), (job, True))
1421
      self.assertRaises(IndexError, queue.GetNextUpdate)
1422

    
1423
      self.assertFalse(queue.IsAcquired())
1424
      self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_RUNNING)
1425
      self.assertFalse(job.cur_opctx)
1426

    
1427
      # Job is running, cancelling shouldn't be possible
1428
      (success, _) = job.Cancel()
1429
      self.assertFalse(success)
1430

    
1431
    opexec = _FakeExecOpCodeForProc(queue, _BeforeStart, _AfterStart)
1432

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

    
1435
    counter = itertools.count()
1436
    while True:
1437
      attempt = counter.next()
1438

    
1439
      self.assertRaises(IndexError, queue.GetNextUpdate)
1440
      self.assertRaises(IndexError, depmgr.GetNextNotification)
1441

    
1442
      if attempt == 0:
1443
        # This will handle the first opcode
1444
        pass
1445
      elif attempt < 4:
1446
        depmgr.AddCheckResult(job, prev_job_id, None,
1447
                              (jqueue._JobDependencyManager.WAIT, "wait"))
1448
      elif attempt == 4:
1449
        # Other job was cancelled
1450
        depmgr.AddCheckResult(job, prev_job_id, None,
1451
                              (jqueue._JobDependencyManager.CANCEL, "cancel"))
1452

    
1453
      if attempt == 0:
1454
        self.assertEqual(depmgr.CountPendingResults(), 0)
1455
      else:
1456
        self.assertEqual(depmgr.CountPendingResults(), 1)
1457

    
1458
      result = jqueue._JobProcessor(queue, opexec, job)()
1459
      if attempt <= 1 or attempt >= 4:
1460
        # Job should only be updated if there was an actual change
1461
        self.assertEqual(queue.GetNextUpdate(), (job, True))
1462
      self.assertRaises(IndexError, queue.GetNextUpdate)
1463
      self.assertFalse(depmgr.CountPendingResults())
1464

    
1465
      if attempt > 0 and attempt < 4:
1466
        # Simulate waiting for other job
1467
        self.assertEqual(result, jqueue._JobProcessor.WAITDEP)
1468
        self.assertTrue(job.cur_opctx)
1469
        self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITING)
1470
        self.assertRaises(IndexError, depmgr.GetNextNotification)
1471
        self.assert_(job.start_timestamp)
1472
        self.assertFalse(job.end_timestamp)
1473
        continue
1474

    
1475
      if result == jqueue._JobProcessor.FINISHED:
1476
        # Last opcode
1477
        self.assertFalse(job.cur_opctx)
1478
        break
1479

    
1480
      self.assertRaises(IndexError, depmgr.GetNextNotification)
1481

    
1482
      self.assertEqual(result, jqueue._JobProcessor.DEFER)
1483
      self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED)
1484
      self.assert_(job.start_timestamp)
1485
      self.assertFalse(job.end_timestamp)
1486

    
1487
    self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_CANCELED)
1488
    self.assertEqual(job.GetInfo(["status"]), [constants.JOB_STATUS_CANCELED])
1489
    self.assertEqual(job.GetInfo(["opstatus", "opresult"]),
1490
                     [[constants.OP_STATUS_SUCCESS,
1491
                       constants.OP_STATUS_CANCELED,
1492
                       constants.OP_STATUS_CANCELED],
1493
                      ["Res0", "Job canceled by request",
1494
                       "Job canceled by request"]])
1495

    
1496
    self._GenericCheckJob(job)
1497

    
1498
    self.assertRaises(IndexError, queue.GetNextUpdate)
1499
    self.assertRaises(IndexError, depmgr.GetNextNotification)
1500
    self.assertFalse(depmgr.CountPendingResults())
1501

    
1502
    # Calling the processor on a finished job should be a no-op
1503
    self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(),
1504
                     jqueue._JobProcessor.FINISHED)
1505
    self.assertRaises(IndexError, queue.GetNextUpdate)
1506

    
1507
  def testJobDependencyWrongstatus(self):
1508
    depmgr = _FakeDependencyManager()
1509
    queue = _FakeQueueForProc(depmgr=depmgr)
1510

    
1511
    self.assertEqual(queue.depmgr, depmgr)
1512

    
1513
    prev_job_id = 9741
1514
    job_id = 11763
1515
    ops = [
1516
      opcodes.OpTestDummy(result="Res0", fail=False),
1517
      opcodes.OpTestDummy(result="Res1", fail=False,
1518
                          depends=[
1519
                            [prev_job_id, None],
1520
                            ]),
1521
      opcodes.OpTestDummy(result="Res2", fail=False),
1522
      ]
1523

    
1524
    # Create job
1525
    job = self._CreateJob(queue, job_id, ops)
1526

    
1527
    def _BeforeStart(timeout, priority):
1528
      if attempt == 0 or attempt > 5:
1529
        # Job should only be updated when it wasn't waiting for another job
1530
        self.assertEqual(queue.GetNextUpdate(), (job, True))
1531
      self.assertRaises(IndexError, queue.GetNextUpdate)
1532
      self.assertFalse(queue.IsAcquired())
1533
      self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITING)
1534
      self.assertFalse(job.cur_opctx)
1535

    
1536
    def _AfterStart(op, cbs):
1537
      self.assertEqual(queue.GetNextUpdate(), (job, True))
1538
      self.assertRaises(IndexError, queue.GetNextUpdate)
1539

    
1540
      self.assertFalse(queue.IsAcquired())
1541
      self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_RUNNING)
1542
      self.assertFalse(job.cur_opctx)
1543

    
1544
      # Job is running, cancelling shouldn't be possible
1545
      (success, _) = job.Cancel()
1546
      self.assertFalse(success)
1547

    
1548
    opexec = _FakeExecOpCodeForProc(queue, _BeforeStart, _AfterStart)
1549

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

    
1552
    counter = itertools.count()
1553
    while True:
1554
      attempt = counter.next()
1555

    
1556
      self.assertRaises(IndexError, queue.GetNextUpdate)
1557
      self.assertRaises(IndexError, depmgr.GetNextNotification)
1558

    
1559
      if attempt == 0:
1560
        # This will handle the first opcode
1561
        pass
1562
      elif attempt < 4:
1563
        depmgr.AddCheckResult(job, prev_job_id, None,
1564
                              (jqueue._JobDependencyManager.WAIT, "wait"))
1565
      elif attempt == 4:
1566
        # Other job failed
1567
        depmgr.AddCheckResult(job, prev_job_id, None,
1568
                              (jqueue._JobDependencyManager.WRONGSTATUS, "w"))
1569

    
1570
      if attempt == 0:
1571
        self.assertEqual(depmgr.CountPendingResults(), 0)
1572
      else:
1573
        self.assertEqual(depmgr.CountPendingResults(), 1)
1574

    
1575
      result = jqueue._JobProcessor(queue, opexec, job)()
1576
      if attempt <= 1 or attempt >= 4:
1577
        # Job should only be updated if there was an actual change
1578
        self.assertEqual(queue.GetNextUpdate(), (job, True))
1579
      self.assertRaises(IndexError, queue.GetNextUpdate)
1580
      self.assertFalse(depmgr.CountPendingResults())
1581

    
1582
      if attempt > 0 and attempt < 4:
1583
        # Simulate waiting for other job
1584
        self.assertEqual(result, jqueue._JobProcessor.WAITDEP)
1585
        self.assertTrue(job.cur_opctx)
1586
        self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITING)
1587
        self.assertRaises(IndexError, depmgr.GetNextNotification)
1588
        self.assert_(job.start_timestamp)
1589
        self.assertFalse(job.end_timestamp)
1590
        continue
1591

    
1592
      if result == jqueue._JobProcessor.FINISHED:
1593
        # Last opcode
1594
        self.assertFalse(job.cur_opctx)
1595
        break
1596

    
1597
      self.assertRaises(IndexError, depmgr.GetNextNotification)
1598

    
1599
      self.assertEqual(result, jqueue._JobProcessor.DEFER)
1600
      self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED)
1601
      self.assert_(job.start_timestamp)
1602
      self.assertFalse(job.end_timestamp)
1603

    
1604
    self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_ERROR)
1605
    self.assertEqual(job.GetInfo(["status"]), [constants.JOB_STATUS_ERROR])
1606
    self.assertEqual(job.GetInfo(["opstatus"]),
1607
                     [[constants.OP_STATUS_SUCCESS,
1608
                       constants.OP_STATUS_ERROR,
1609
                       constants.OP_STATUS_ERROR]]),
1610

    
1611
    (opresult, ) = job.GetInfo(["opresult"])
1612
    self.assertEqual(len(opresult), len(ops))
1613
    self.assertEqual(opresult[0], "Res0")
1614
    self.assertTrue(errors.GetEncodedError(opresult[1]))
1615
    self.assertTrue(errors.GetEncodedError(opresult[2]))
1616

    
1617
    self._GenericCheckJob(job)
1618

    
1619
    self.assertRaises(IndexError, queue.GetNextUpdate)
1620
    self.assertRaises(IndexError, depmgr.GetNextNotification)
1621
    self.assertFalse(depmgr.CountPendingResults())
1622

    
1623
    # Calling the processor on a finished job should be a no-op
1624
    self.assertEqual(jqueue._JobProcessor(queue, opexec, job)(),
1625
                     jqueue._JobProcessor.FINISHED)
1626
    self.assertRaises(IndexError, queue.GetNextUpdate)
1627

    
1628

    
1629
class TestEvaluateJobProcessorResult(unittest.TestCase):
1630
  def testFinished(self):
1631
    depmgr = _FakeDependencyManager()
1632
    job = _IdOnlyFakeJob(30953)
1633
    jqueue._EvaluateJobProcessorResult(depmgr, job,
1634
                                       jqueue._JobProcessor.FINISHED)
1635
    self.assertEqual(depmgr.GetNextNotification(), job.id)
1636
    self.assertRaises(IndexError, depmgr.GetNextNotification)
1637

    
1638
  def testDefer(self):
1639
    depmgr = _FakeDependencyManager()
1640
    job = _IdOnlyFakeJob(11326, priority=5463)
1641
    try:
1642
      jqueue._EvaluateJobProcessorResult(depmgr, job,
1643
                                         jqueue._JobProcessor.DEFER)
1644
    except workerpool.DeferTask, err:
1645
      self.assertEqual(err.priority, 5463)
1646
    else:
1647
      self.fail("Didn't raise exception")
1648
    self.assertRaises(IndexError, depmgr.GetNextNotification)
1649

    
1650
  def testWaitdep(self):
1651
    depmgr = _FakeDependencyManager()
1652
    job = _IdOnlyFakeJob(21317)
1653
    jqueue._EvaluateJobProcessorResult(depmgr, job,
1654
                                       jqueue._JobProcessor.WAITDEP)
1655
    self.assertRaises(IndexError, depmgr.GetNextNotification)
1656

    
1657
  def testOther(self):
1658
    depmgr = _FakeDependencyManager()
1659
    job = _IdOnlyFakeJob(5813)
1660
    self.assertRaises(errors.ProgrammerError,
1661
                      jqueue._EvaluateJobProcessorResult,
1662
                      depmgr, job, "Other result")
1663
    self.assertRaises(IndexError, depmgr.GetNextNotification)
1664

    
1665

    
1666
class _FakeTimeoutStrategy:
1667
  def __init__(self, timeouts):
1668
    self.timeouts = timeouts
1669
    self.attempts = 0
1670
    self.last_timeout = None
1671

    
1672
  def NextAttempt(self):
1673
    self.attempts += 1
1674
    if self.timeouts:
1675
      timeout = self.timeouts.pop(0)
1676
    else:
1677
      timeout = None
1678
    self.last_timeout = timeout
1679
    return timeout
1680

    
1681

    
1682
class TestJobProcessorTimeouts(unittest.TestCase, _JobProcessorTestUtils):
1683
  def setUp(self):
1684
    self.queue = _FakeQueueForProc()
1685
    self.job = None
1686
    self.curop = None
1687
    self.opcounter = None
1688
    self.timeout_strategy = None
1689
    self.retries = 0
1690
    self.prev_tsop = None
1691
    self.prev_prio = None
1692
    self.prev_status = None
1693
    self.lock_acq_prio = None
1694
    self.gave_lock = None
1695
    self.done_lock_before_blocking = False
1696

    
1697
  def _BeforeStart(self, timeout, priority):
1698
    job = self.job
1699

    
1700
    # If status has changed, job must've been written
1701
    if self.prev_status != self.job.ops[self.curop].status:
1702
      self.assertEqual(self.queue.GetNextUpdate(), (job, True))
1703
    self.assertRaises(IndexError, self.queue.GetNextUpdate)
1704

    
1705
    self.assertFalse(self.queue.IsAcquired())
1706
    self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITING)
1707

    
1708
    ts = self.timeout_strategy
1709

    
1710
    self.assert_(timeout is None or isinstance(timeout, (int, float)))
1711
    self.assertEqual(timeout, ts.last_timeout)
1712
    self.assertEqual(priority, job.ops[self.curop].priority)
1713

    
1714
    self.gave_lock = True
1715
    self.lock_acq_prio = priority
1716

    
1717
    if (self.curop == 3 and
1718
        job.ops[self.curop].priority == constants.OP_PRIO_HIGHEST + 3):
1719
      # Give locks before running into blocking acquire
1720
      assert self.retries == 7
1721
      self.retries = 0
1722
      self.done_lock_before_blocking = True
1723
      return
1724

    
1725
    if self.retries > 0:
1726
      self.assert_(timeout is not None)
1727
      self.retries -= 1
1728
      self.gave_lock = False
1729
      raise mcpu.LockAcquireTimeout()
1730

    
1731
    if job.ops[self.curop].priority == constants.OP_PRIO_HIGHEST:
1732
      assert self.retries == 0, "Didn't exhaust all retries at highest priority"
1733
      assert not ts.timeouts
1734
      self.assert_(timeout is None)
1735

    
1736
  def _AfterStart(self, op, cbs):
1737
    job = self.job
1738

    
1739
    # Setting to "running" requires an update
1740
    self.assertEqual(self.queue.GetNextUpdate(), (job, True))
1741
    self.assertRaises(IndexError, self.queue.GetNextUpdate)
1742

    
1743
    self.assertFalse(self.queue.IsAcquired())
1744
    self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_RUNNING)
1745

    
1746
    # Job is running, cancelling shouldn't be possible
1747
    (success, _) = job.Cancel()
1748
    self.assertFalse(success)
1749

    
1750
  def _NextOpcode(self):
1751
    self.curop = self.opcounter.next()
1752
    self.prev_prio = self.job.ops[self.curop].priority
1753
    self.prev_status = self.job.ops[self.curop].status
1754

    
1755
  def _NewTimeoutStrategy(self):
1756
    job = self.job
1757

    
1758
    self.assertEqual(self.retries, 0)
1759

    
1760
    if self.prev_tsop == self.curop:
1761
      # Still on the same opcode, priority must've been increased
1762
      self.assertEqual(self.prev_prio, job.ops[self.curop].priority + 1)
1763

    
1764
    if self.curop == 1:
1765
      # Normal retry
1766
      timeouts = range(10, 31, 10)
1767
      self.retries = len(timeouts) - 1
1768

    
1769
    elif self.curop == 2:
1770
      # Let this run into a blocking acquire
1771
      timeouts = range(11, 61, 12)
1772
      self.retries = len(timeouts)
1773

    
1774
    elif self.curop == 3:
1775
      # Wait for priority to increase, but give lock before blocking acquire
1776
      timeouts = range(12, 100, 14)
1777
      self.retries = len(timeouts)
1778

    
1779
      self.assertFalse(self.done_lock_before_blocking)
1780

    
1781
    elif self.curop == 4:
1782
      self.assert_(self.done_lock_before_blocking)
1783

    
1784
      # Timeouts, but no need to retry
1785
      timeouts = range(10, 31, 10)
1786
      self.retries = 0
1787

    
1788
    elif self.curop == 5:
1789
      # Normal retry
1790
      timeouts = range(19, 100, 11)
1791
      self.retries = len(timeouts)
1792

    
1793
    else:
1794
      timeouts = []
1795
      self.retries = 0
1796

    
1797
    assert len(job.ops) == 10
1798
    assert self.retries <= len(timeouts)
1799

    
1800
    ts = _FakeTimeoutStrategy(timeouts)
1801

    
1802
    self.timeout_strategy = ts
1803
    self.prev_tsop = self.curop
1804
    self.prev_prio = job.ops[self.curop].priority
1805

    
1806
    return ts
1807

    
1808
  def testTimeout(self):
1809
    ops = [opcodes.OpTestDummy(result="Res%s" % i, fail=False)
1810
           for i in range(10)]
1811

    
1812
    # Create job
1813
    job_id = 15801
1814
    job = self._CreateJob(self.queue, job_id, ops)
1815
    self.job = job
1816

    
1817
    self.opcounter = itertools.count(0)
1818

    
1819
    opexec = _FakeExecOpCodeForProc(self.queue, self._BeforeStart,
1820
                                    self._AfterStart)
1821
    tsf = self._NewTimeoutStrategy
1822

    
1823
    self.assertFalse(self.done_lock_before_blocking)
1824

    
1825
    while True:
1826
      proc = jqueue._JobProcessor(self.queue, opexec, job,
1827
                                  _timeout_strategy_factory=tsf)
1828

    
1829
      self.assertRaises(IndexError, self.queue.GetNextUpdate)
1830

    
1831
      if self.curop is not None:
1832
        self.prev_status = self.job.ops[self.curop].status
1833

    
1834
      self.lock_acq_prio = None
1835

    
1836
      result = proc(_nextop_fn=self._NextOpcode)
1837
      assert self.curop is not None
1838

    
1839
      if result == jqueue._JobProcessor.FINISHED or self.gave_lock:
1840
        # Got lock and/or job is done, result must've been written
1841
        self.assertFalse(job.cur_opctx)
1842
        self.assertEqual(self.queue.GetNextUpdate(), (job, True))
1843
        self.assertRaises(IndexError, self.queue.GetNextUpdate)
1844
        self.assertEqual(self.lock_acq_prio, job.ops[self.curop].priority)
1845
        self.assert_(job.ops[self.curop].exec_timestamp)
1846

    
1847
      if result == jqueue._JobProcessor.FINISHED:
1848
        self.assertFalse(job.cur_opctx)
1849
        break
1850

    
1851
      self.assertEqual(result, jqueue._JobProcessor.DEFER)
1852

    
1853
      if self.curop == 0:
1854
        self.assertEqual(job.ops[self.curop].start_timestamp,
1855
                         job.start_timestamp)
1856

    
1857
      if self.gave_lock:
1858
        # Opcode finished, but job not yet done
1859
        self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_QUEUED)
1860
      else:
1861
        # Did not get locks
1862
        self.assert_(job.cur_opctx)
1863
        self.assertEqual(job.cur_opctx._timeout_strategy._fn,
1864
                         self.timeout_strategy.NextAttempt)
1865
        self.assertFalse(job.ops[self.curop].exec_timestamp)
1866
        self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_WAITING)
1867

    
1868
        # If priority has changed since acquiring locks, the job must've been
1869
        # updated
1870
        if self.lock_acq_prio != job.ops[self.curop].priority:
1871
          self.assertEqual(self.queue.GetNextUpdate(), (job, True))
1872

    
1873
      self.assertRaises(IndexError, self.queue.GetNextUpdate)
1874

    
1875
      self.assert_(job.start_timestamp)
1876
      self.assertFalse(job.end_timestamp)
1877

    
1878
    self.assertEqual(self.curop, len(job.ops) - 1)
1879
    self.assertEqual(self.job, job)
1880
    self.assertEqual(self.opcounter.next(), len(job.ops))
1881
    self.assert_(self.done_lock_before_blocking)
1882

    
1883
    self.assertRaises(IndexError, self.queue.GetNextUpdate)
1884
    self.assertEqual(job.CalcStatus(), constants.JOB_STATUS_SUCCESS)
1885
    self.assertEqual(job.GetInfo(["status"]), [constants.JOB_STATUS_SUCCESS])
1886
    self.assertEqual(job.GetInfo(["opresult"]),
1887
                     [[op.input.result for op in job.ops]])
1888
    self.assertEqual(job.GetInfo(["opstatus"]),
1889
                     [len(job.ops) * [constants.OP_STATUS_SUCCESS]])
1890
    self.assert_(compat.all(op.start_timestamp and op.end_timestamp
1891
                            for op in job.ops))
1892

    
1893
    # Calling the processor on a finished job should be a no-op
1894
    self.assertEqual(jqueue._JobProcessor(self.queue, opexec, job)(),
1895
                     jqueue._JobProcessor.FINISHED)
1896
    self.assertRaises(IndexError, self.queue.GetNextUpdate)
1897

    
1898

    
1899
class _IdOnlyFakeJob:
1900
  def __init__(self, job_id, priority=NotImplemented):
1901
    self.id = str(job_id)
1902
    self._priority = priority
1903

    
1904
  def CalcPriority(self):
1905
    return self._priority
1906

    
1907

    
1908
class TestJobDependencyManager(unittest.TestCase):
1909
  def setUp(self):
1910
    self._status = []
1911
    self._queue = []
1912
    self.jdm = jqueue._JobDependencyManager(self._GetStatus, self._Enqueue)
1913

    
1914
  def _GetStatus(self, job_id):
1915
    (exp_job_id, result) = self._status.pop(0)
1916
    self.assertEqual(exp_job_id, job_id)
1917
    return result
1918

    
1919
  def _Enqueue(self, jobs):
1920
    self._queue.append(jobs)
1921

    
1922
  def testNotFinalizedThenCancel(self):
1923
    job = _IdOnlyFakeJob(17697)
1924
    job_id = str(28625)
1925

    
1926
    self._status.append((job_id, constants.JOB_STATUS_RUNNING))
1927
    (result, _) = self.jdm.CheckAndRegister(job, job_id, [])
1928
    self.assertEqual(result, self.jdm.WAIT)
1929
    self.assertFalse(self._status)
1930
    self.assertFalse(self._queue)
1931
    self.assertTrue(self.jdm.JobWaiting(job))
1932
    self.assertEqual(self.jdm._waiters, {
1933
      job_id: set([job]),
1934
      })
1935
    self.assertEqual(self.jdm.GetLockInfo([query.LQ_PENDING]), [
1936
      ("job/28625", None, None, [("job", [job.id])])
1937
      ])
1938

    
1939
    self._status.append((job_id, constants.JOB_STATUS_CANCELED))
1940
    (result, _) = self.jdm.CheckAndRegister(job, job_id, [])
1941
    self.assertEqual(result, self.jdm.CANCEL)
1942
    self.assertFalse(self._status)
1943
    self.assertFalse(self._queue)
1944
    self.assertFalse(self.jdm.JobWaiting(job))
1945
    self.assertFalse(self.jdm.GetLockInfo([query.LQ_PENDING]))
1946

    
1947
  def testRequireCancel(self):
1948
    job = _IdOnlyFakeJob(5278)
1949
    job_id = str(9610)
1950
    dep_status = [constants.JOB_STATUS_CANCELED]
1951

    
1952
    self._status.append((job_id, constants.JOB_STATUS_WAITING))
1953
    (result, _) = self.jdm.CheckAndRegister(job, job_id, dep_status)
1954
    self.assertEqual(result, self.jdm.WAIT)
1955
    self.assertFalse(self._status)
1956
    self.assertFalse(self._queue)
1957
    self.assertTrue(self.jdm.JobWaiting(job))
1958
    self.assertEqual(self.jdm._waiters, {
1959
      job_id: set([job]),
1960
      })
1961
    self.assertEqual(self.jdm.GetLockInfo([query.LQ_PENDING]), [
1962
      ("job/9610", None, None, [("job", [job.id])])
1963
      ])
1964

    
1965
    self._status.append((job_id, constants.JOB_STATUS_CANCELED))
1966
    (result, _) = self.jdm.CheckAndRegister(job, job_id, dep_status)
1967
    self.assertEqual(result, self.jdm.CONTINUE)
1968
    self.assertFalse(self._status)
1969
    self.assertFalse(self._queue)
1970
    self.assertFalse(self.jdm.JobWaiting(job))
1971
    self.assertFalse(self.jdm.GetLockInfo([query.LQ_PENDING]))
1972

    
1973
  def testRequireError(self):
1974
    job = _IdOnlyFakeJob(21459)
1975
    job_id = str(25519)
1976
    dep_status = [constants.JOB_STATUS_ERROR]
1977

    
1978
    self._status.append((job_id, constants.JOB_STATUS_WAITING))
1979
    (result, _) = self.jdm.CheckAndRegister(job, job_id, dep_status)
1980
    self.assertEqual(result, self.jdm.WAIT)
1981
    self.assertFalse(self._status)
1982
    self.assertFalse(self._queue)
1983
    self.assertTrue(self.jdm.JobWaiting(job))
1984
    self.assertEqual(self.jdm._waiters, {
1985
      job_id: set([job]),
1986
      })
1987

    
1988
    self._status.append((job_id, constants.JOB_STATUS_ERROR))
1989
    (result, _) = self.jdm.CheckAndRegister(job, job_id, dep_status)
1990
    self.assertEqual(result, self.jdm.CONTINUE)
1991
    self.assertFalse(self._status)
1992
    self.assertFalse(self._queue)
1993
    self.assertFalse(self.jdm.JobWaiting(job))
1994
    self.assertFalse(self.jdm.GetLockInfo([query.LQ_PENDING]))
1995

    
1996
  def testRequireMultiple(self):
1997
    dep_status = list(constants.JOBS_FINALIZED)
1998

    
1999
    for end_status in dep_status:
2000
      job = _IdOnlyFakeJob(21343)
2001
      job_id = str(14609)
2002

    
2003
      self._status.append((job_id, constants.JOB_STATUS_WAITING))
2004
      (result, _) = self.jdm.CheckAndRegister(job, job_id, dep_status)
2005
      self.assertEqual(result, self.jdm.WAIT)
2006
      self.assertFalse(self._status)
2007
      self.assertFalse(self._queue)
2008
      self.assertTrue(self.jdm.JobWaiting(job))
2009
      self.assertEqual(self.jdm._waiters, {
2010
        job_id: set([job]),
2011
        })
2012
      self.assertEqual(self.jdm.GetLockInfo([query.LQ_PENDING]), [
2013
        ("job/14609", None, None, [("job", [job.id])])
2014
        ])
2015

    
2016
      self._status.append((job_id, end_status))
2017
      (result, _) = self.jdm.CheckAndRegister(job, job_id, dep_status)
2018
      self.assertEqual(result, self.jdm.CONTINUE)
2019
      self.assertFalse(self._status)
2020
      self.assertFalse(self._queue)
2021
      self.assertFalse(self.jdm.JobWaiting(job))
2022
      self.assertFalse(self.jdm.GetLockInfo([query.LQ_PENDING]))
2023

    
2024
  def testNotify(self):
2025
    job = _IdOnlyFakeJob(8227)
2026
    job_id = str(4113)
2027

    
2028
    self._status.append((job_id, constants.JOB_STATUS_RUNNING))
2029
    (result, _) = self.jdm.CheckAndRegister(job, job_id, [])
2030
    self.assertEqual(result, self.jdm.WAIT)
2031
    self.assertFalse(self._status)
2032
    self.assertFalse(self._queue)
2033
    self.assertTrue(self.jdm.JobWaiting(job))
2034
    self.assertEqual(self.jdm._waiters, {
2035
      job_id: set([job]),
2036
      })
2037

    
2038
    self.jdm.NotifyWaiters(job_id)
2039
    self.assertFalse(self._status)
2040
    self.assertFalse(self.jdm._waiters)
2041
    self.assertFalse(self.jdm.JobWaiting(job))
2042
    self.assertEqual(self._queue, [set([job])])
2043

    
2044
  def testWrongStatus(self):
2045
    job = _IdOnlyFakeJob(10102)
2046
    job_id = str(1271)
2047

    
2048
    self._status.append((job_id, constants.JOB_STATUS_QUEUED))
2049
    (result, _) = self.jdm.CheckAndRegister(job, job_id,
2050
                                            [constants.JOB_STATUS_SUCCESS])
2051
    self.assertEqual(result, self.jdm.WAIT)
2052
    self.assertFalse(self._status)
2053
    self.assertFalse(self._queue)
2054
    self.assertTrue(self.jdm.JobWaiting(job))
2055
    self.assertEqual(self.jdm._waiters, {
2056
      job_id: set([job]),
2057
      })
2058

    
2059
    self._status.append((job_id, constants.JOB_STATUS_ERROR))
2060
    (result, _) = self.jdm.CheckAndRegister(job, job_id,
2061
                                            [constants.JOB_STATUS_SUCCESS])
2062
    self.assertEqual(result, self.jdm.WRONGSTATUS)
2063
    self.assertFalse(self._status)
2064
    self.assertFalse(self._queue)
2065
    self.assertFalse(self.jdm.JobWaiting(job))
2066

    
2067
  def testCorrectStatus(self):
2068
    job = _IdOnlyFakeJob(24273)
2069
    job_id = str(23885)
2070

    
2071
    self._status.append((job_id, constants.JOB_STATUS_QUEUED))
2072
    (result, _) = self.jdm.CheckAndRegister(job, job_id,
2073
                                            [constants.JOB_STATUS_SUCCESS])
2074
    self.assertEqual(result, self.jdm.WAIT)
2075
    self.assertFalse(self._status)
2076
    self.assertFalse(self._queue)
2077
    self.assertTrue(self.jdm.JobWaiting(job))
2078
    self.assertEqual(self.jdm._waiters, {
2079
      job_id: set([job]),
2080
      })
2081

    
2082
    self._status.append((job_id, constants.JOB_STATUS_SUCCESS))
2083
    (result, _) = self.jdm.CheckAndRegister(job, job_id,
2084
                                            [constants.JOB_STATUS_SUCCESS])
2085
    self.assertEqual(result, self.jdm.CONTINUE)
2086
    self.assertFalse(self._status)
2087
    self.assertFalse(self._queue)
2088
    self.assertFalse(self.jdm.JobWaiting(job))
2089

    
2090
  def testFinalizedRightAway(self):
2091
    job = _IdOnlyFakeJob(224)
2092
    job_id = str(3081)
2093

    
2094
    self._status.append((job_id, constants.JOB_STATUS_SUCCESS))
2095
    (result, _) = self.jdm.CheckAndRegister(job, job_id,
2096
                                            [constants.JOB_STATUS_SUCCESS])
2097
    self.assertEqual(result, self.jdm.CONTINUE)
2098
    self.assertFalse(self._status)
2099
    self.assertFalse(self._queue)
2100
    self.assertFalse(self.jdm.JobWaiting(job))
2101
    self.assertEqual(self.jdm._waiters, {
2102
      job_id: set(),
2103
      })
2104

    
2105
    # Force cleanup
2106
    self.jdm.NotifyWaiters("0")
2107
    self.assertFalse(self.jdm._waiters)
2108
    self.assertFalse(self._status)
2109
    self.assertFalse(self._queue)
2110

    
2111
  def testMultipleWaiting(self):
2112
    # Use a deterministic random generator
2113
    rnd = random.Random(21402)
2114

    
2115
    job_ids = map(str, rnd.sample(range(1, 10000), 150))
2116

    
2117
    waiters = dict((job_ids.pop(),
2118
                    set(map(_IdOnlyFakeJob,
2119
                            [job_ids.pop()
2120
                             for _ in range(rnd.randint(1, 20))])))
2121
                   for _ in range(10))
2122

    
2123
    # Ensure there are no duplicate job IDs
2124
    assert not utils.FindDuplicates(waiters.keys() +
2125
                                    [job.id
2126
                                     for jobs in waiters.values()
2127
                                     for job in jobs])
2128

    
2129
    # Register all jobs as waiters
2130
    for job_id, job in [(job_id, job)
2131
                        for (job_id, jobs) in waiters.items()
2132
                        for job in jobs]:
2133
      self._status.append((job_id, constants.JOB_STATUS_QUEUED))
2134
      (result, _) = self.jdm.CheckAndRegister(job, job_id,
2135
                                              [constants.JOB_STATUS_SUCCESS])
2136
      self.assertEqual(result, self.jdm.WAIT)
2137
      self.assertFalse(self._status)
2138
      self.assertFalse(self._queue)
2139
      self.assertTrue(self.jdm.JobWaiting(job))
2140

    
2141
    self.assertEqual(self.jdm._waiters, waiters)
2142

    
2143
    def _MakeSet((name, mode, owner_names, pending)):
2144
      return (name, mode, owner_names,
2145
              [(pendmode, set(pend)) for (pendmode, pend) in pending])
2146

    
2147
    def _CheckLockInfo():
2148
      info = self.jdm.GetLockInfo([query.LQ_PENDING])
2149
      self.assertEqual(sorted(map(_MakeSet, info)), sorted([
2150
        ("job/%s" % job_id, None, None,
2151
         [("job", set([job.id for job in jobs]))])
2152
        for job_id, jobs in waiters.items()
2153
        if jobs
2154
        ]))
2155

    
2156
    _CheckLockInfo()
2157

    
2158
    # Notify in random order
2159
    for job_id in rnd.sample(waiters, len(waiters)):
2160
      # Remove from pending waiter list
2161
      jobs = waiters.pop(job_id)
2162
      for job in jobs:
2163
        self._status.append((job_id, constants.JOB_STATUS_SUCCESS))
2164
        (result, _) = self.jdm.CheckAndRegister(job, job_id,
2165
                                                [constants.JOB_STATUS_SUCCESS])
2166
        self.assertEqual(result, self.jdm.CONTINUE)
2167
        self.assertFalse(self._status)
2168
        self.assertFalse(self._queue)
2169
        self.assertFalse(self.jdm.JobWaiting(job))
2170

    
2171
      _CheckLockInfo()
2172

    
2173
    self.assertFalse(self.jdm.GetLockInfo([query.LQ_PENDING]))
2174

    
2175
    assert not waiters
2176

    
2177
  def testSelfDependency(self):
2178
    job = _IdOnlyFakeJob(18937)
2179

    
2180
    self._status.append((job.id, constants.JOB_STATUS_SUCCESS))
2181
    (result, _) = self.jdm.CheckAndRegister(job, job.id, [])
2182
    self.assertEqual(result, self.jdm.ERROR)
2183

    
2184
  def testJobDisappears(self):
2185
    job = _IdOnlyFakeJob(30540)
2186
    job_id = str(23769)
2187

    
2188
    def _FakeStatus(_):
2189
      raise errors.JobLost("#msg#")
2190

    
2191
    jdm = jqueue._JobDependencyManager(_FakeStatus, None)
2192
    (result, _) = jdm.CheckAndRegister(job, job_id, [])
2193
    self.assertEqual(result, self.jdm.ERROR)
2194
    self.assertFalse(jdm.JobWaiting(job))
2195
    self.assertFalse(jdm.GetLockInfo([query.LQ_PENDING]))
2196

    
2197

    
2198
if __name__ == "__main__":
2199
  testutils.GanetiTestProgram()