1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
|
# Copyright 2009 Canonical Ltd. This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).
"""Unit tests for methods of CodeImportMachine.
Other tests are in codeimport-machine.txt."""
from zope.component import getUtility
from lp.code.enums import (
CodeImportMachineOfflineReason,
CodeImportMachineState,
)
from lp.code.interfaces.codeimportjob import ICodeImportJobWorkflow
from lp.services.database.constants import UTC_NOW
from lp.testing import TestCaseWithFactory
from lp.testing.layers import DatabaseFunctionalLayer
class TestCodeImportMachineShouldLookForJob(TestCaseWithFactory):
"""Tests for `CodeImportMachine.shouldLookForJob`."""
layer = DatabaseFunctionalLayer
def setUp(self):
super(TestCodeImportMachineShouldLookForJob, self).setUp(
'admin@canonical.com')
self.machine = self.factory.makeCodeImportMachine(set_online=True)
def createJobRunningOnMachine(self, machine):
"""Create a job in the database and mark it as running on `machine`.
"""
job = self.factory.makeCodeImportJob()
getUtility(ICodeImportJobWorkflow).startJob(job, machine)
def test_machineIsOffline(self):
# When the machine is offline, we shouldn't look for any jobs.
self.machine.setOffline(CodeImportMachineOfflineReason.STOPPED)
self.assertFalse(self.machine.shouldLookForJob(10))
def test_machineIsQuiescingNoJobsRunning(self):
# When the machine is quiescing and no jobs are running on this
# machine, we should set the machine to OFFLINE and not look for jobs.
self.machine.setQuiescing(self.factory.makePerson())
self.assertFalse(self.machine.shouldLookForJob(10))
self.assertEqual(self.machine.state, CodeImportMachineState.OFFLINE)
def test_machineIsQuiescingWithJobsRunning(self):
# When the machine is quiescing and there are jobs running on this
# machine, we shouldn't look for any more jobs.
self.createJobRunningOnMachine(self.machine)
self.machine.setQuiescing(self.factory.makePerson())
self.assertFalse(self.machine.shouldLookForJob(10))
self.assertEqual(self.machine.state, CodeImportMachineState.QUIESCING)
def test_enoughJobsRunningOnMachine(self):
# When there are already enough jobs running on this machine, we
# shouldn't look for any more jobs.
self.createJobRunningOnMachine(self.machine)
self.assertFalse(self.machine.shouldLookForJob(worker_limit=1))
def test_shouldLook(self):
# If the machine is online and there are not already
# max_jobs_per_machine jobs running, then we should look for a job.
self.assertTrue(self.machine.shouldLookForJob(worker_limit=1))
def test_noHeartbeatWhenCreated(self):
# Machines are created with a NULL heartbeat.
self.assertTrue(self.machine.heartbeat is None)
def test_noHeartbeatUpdateWhenOffline(self):
# When the machine is offline, the heartbeat is not updated.
self.machine.setOffline(CodeImportMachineOfflineReason.STOPPED)
self.machine.shouldLookForJob(10)
self.assertTrue(self.machine.heartbeat is None)
def test_heartbeatUpdateWhenQuiescing(self):
# When the machine is quiescing, the heartbeat is updated.
self.machine.setQuiescing(self.factory.makePerson())
self.machine.shouldLookForJob(10)
self.assertSqlAttributeEqualsDate(self.machine, 'heartbeat', UTC_NOW)
def test_heartbeatUpdateWhenOnline(self):
# When the machine is online, the heartbeat is updated.
self.machine.shouldLookForJob(10)
self.assertSqlAttributeEqualsDate(self.machine, 'heartbeat', UTC_NOW)
|