← Back to branch summary

~bzr-pqm/bzr/bzr.dev

« back to all changes in this revision

Viewing changes to bzrlib/tests/script.py

  • Committer: Jelmer Vernooij
  • Date: 2010-08-29 14:37:51 UTC
  • mto: This revision was merged to the branch mainline in revision 5418.
  • Revision ID: jelmer@samba.org-20100829143751-9ry91e6u887gswiz
Move some bzrdir-specific tests to bzrlib.tests.per_bzrdir.

Show diffs side-by-side

added added

removed removed

Lines of Context:
bzrlib/tests/script.py
 
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)