print(data, end=end)
sys.stdout.flush()
-# TODO: do we need to support '\n' too
-sep = "\r\n"
+sep = "\n"
rundir = None
parser = argparse.ArgumentParser(
parser.add_argument('--crlf', dest='crlf', action='store_true',
help="Write \\r\\n instead of \\n to the input")
-import errno
-def list_fds():
- """List process currently open FDs and their target """
- if sys.platform != 'linux2':
- raise NotImplementedError('Unsupported platform: %s' % sys.platform)
-
- ret = {}
- base = '/proc/self/fd'
- for num in os.listdir(base):
- path = None
- try:
- path = os.readlink(os.path.join(base, num))
- except OSError as err:
- # Last FD is always the "listdir" one (which may be closed)
- if err.errno != errno.ENOENT:
- raise
- ret[int(num)] = path
-
- return ret
-
class Runner():
def __init__(self, args, no_pty=False, line_break="\n"):
#print "args: %s" % repr(args)
env['TERM'] = 'dumb'
env['INPUTRC'] = '/dev/null'
env['PERL_RL'] = 'false'
- print("FDS before: %s" % list_fds())
if no_pty:
self.p = Popen(args, bufsize=0,
stdin=PIPE, stdout=PIPE, stderr=STDOUT,
preexec_fn=os.setsid,
- env=env, close_fds=True)
- #env=env)
+ env=env)
self.stdin = self.p.stdin
self.stdout = self.p.stdout
else:
self.p = Popen(args, bufsize=0,
stdin=slave, stdout=slave, stderr=STDOUT,
preexec_fn=os.setsid,
- env=env, close_fds=True)
- #env=env)
+ env=env)
# Now close slave so that we will get an exception from
# read when the child exits early
# http://stackoverflow.com/questions/11165521
self.stdin = os.fdopen(master, 'r+b', 0)
self.stdout = self.stdin
- print("FDS after: %s" % list_fds())
-
#print "started"
self.buf = ""
self.last_prompt = ""
#print("new_data: '%s'" % new_data)
debug(new_data)
# Perform newline cleanup
- if self.no_pty:
- self.buf += new_data.replace("\n", "\r\n")
- else:
- self.buf += new_data
- self.buf = self.buf.replace("\r\r", "\r")
- # Remove ANSI codes generally
- #ansi_escape = re.compile(r'\x1B\[[0-?]*[ -/]*[@-~]')
- # Remove rustyline ANSI CSI codes:
- # - [6C - CR + cursor forward
- # - [6K - CR + erase in line
- ansi_escape = re.compile(r'\r\x1B\[[0-9]*[CK]')
- self.buf = ansi_escape.sub('', self.buf)
+ self.buf += new_data.replace("\r", "")
for prompt in prompts:
regexp = re.compile(prompt)
match = regexp.search(self.buf)
buf = self.buf[0:match.start()]
self.buf = self.buf[end:]
self.last_prompt = prompt
- return buf.replace("^M", "\r")
+ return buf
return None
def writeline(self, str):
break
if self.ret != None: break
- if self.out[-2:] == sep and not self.ret:
+ if self.out[-1:] == sep and not self.ret:
# If there is no return value, output should not end in
# separator
- self.out = self.out[0:-2]
+ self.out = self.out[0:-1]
return self.form
args = parser.parse_args(sys.argv[1:])
# The repeated form is to get around an occasional OS X issue
# where the form is repeated.
# https://github.com/kanaka/mal/issues/30
- expects = ["%s%s%s%s" % (re.escape(t.form), sep,
- t.out, re.escape(t.ret)),
- "%s%s%s%s%s%s" % (re.escape(t.form), sep,
- re.escape(t.form), sep,
- t.out, re.escape(t.ret))]
+ expects = [".*%s%s%s" % (sep, t.out, re.escape(t.ret)),
+ ".*%s.*%s%s%s" % (sep, sep, t.out, re.escape(t.ret))]
r.writeline(t.form)
try: