~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/script.py

  • Committer: Canonical.com Patch Queue Manager
  • Date: 2009-05-07 00:59:42 UTC
  • mfrom: (4340.1.1 bzr.dev)
  • Revision ID: pqm@pqm.ubuntu.com-20090507005942-wnvvomd3130f1vph
(Jelmer) Mention --force in the error message when unable to access
        in the master branch in 'bzr switch'.

Show diffs side-by-side

added added

removed removed

Lines of Context:
1
 
# Copyright (C) 2009, 2010 Canonical Ltd
2
 
#
3
 
# This program is free software; you can redistribute it and/or modify
4
 
# it under the terms of the GNU General Public License as published by
5
 
# the Free Software Foundation; either version 2 of the License, or
6
 
# (at your option) any later version.
7
 
#
8
 
# This program is distributed in the hope that it will be useful,
9
 
# but WITHOUT ANY WARRANTY; without even the implied warranty of
10
 
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
11
 
# GNU General Public License for more details.
12
 
#
13
 
# You should have received a copy of the GNU General Public License
14
 
# along with this program; if not, write to the Free Software
15
 
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA
16
 
 
17
 
"""Shell-like test scripts.
18
 
 
19
 
See developers/testing.html for more explanations.
20
 
"""
21
 
 
22
 
import doctest
23
 
import errno
24
 
import glob
25
 
import os
26
 
import shlex
27
 
import textwrap
28
 
from cStringIO import StringIO
29
 
 
30
 
from bzrlib import (
31
 
    osutils,
32
 
    tests,
33
 
    )
34
 
 
35
 
 
36
 
def split(s):
37
 
    """Split a command line respecting quotes."""
38
 
    scanner = shlex.shlex(s)
39
 
    scanner.quotes = '\'"`'
40
 
    scanner.whitespace_split = True
41
 
    for t in list(scanner):
42
 
        yield t
43
 
 
44
 
 
45
 
def _script_to_commands(text, file_name=None):
46
 
    """Turn a script into a list of commands with their associated IOs.
47
 
 
48
 
    Each command appears on a line by itself starting with '$ '. It can be
49
 
    associated with an input that will feed it and an expected output.
50
 
 
51
 
    Comments starts with '#' until the end of line.
52
 
    Empty lines are ignored.
53
 
 
54
 
    Input and output are full lines terminated by a '\n'.
55
 
 
56
 
    Input lines start with '<'.
57
 
    Output lines start with nothing.
58
 
    Error lines start with '2>'.
59
 
    """
60
 
 
61
 
    commands = []
62
 
 
63
 
    def add_command(cmd, input, output, error):
64
 
        if cmd is not None:
65
 
            if input is not None:
66
 
                input = ''.join(input)
67
 
            if output is not None:
68
 
                output = ''.join(output)
69
 
            if error is not None:
70
 
                error = ''.join(error)
71
 
            commands.append((cmd, input, output, error))
72
 
 
73
 
    cmd_cur = None
74
 
    cmd_line = 1
75
 
    lineno = 0
76
 
    input, output, error = None, None, None
77
 
    text = textwrap.dedent(text)
78
 
    for line in text.split('\n'):
79
 
        lineno += 1
80
 
        # Keep a copy for error reporting
81
 
        orig = line
82
 
        comment =  line.find('#')
83
 
        if comment >= 0:
84
 
            # Delete comments
85
 
            line = line[0:comment]
86
 
            line = line.rstrip()
87
 
        if line == '':
88
 
            # Ignore empty lines
89
 
            continue
90
 
        if line.startswith('$'):
91
 
            # Time to output the current command
92
 
            add_command(cmd_cur, input, output, error)
93
 
            # And start a new one
94
 
            cmd_cur = list(split(line[1:]))
95
 
            cmd_line = lineno
96
 
            input, output, error = None, None, None
97
 
        elif line.startswith('<'):
98
 
            if input is None:
99
 
                if cmd_cur is None:
100
 
                    raise SyntaxError('No command for that input',
101
 
                                      (file_name, lineno, 1, orig))
102
 
                input = []
103
 
            input.append(line[1:] + '\n')
104
 
        elif line.startswith('2>'):
105
 
            if error is None:
106
 
                if cmd_cur is None:
107
 
                    raise SyntaxError('No command for that error',
108
 
                                      (file_name, lineno, 1, orig))
109
 
                error = []
110
 
            error.append(line[2:] + '\n')
111
 
        else:
112
 
            # can happen if the first line is not recognized as a command, eg
113
 
            # if the prompt has leading whitespace
114
 
            if output is None:
115
 
                if cmd_cur is None:
116
 
                    raise SyntaxError('No command for line %r' % (line,),
117
 
                                      (file_name, lineno, 1, orig))
118
 
                output = []
119
 
            output.append(line + '\n')
120
 
    # Add the last seen command
121
 
    add_command(cmd_cur, input, output, error)
122
 
    return commands
123
 
 
124
 
 
125
 
def _scan_redirection_options(args):
126
 
    """Recognize and process input and output redirections.
127
 
 
128
 
    :param args: The command line arguments
129
 
 
130
 
    :return: A tuple containing: 
131
 
        - The file name redirected from or None
132
 
        - The file name redirected to or None
133
 
        - The mode to open the output file or None
134
 
        - The reamining arguments
135
 
    """
136
 
    def redirected_file_name(direction, name, args):
137
 
        if name == '':
138
 
            try:
139
 
                name = args.pop(0)
140
 
            except IndexError:
141
 
                # We leave the error handling to higher levels, an empty name
142
 
                # can't be legal.
143
 
                name = ''
144
 
        return name
145
 
 
146
 
    remaining = []
147
 
    in_name = None
148
 
    out_name, out_mode = None, None
149
 
    while args:
150
 
        arg = args.pop(0)
151
 
        if arg.startswith('<'):
152
 
            in_name = redirected_file_name('<', arg[1:], args)
153
 
        elif arg.startswith('>>'):
154
 
            out_name = redirected_file_name('>>', arg[2:], args)
155
 
            out_mode = 'ab+'
156
 
        elif arg.startswith('>',):
157
 
            out_name = redirected_file_name('>', arg[1:], args)
158
 
            out_mode = 'wb+'
159
 
        else:
160
 
            remaining.append(arg)
161
 
    return in_name, out_name, out_mode, remaining
162
 
 
163
 
 
164
 
class ScriptRunner(object):
165
 
    """Run a shell-like script from a test.
166
 
    
167
 
    Can be used as:
168
 
 
169
 
    from bzrlib.tests import script
170
 
 
171
 
    ...
172
 
 
173
 
        def test_bug_nnnnn(self):
174
 
            sr = script.ScriptRunner()
175
 
            sr.run_script(self, '''
176
 
            $ bzr init
177
 
            $ bzr do-this
178
 
            # Boom, error
179
 
            ''')
180
 
    """
181
 
 
182
 
    def __init__(self):
183
 
        self.output_checker = doctest.OutputChecker()
184
 
        self.check_options = doctest.ELLIPSIS
185
 
 
186
 
    def run_script(self, test_case, text):
187
 
        """Run a shell-like script as a test.
188
 
 
189
 
        :param test_case: A TestCase instance that should provide the fail(),
190
 
            assertEqualDiff and _run_bzr_core() methods as well as a 'test_dir'
191
 
            attribute used as a jail root.
192
 
 
193
 
        :param text: A shell-like script (see _script_to_commands for syntax).
194
 
        """
195
 
        for cmd, input, output, error in _script_to_commands(text):
196
 
            self.run_command(test_case, cmd, input, output, error)
197
 
 
198
 
    def run_command(self, test_case, cmd, input, output, error):
199
 
        mname = 'do_' + cmd[0]
200
 
        method = getattr(self, mname, None)
201
 
        if method is None:
202
 
            raise SyntaxError('Command not found "%s"' % (cmd[0],),
203
 
                              None, 1, ' '.join(cmd))
204
 
        if input is None:
205
 
            str_input = ''
206
 
        else:
207
 
            str_input = ''.join(input)
208
 
        args = list(self._pre_process_args(cmd[1:]))
209
 
        retcode, actual_output, actual_error = method(test_case,
210
 
                                                      str_input, args)
211
 
 
212
 
        self._check_output(output, actual_output, test_case)
213
 
        self._check_output(error, actual_error, test_case)
214
 
        if retcode and not error and actual_error:
215
 
            test_case.fail('In \n\t%s\nUnexpected error: %s'
216
 
                           % (' '.join(cmd), actual_error))
217
 
        return retcode, actual_output, actual_error
218
 
 
219
 
    def _check_output(self, expected, actual, test_case):
220
 
        if expected is None:
221
 
            # Specifying None means: any output is accepted
222
 
            return
223
 
        if actual is None:
224
 
            test_case.fail('We expected output: %r, but found None'
225
 
                           % (expected,))
226
 
        matching = self.output_checker.check_output(
227
 
            expected, actual, self.check_options)
228
 
        if not matching:
229
 
            # Note that we can't use output_checker.output_difference() here
230
 
            # because... the API is broken ('expected' must be a doctest
231
 
            # specific object of which a 'want' attribute will be our
232
 
            # 'expected' parameter. So we just fallback to our good old
233
 
            # assertEqualDiff since we know there *are* differences and the
234
 
            # output should be decently readable.
235
 
            test_case.assertEqualDiff(expected, actual)
236
 
 
237
 
    def _pre_process_args(self, args):
238
 
        new_args = []
239
 
        for arg in args:
240
 
            # Strip the simple and double quotes since we don't care about
241
 
            # them.  We leave the backquotes in place though since they have a
242
 
            # different semantic.
243
 
            if arg[0] in  ('"', "'") and arg[0] == arg[-1]:
244
 
                yield arg[1:-1]
245
 
            else:
246
 
                if glob.has_magic(arg):
247
 
                    matches = glob.glob(arg)
248
 
                    if matches:
249
 
                        # We care more about order stability than performance
250
 
                        # here
251
 
                        matches.sort()
252
 
                        for m in matches:
253
 
                            yield m
254
 
                else:
255
 
                    yield arg
256
 
 
257
 
    def _read_input(self, input, in_name):
258
 
        if in_name is not None:
259
 
            infile = open(in_name, 'rb')
260
 
            try:
261
 
                # Command redirection takes precedence over provided input
262
 
                input = infile.read()
263
 
            finally:
264
 
                infile.close()
265
 
        return input
266
 
 
267
 
    def _write_output(self, output, out_name, out_mode):
268
 
        if out_name is not None:
269
 
            outfile = open(out_name, out_mode)
270
 
            try:
271
 
                outfile.write(output)
272
 
            finally:
273
 
                outfile.close()
274
 
            output = None
275
 
        return output
276
 
 
277
 
    def do_bzr(self, test_case, input, args):
278
 
        retcode, out, err = test_case._run_bzr_core(
279
 
            args, retcode=None, encoding=None, stdin=input, working_dir=None)
280
 
        return retcode, out, err
281
 
 
282
 
    def do_cat(self, test_case, input, args):
283
 
        (in_name, out_name, out_mode, args) = _scan_redirection_options(args)
284
 
        if args and in_name is not None:
285
 
            raise SyntaxError('Specify a file OR use redirection')
286
 
 
287
 
        inputs = []
288
 
        if input:
289
 
            inputs.append(input)
290
 
        input_names = args
291
 
        if in_name:
292
 
            args.append(in_name)
293
 
        for in_name in input_names:
294
 
            try:
295
 
                inputs.append(self._read_input(None, in_name))
296
 
            except IOError, e:
297
 
                # Some filenames are illegal on Windows and generate EINVAL
298
 
                # rather than just saying the filename doesn't exist
299
 
                if e.errno in (errno.ENOENT, errno.EINVAL):
300
 
                    return (1, None,
301
 
                            '%s: No such file or directory\n' % (in_name,))
302
 
                raise
303
 
        # Basically cat copy input to output
304
 
        output = ''.join(inputs)
305
 
        # Handle output redirections
306
 
        try:
307
 
            output = self._write_output(output, out_name, out_mode)
308
 
        except IOError, e:
309
 
            # If out_name cannot be created, we may get 'ENOENT', however if
310
 
            # out_name is something like '', we can get EINVAL
311
 
            if e.errno in (errno.ENOENT, errno.EINVAL):
312
 
                return 1, None, '%s: No such file or directory\n' % (out_name,)
313
 
            raise
314
 
        return 0, output, None
315
 
 
316
 
    def do_echo(self, test_case, input, args):
317
 
        (in_name, out_name, out_mode, args) = _scan_redirection_options(args)
318
 
        if input or in_name:
319
 
            raise SyntaxError('echo doesn\'t read from stdin')
320
 
        if args:
321
 
            input = ' '.join(args)
322
 
        # Always append a \n'
323
 
        input += '\n'
324
 
        # Process output
325
 
        output = input
326
 
        # Handle output redirections
327
 
        try:
328
 
            output = self._write_output(output, out_name, out_mode)
329
 
        except IOError, e:
330
 
            if e.errno in (errno.ENOENT, errno.EINVAL):
331
 
                return 1, None, '%s: No such file or directory\n' % (out_name,)
332
 
            raise
333
 
        return 0, output, None
334
 
 
335
 
    def _get_jail_root(self, test_case):
336
 
        return test_case.test_dir
337
 
 
338
 
    def _ensure_in_jail(self, test_case, path):
339
 
        jail_root = self._get_jail_root(test_case)
340
 
        if not osutils.is_inside(jail_root, osutils.normalizepath(path)):
341
 
            raise ValueError('%s is not inside %s' % (path, jail_root))
342
 
 
343
 
    def do_cd(self, test_case, input, args):
344
 
        if len(args) > 1:
345
 
            raise SyntaxError('Usage: cd [dir]')
346
 
        if len(args) == 1:
347
 
            d = args[0]
348
 
            self._ensure_in_jail(test_case, d)
349
 
        else:
350
 
            # The test "home" directory is the root of its jail
351
 
            d = self._get_jail_root(test_case)
352
 
        os.chdir(d)
353
 
        return 0, None, None
354
 
 
355
 
    def do_mkdir(self, test_case, input, args):
356
 
        if not args or len(args) != 1:
357
 
            raise SyntaxError('Usage: mkdir dir')
358
 
        d = args[0]
359
 
        self._ensure_in_jail(test_case, d)
360
 
        os.mkdir(d)
361
 
        return 0, None, None
362
 
 
363
 
    def do_rm(self, test_case, input, args):
364
 
        err = None
365
 
 
366
 
        def error(msg, path):
367
 
            return  "rm: cannot remove '%s': %s\n" % (path, msg)
368
 
 
369
 
        force, recursive = False, False
370
 
        opts = None
371
 
        if args and args[0][0] == '-':
372
 
            opts = args.pop(0)[1:]
373
 
            if 'f' in opts:
374
 
                force = True
375
 
                opts = opts.replace('f', '', 1)
376
 
            if 'r' in opts:
377
 
                recursive = True
378
 
                opts = opts.replace('r', '', 1)
379
 
        if not args or opts:
380
 
            raise SyntaxError('Usage: rm [-fr] path+')
381
 
        for p in args:
382
 
            self._ensure_in_jail(test_case, p)
383
 
            # FIXME: Should we put that in osutils ?
384
 
            try:
385
 
                os.remove(p)
386
 
            except OSError, e:
387
 
                # Various OSes raises different exceptions (linux: EISDIR,
388
 
                #   win32: EACCES, OSX: EPERM) when invoked on a directory
389
 
                if e.errno in (errno.EISDIR, errno.EPERM, errno.EACCES):
390
 
                    if recursive:
391
 
                        osutils.rmtree(p)
392
 
                    else:
393
 
                        err = error('Is a directory', p)
394
 
                        break
395
 
                elif e.errno == errno.ENOENT:
396
 
                    if not force:
397
 
                        err =  error('No such file or directory', p)
398
 
                        break
399
 
                else:
400
 
                    raise
401
 
        if err:
402
 
            retcode = 1
403
 
        else:
404
 
            retcode = 0
405
 
        return retcode, None, err
406
 
 
407
 
    def do_mv(self, test_case, input, args):
408
 
        err = None
409
 
        def error(msg, src, dst):
410
 
            return "mv: cannot move %s to %s: %s\n" % (src, dst, msg)
411
 
 
412
 
        if not args or len(args) != 2:
413
 
            raise SyntaxError("Usage: mv path1 path2")
414
 
        src, dst = args
415
 
        try:
416
 
            real_dst = dst
417
 
            if os.path.isdir(dst):
418
 
                real_dst = os.path.join(dst, os.path.basename(src))
419
 
            os.rename(src, real_dst)
420
 
        except OSError, e:
421
 
            if e.errno == errno.ENOENT:
422
 
                err = error('No such file or directory', src, dst)
423
 
            else:
424
 
                raise
425
 
        if err:
426
 
            retcode = 1
427
 
        else:
428
 
            retcode = 0
429
 
        return retcode, None, err
430
 
 
431
 
 
432
 
 
433
 
class TestCaseWithMemoryTransportAndScript(tests.TestCaseWithMemoryTransport):
434
 
    """Helper class to experiment shell-like test and memory fs.
435
 
 
436
 
    This not intended to be used outside of experiments in implementing memoy
437
 
    based file systems and evolving bzr so that test can use only memory based
438
 
    resources.
439
 
    """
440
 
 
441
 
    def setUp(self):
442
 
        super(TestCaseWithMemoryTransportAndScript, self).setUp()
443
 
        self.script_runner = ScriptRunner()
444
 
 
445
 
    def run_script(self, script):
446
 
        return self.script_runner.run_script(self, script)
447
 
 
448
 
    def run_command(self, cmd, input, output, error):
449
 
        return self.script_runner.run_command(self, cmd, input, output, error)
450
 
 
451
 
 
452
 
class TestCaseWithTransportAndScript(tests.TestCaseWithTransport):
453
 
    """Helper class to quickly define shell-like tests.
454
 
 
455
 
    Can be used as:
456
 
 
457
 
    from bzrlib.tests import script
458
 
 
459
 
 
460
 
    class TestBug(script.TestCaseWithTransportAndScript):
461
 
 
462
 
        def test_bug_nnnnn(self):
463
 
            self.run_script('''
464
 
            $ bzr init
465
 
            $ bzr do-this
466
 
            # Boom, error
467
 
            ''')
468
 
    """
469
 
 
470
 
    def setUp(self):
471
 
        super(TestCaseWithTransportAndScript, self).setUp()
472
 
        self.script_runner = ScriptRunner()
473
 
 
474
 
    def run_script(self, script):
475
 
        return self.script_runner.run_script(self, script)
476
 
 
477
 
    def run_command(self, cmd, input, output, error):
478
 
        return self.script_runner.run_command(self, cmd, input, output, error)
479
 
 
480
 
 
481
 
def run_script(test_case, script_string):
482
 
    """Run the given script within a testcase"""
483
 
    return ScriptRunner().run_script(test_case, script_string)