| 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() |
