← Back to branch summary

~abentley/bzrtools/bzrtools.dev

« back to all changes in this revision

Viewing changes to fai/arch/tests/test_twisted.py

  • Committer: Robert Collins
  • Date: 2005-09-13 15:11:39 UTC
  • mto: (147.2.6) (364.1.3 bzrtools)
  • mto: This revision was merged to the branch mainline in revision 324.
  • Revision ID: robertc@robertcollins.net-20050913151139-9ac920fc9d7bda31
TODOification

Show diffs side-by-side

added added

removed removed

Lines of Context:
fai/arch/tests/test_twisted.py
 
1
# arch-tag: 4b63b80e-9add-46e1-941f-50556f7af04a
 
2
# Copyright (C) 2004 Canonical Ltd.
 
3
#               Author: David Allouche <david@canonical.com>
 
4
#
 
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.
 
9
#
 
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.
 
14
#
 
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
 
18
 
 
19
"""Test Twisted support
 
20
"""
 
21
 
 
22
import os
 
23
import shutil
 
24
from twisted.trial import unittest
 
25
from framework_twisted import run_in_thread
 
26
 
 
27
 
 
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)
 
31
    os.makedirs(path)
 
32
    return path
 
33
 
 
34
def make_spawn(command):
 
35
    from arch.backends import knotted, logger
 
36
    return knotted.TwistedSpawningStrategy(command, logger.Logger())
 
37
 
 
38
 
 
39
class TwistedStatusCmd(unittest.TestCase):
 
40
 
 
41
    def test_null_cmd(self):
 
42
        """TwistedSpawningStrategy.null_cmd works"""
 
43
        run_in_thread(self._null_cmd)
 
44
 
 
45
    def _null_cmd(self):
 
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))
 
51
 
 
52
    def test_exec_problem(self):
 
53
        """TwistedSpawningStrategy.null_cmd raises ExecProblem"""
 
54
        run_in_thread(self._exec_problem)
 
55
 
 
56
    def _exec_problem(self):
 
57
        from arch import errors
 
58
        spawn = make_spawn('false')
 
59
        def thunk(): spawn.status_cmd((), None, (0,))
 
60
        # thunk()
 
61
        self.assertRaises(errors.ExecProblem, thunk)
 
62
        try:
 
63
            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)
 
68
 
 
69
    def test_error(self):
 
70
        """TwistedSpawningStrategy.null_cmd records error output"""
 
71
        run_in_thread(self._error)
 
72
 
 
73
    def _error(self):
 
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"
 
82
        script.close()
 
83
        from arch import errors
 
84
        spawn = make_spawn('sh')
 
85
        try:
 
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)
 
91
 
 
92
    def test_status(self):
 
93
        """TwistedSpawningStrategy.status_cmd works"""
 
94
        run_in_thread(self._status)
 
95
 
 
96
    def _status(self):
 
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)
 
105
 
 
106
    def test_chdir(self):
 
107
        """TwistedSpawningStrategy.status_cmd does chdir"""
 
108
        run_in_thread(self._chdir)
 
109
 
 
110
    def _chdir(self):
 
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))
 
117
 
 
118
    # TODO: test signal
 
119
 
 
120
 
 
121
class TwistedSequenceCmd(unittest.TestCase):
 
122
 
 
123
    def test_sequence(self):
 
124
        """TwistedSpawningStrategy.sequence_cmd works"""
 
125
        run_in_thread(self._sequence)
 
126
 
 
127
    def _sequence(self):
 
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)
 
134
        result = list(cmd)
 
135
        self.assertEqual(['aaa', 'bbb'], result)
 
136
        self.assertEqual(True, cmd.finished)
 
137
        self.assertEqual(1, cmd.status)
 
138
 
 
139
    def test_failure(self):
 
140
        """TwistedSpawningStrategy.sequence_cmd raises ExecProblem"""
 
141
        run_in_thread(self._failure)
 
142
 
 
143
    def _failure(self):
 
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,))
 
148
        result = cmd.next()
 
149
        self.assertEqual("aaa", result)
 
150
        try:
 
151
            cmd.next()
 
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)
 
157
 
 
158
    def test_chdir(self):
 
159
        """TwistedSpawningStrategy.sequence_cmd does chdir"""
 
160
        run_in_thread(self._chdir)
 
161
 
 
162
    def _chdir(self):
 
163
        spawn = make_spawn('pwd')
 
164
        path = make_test_dir(self, 'pwd')
 
165
        seq = spawn.sequence_cmd((), path, expected=(0,))
 
166
        lines = list(seq)
 
167
        self.assertEqual([path], lines)
 
168
 
 
169
 
 
170
class TwistedStatusTextCmd(unittest.TestCase):
 
171
 
 
172
    def test_status_text(self):
 
173
        """TwistedSpawningStrategy.status_text_cmd works"""
 
174
        run_in_thread(self._status_text)
 
175
 
 
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)
 
182
 
 
183
    def test_failure(self):
 
184
        """TwistedSpawningStrategy.status_text_cmd raises ExecProblem"""
 
185
        run_in_thread(self._failure)
 
186
 
 
187
    def _failure(self):
 
188
        from arch import errors
 
189
        spawn = make_spawn('sh')
 
190
        args = ('-c', "echo aaa ; echo bbb >&2 ; exit 2")
 
191
        try:
 
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)
 
198
 
 
199
    def test_chdir(self):
 
200
        """TwistedSpawningStrategy.text_cmd does chdir"""
 
201
        run_in_thread(self._chdir)
 
202
 
 
203
    def _chdir(self):
 
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)