1
# arch-tag: 4b63b80e-9add-46e1-941f-50556f7af04a
2
# Copyright (C) 2004 Canonical Ltd.
3
# Author: David Allouche <david@canonical.com>
5
# This program is free software; you can redistribute it and/or modify
6
# it under the terms of the GNU General Public License as published by
7
# the Free Software Foundation; either version 2 of the License, or
8
# (at your option) any later version.
10
# This program is distributed in the hope that it will be useful,
11
# but WITHOUT ANY WARRANTY; without even the implied warranty of
12
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13
# GNU General Public License for more details.
15
# You should have received a copy of the GNU General Public License
16
# along with this program; if not, write to the Free Software
17
# Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111-1307 USA
19
"""Test Twisted support
24
from twisted.trial import unittest
25
from framework_twisted import run_in_thread
28
def make_test_dir(testcase, subdir):
29
path = os.path.join(os.getcwd(), testcase.__class__.__name__)
30
if os.path.exists(path): shutil.rmtree(path)
34
def make_spawn(command):
35
from arch.backends import knotted, logger
36
return knotted.TwistedSpawningStrategy(command, logger.Logger())
39
class TwistedStatusCmd(unittest.TestCase):
41
def test_null_cmd(self):
42
"""TwistedSpawningStrategy.null_cmd works"""
43
run_in_thread(self._null_cmd)
46
spawn = make_spawn('touch')
47
path = make_test_dir(self, 'null_cmd')
48
foo = os.path.join(path, 'foo')
49
spawn.status_cmd((foo,), None, (0,))
50
self.assert_(os.path.isfile(foo))
52
def test_exec_problem(self):
53
"""TwistedSpawningStrategy.null_cmd raises ExecProblem"""
54
run_in_thread(self._exec_problem)
56
def _exec_problem(self):
57
from arch import errors
58
spawn = make_spawn('false')
59
def thunk(): spawn.status_cmd((), None, (0,))
61
self.assertRaises(errors.ExecProblem, thunk)
64
self.fail('did not raise')
65
except errors.ExecProblem, e:
66
self.assertEqual(1, e.proc.status)
67
self.assertEqual(None, e.proc.signal)
70
"""TwistedSpawningStrategy.null_cmd records error output"""
71
run_in_thread(self._error)
74
path = make_test_dir(self, 'error')
75
name = os.path.join(path, 'printerr')
76
script = open(name, 'w')
77
print >> script, "echo -n first line >&2"
78
print >> script, "sleep 1"
79
print >> script, "echo >&2"
80
print >> script, "echo second line >&2"
81
print >> script, "exit 1"
83
from arch import errors
84
spawn = make_spawn('sh')
86
spawn.status_cmd((name,), None, (0,))
87
self.fail('did not raise')
88
except errors.ExecProblem, e:
89
expected = '\n'.join(["first line", "second line", ""])
90
self.assertEqual(expected, e.proc.error)
92
def test_status(self):
93
"""TwistedSpawningStrategy.status_cmd works"""
94
run_in_thread(self._status)
97
from arch import errors
98
spawn = make_spawn('sh')
99
status = spawn.status_cmd(('-c', 'exit 0'), None, expected=(0,1))
100
self.assertEqual(0, status)
101
status = spawn.status_cmd(('-c', 'exit 1'), None, expected=(0,1))
102
self.assertEqual(1, status)
103
def thunk(): spawn.status_cmd(('-c', 'exit 2'), None, expected=(0,1))
104
self.assertRaises(errors.ExecProblem, thunk)
106
def test_chdir(self):
107
"""TwistedSpawningStrategy.status_cmd does chdir"""
108
run_in_thread(self._chdir)
111
spawn = make_spawn('touch')
112
path = make_test_dir(self, 'chdir')
113
status = spawn.status_cmd(('foo',), path, expected=(0,))
114
self.assertEqual(0, status)
115
name = os.path.join(path, 'foo')
116
self.assert_(os.path.isfile(name))
121
class TwistedSequenceCmd(unittest.TestCase):
123
def test_sequence(self):
124
"""TwistedSpawningStrategy.sequence_cmd works"""
125
run_in_thread(self._sequence)
128
spawn = make_spawn('sh')
129
args = ('-c', "echo aaa ; echo bbb ; exit 1")
130
cmd = spawn.sequence_cmd(args, None, (0, 1))
131
self.assertEqual(False, cmd.finished)
132
thunk = lambda : cmd.status
133
self.assertRaises(AttributeError, thunk)
135
self.assertEqual(['aaa', 'bbb'], result)
136
self.assertEqual(True, cmd.finished)
137
self.assertEqual(1, cmd.status)
139
def test_failure(self):
140
"""TwistedSpawningStrategy.sequence_cmd raises ExecProblem"""
141
run_in_thread(self._failure)
144
from arch import errors
145
spawn = make_spawn('sh')
146
script = "echo aaa ; echo bbb >&2 ; exit 1 ;"
147
cmd = spawn.sequence_cmd(('-c', script), None, (0,))
149
self.assertEqual("aaa", result)
152
self.fail("did not raise")
153
except errors.ExecProblem, e:
154
self.assertEqual(1, e.proc.status)
155
self.assertEqual(None, e.proc.signal)
156
self.assertEqual("bbb\n", e.proc.error)
158
def test_chdir(self):
159
"""TwistedSpawningStrategy.sequence_cmd does chdir"""
160
run_in_thread(self._chdir)
163
spawn = make_spawn('pwd')
164
path = make_test_dir(self, 'pwd')
165
seq = spawn.sequence_cmd((), path, expected=(0,))
167
self.assertEqual([path], lines)
170
class TwistedStatusTextCmd(unittest.TestCase):
172
def test_status_text(self):
173
"""TwistedSpawningStrategy.status_text_cmd works"""
174
run_in_thread(self._status_text)
176
def _status_text(self):
177
spawn = make_spawn('sh')
178
args = ('-c', "echo aaa ; echo bbb ; exit 1")
179
(status, text) = spawn.status_text_cmd(args, None, (1,))
180
self.assertEqual(1, status)
181
self.assertEqual("aaa\nbbb\n", text)
183
def test_failure(self):
184
"""TwistedSpawningStrategy.status_text_cmd raises ExecProblem"""
185
run_in_thread(self._failure)
188
from arch import errors
189
spawn = make_spawn('sh')
190
args = ('-c', "echo aaa ; echo bbb >&2 ; exit 2")
192
status, text = spawn.status_text_cmd(args, None, (1,))
193
self.fail("did not raise")
194
except errors.ExecProblem, e:
195
self.assertEqual(2, e.proc.status)
196
self.assertEqual(None, e.proc.signal)
197
self.assertEqual("bbb\n", e.proc.error)
199
def test_chdir(self):
200
"""TwistedSpawningStrategy.text_cmd does chdir"""
201
run_in_thread(self._chdir)
204
spawn = make_spawn('pwd')
205
path = make_test_dir(self, 'pwd')
206
status, text = spawn.status_text_cmd((), path, expected=(0,))
207
self.assertEqual(0, status)
208
self.assertEqual(path+'\n', text)
added
removed
fai/arch/tests/test_twisted.py
