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
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
|
# This file is part of Buildbot. Buildbot is free software: you can
# redistribute it and/or modify it under the terms of the GNU General Public
# License as published by the Free Software Foundation, version 2.
#
# This program is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS
# FOR A PARTICULAR PURPOSE. See the GNU General Public License for more
# details.
#
# You should have received a copy of the GNU General Public License along with
# this program; if not, write to the Free Software Foundation, Inc., 51
# Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
#
# Copyright Buildbot Team Members
from twisted.trial import unittest
from twisted.internet import defer
from buildslave.test.util.command import CommandTestMixin
from buildslave.commands.base import Command
# set up a fake Command subclass to test the handling in Command. Think of
# this as testing Command's subclassability.
class DummyCommand(Command):
def setup(self, args):
self.setup_done = True
self.interrupted = False
self.started = False
def start(self):
self.started = True
self.sendStatus(self.args)
self.cmd_deferred = defer.Deferred()
return self.cmd_deferred
def interrupt(self):
self.interrupted = True
self.finishCommand()
def finishCommand(self):
d = self.cmd_deferred
self.cmd_deferred = None
d.callback(None)
def failCommand(self):
d = self.cmd_deferred
self.cmd_deferred = None
d.errback(RuntimeError("forced failure"))
class TestDummyCommand(CommandTestMixin, unittest.TestCase):
def setUp(self):
self.setUpCommand()
def tearDown(self):
self.tearDownCommand()
def assertState(self, setup_done, running, started, interrupted, msg=None):
self.assertEqual(
{
'setup_done' : self.cmd.setup_done,
'running' : self.cmd.running,
'started' : self.cmd.started,
'interrupted' : self.cmd.interrupted,
}, {
'setup_done' : setup_done,
'running' : running,
'started' : started,
'interrupted' : interrupted,
}, msg)
def test_run(self):
cmd = self.make_command(DummyCommand, { 'stdout' : 'yay' })
self.assertState(True, False, False, False, "setup called by constructor")
# start the command
d = self.run_command()
self.assertState(True, True, True, False, "started and running both set")
# allow the command to finish and check the result
cmd.finishCommand()
def check(_):
self.assertState(True, False, True, False, "started and not running when done")
d.addCallback(check)
def checkresult(_):
self.assertUpdates([ { 'stdout' : 'yay' } ], "updates processed")
d.addCallback(checkresult)
return d
def test_run_failure(self):
cmd = self.make_command(DummyCommand, {})
self.assertState(True, False, False, False, "setup called by constructor")
# start the command
d = self.run_command()
self.assertState(True, True, True, False, "started and running both set")
# fail the command with an exception, and check the result
cmd.failCommand()
def check(_):
self.assertState(True, False, True, False, "started and not running when done")
d.addErrback(check)
def checkresult(_):
self.assertUpdates([ {} ], "updates processed")
d.addCallback(checkresult)
return d
def test_run_interrupt(self):
cmd = self.make_command(DummyCommand, {})
self.assertState(True, False, False, False, "setup called by constructor")
# start the command
d = self.run_command()
self.assertState(True, True, True, False, "started and running both set")
# interrupt the command
cmd.doInterrupt()
self.assertTrue(cmd.interrupted)
def check(_):
self.assertState(True, False, True, True, "finishes with interrupted set")
d.addCallback(check)
return d
|
