import itertools
from ShCommands import Command, Pipeline
def tcl_preprocess(data):
i = data.find('\\\n')
if i == -1:
return data
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):
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):
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('{'):
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 == '|&':
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
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()