utils/lit/TclUtil.py

import itertools

from ShCommands import Command, Pipeline

def tcl_preprocess(data):
    # Tcl has a preprocessing step to replace escaped newlines.
    i = data.find('\\\n')
    if i == -1:
        return data

    # Replace '\\\n' and subsequent whitespace by a single space.
    n = len(data)
    str = data[:i]
    i += 2
    while i < n and data[i] in ' \t':
        i += 1
    return str + ' ' + data[i:]

class TclLexer:
    """TclLexer - Lex a string into "words", following the Tcl syntax."""

    def __init__(self, data):
        self.data = tcl_preprocess(data)
        self.pos = 0
        self.end = len(self.data)

    def at_end(self):
        return self.pos == self.end

    def eat(self):
        c = self.data[self.pos]
        self.pos += 1
        return c

    def look(self):
        return self.data[self.pos]

    def maybe_eat(self, c):
        """
        maybe_eat(c) - Consume the character c if it is the next character,
        returning True if a character was consumed. """
        if self.data[self.pos] == c:
            self.pos += 1
            return True
        return False

    def escape(self, c):
        if c == 'a':
            return '\x07'
        elif c == 'b':
            return '\x08'
        elif c == 'f':
            return '\x0c'
        elif c == 'n':
            return '\n'
        elif c == 'r':
            return '\r'
        elif c == 't':
            return '\t'
        elif c == 'v':
            return '\x0b'
        elif c in 'uxo':
            raise ValueError,'Invalid quoted character %r' % c
        else:
            return c
        
    def lex_braced(self):
        # Lex until whitespace or end of string, the opening brace has already
        # been consumed.

        str = ''        
        while 1:
            if self.at_end():
                raise ValueError,"Unterminated '{' quoted word"
            
            c = self.eat()
            if c == '}':
                break
            elif c == '{':
                str += '{' + self.lex_braced() + '}'
            elif c == '\\' and self.look() in '{}':
                str += self.eat()
            else:
                str += c

        return str

    def lex_quoted(self):
        str = ''

        while 1:
            if self.at_end():
                raise ValueError,"Unterminated '\"' quoted word"
            
            c = self.eat()
            if c == '"':
                break
            elif c == '\\':
                if self.at_end():
                    raise ValueError,'Missing quoted character'

                str += self.escape(self.eat())
            else:
                str += c

        return str

    def lex_unquoted(self, process_all=False):
        # Lex until whitespace or end of string.
        str = ''
        while not self.at_end():
            if not process_all:
                if self.look().isspace() or self.look() == ';':
                    break

            c = self.eat()
            if c == '\\':
                if self.at_end():
                    raise ValueError,'Missing quoted character'

                str += self.escape(self.eat())
            elif c == '[':
                raise NotImplementedError, ('Command substitution is '
                                            'not supported')
            elif c == '$' and not self.at_end() and (self.look().isalpha() or
                                                     self.look() == '{'):
                raise NotImplementedError, ('Variable substitution is '
                                            'not supported')
            else:
                str += c

        return str

    def lex_one_token(self):
        if self.maybe_eat('"'):
            return self.lex_quoted()
        elif self.maybe_eat('{'):
            # Check for argument substitution.
            if not self.maybe_eat('*'):
                return self.lex_braced()

            if not self.maybe_eat('}'):
                    return '*' + self.lex_braced()
                
            if self.at_end() or self.look().isspace():
                return '*'

            raise NotImplementedError, "Argument substitution is unsupported"
        else:
            return self.lex_unquoted()

    def lex(self):
        while not self.at_end():
            c = self.look()
            if c in ' \t':
                self.eat()
            elif c in ';\n':
                self.eat()
                yield (';',)
            else:
                yield self.lex_one_token()

class TclExecCommand:
    kRedirectPrefixes1 = ('<', '>')
    kRedirectPrefixes2 = ('<@', '<<', '2>', '>&', '>>', '>@')
    kRedirectPrefixes3 = ('2>@', '2>>', '>>&', '>&@')
    kRedirectPrefixes4 = ('2>@1',)

    def __init__(self, args):
        self.args = iter(args)

    def lex(self):
        try:
            return self.args.next()
        except StopIteration:
            return None

    def look(self):
        next = self.lex()
        if next is not None:
            self.args = itertools.chain([next], self.args)
        return next

    def parse_redirect(self, tok, length):
        if len(tok) == length:
            arg = self.lex()
            if arg is None:
                raise ValueError,'Missing argument to %r redirection' % tok
        else:
            tok,arg = tok[:length],tok[length:]

        if tok[0] == '2':
            op = (tok[1:],2)
        else:
            op = (tok,)
        return (op, arg)

    def parse_pipeline(self):
        if self.look() is None:
            raise ValueError,"Expected at least one argument to exec"

        commands = [Command([],[])]
        while 1:
            arg = self.lex()
            if arg is None:
                break
            elif arg == '|':
                commands.append(Command([],[]))
            elif arg == '|&':
                # Write this as a redirect of stderr; it must come first because
                # stdout may have already been redirected.
                commands[-1].redirects.insert(0, (('>&',2),'1'))
                commands.append(Command([],[]))
            elif arg[:4] in TclExecCommand.kRedirectPrefixes4:
                commands[-1].redirects.append(self.parse_redirect(arg, 4))
            elif arg[:3] in TclExecCommand.kRedirectPrefixes3:
                commands[-1].redirects.append(self.parse_redirect(arg, 3))
            elif arg[:2] in TclExecCommand.kRedirectPrefixes2:
                commands[-1].redirects.append(self.parse_redirect(arg, 2))
            elif arg[:1] in TclExecCommand.kRedirectPrefixes1:
                commands[-1].redirects.append(self.parse_redirect(arg, 1))
            else:
                commands[-1].args.append(arg)

        return Pipeline(commands, False, pipe_err=True)

    def parse(self):
        ignoreStderr = False
        keepNewline = False

        # Parse arguments.
        while 1:
            next = self.look()
            if not isinstance(next, str) or next[0] != '-':
                break

            if next == '--':
                self.lex()
                break
            elif next == '-ignorestderr':
                ignoreStderr = True
            elif next == '-keepnewline':
                keepNewline = True
            else:
                raise ValueError,"Invalid exec argument %r" % next

        return (ignoreStderr, keepNewline, self.parse_pipeline())

###

import unittest

class TestTclLexer(unittest.TestCase):
    def lex(self, str, *args, **kwargs):
        return list(TclLexer(str, *args, **kwargs).lex())

    def test_preprocess(self):
        self.assertEqual(tcl_preprocess('a b'), 'a b')
        self.assertEqual(tcl_preprocess('a\\\nb c'), 'a b c')

    def test_unquoted(self):
        self.assertEqual(self.lex('a b c'),
                         ['a', 'b', 'c'])
        self.assertEqual(self.lex(r'a\nb\tc\ '),
                         ['a\nb\tc '])
        self.assertEqual(self.lex(r'a \\\$b c $\\'),
                         ['a', r'\$b', 'c', '$\\'])

    def test_braced(self):
        self.assertEqual(self.lex('a {b c} {}'),
                         ['a', 'b c', ''])
        self.assertEqual(self.lex(r'a {b {c\n}}'),
                         ['a', 'b {c\\n}'])
        self.assertEqual(self.lex(r'a {b\{}'),
                         ['a', 'b{'])
        self.assertEqual(self.lex(r'{*}'), ['*'])
        self.assertEqual(self.lex(r'{*} a'), ['*', 'a'])
        self.assertEqual(self.lex(r'{*} a'), ['*', 'a'])
        self.assertEqual(self.lex('{a\\\n   b}'),
                         ['a b'])

    def test_quoted(self):
        self.assertEqual(self.lex('a "b c"'),
                         ['a', 'b c'])

    def test_terminators(self):
        self.assertEqual(self.lex('a\nb'),
                         ['a', (';',), 'b'])
        self.assertEqual(self.lex('a;b'),
                         ['a', (';',), 'b'])
        self.assertEqual(self.lex('a   ;   b'),
                         ['a', (';',), 'b'])

class TestTclExecCommand(unittest.TestCase):
    def parse(self, str):
        return TclExecCommand(list(TclLexer(str).lex())).parse()

    def test_basic(self):
        self.assertEqual(self.parse('echo hello'),
                         (False, False,
                          Pipeline([Command(['echo', 'hello'], [])],
                                   False, True)))
        self.assertEqual(self.parse('echo hello | grep hello'),
                         (False, False,
                          Pipeline([Command(['echo', 'hello'], []),
                                    Command(['grep', 'hello'], [])],
                                   False, True)))

    def test_redirect(self):
        self.assertEqual(self.parse('echo hello > a >b >>c 2> d |& e'),
                         (False, False,
                          Pipeline([Command(['echo', 'hello'],
                                            [(('>&',2),'1'),
                                             (('>',),'a'),
                                             (('>',),'b'),
                                             (('>>',),'c'),
                                             (('>',2),'d')]),
                                    Command(['e'], [])],
                                   False, True)))

if __name__ == '__main__':
    unittest.main()