Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def execute(cmd):
output, status = pexpect.run(cmd
, withexitstatus=1
, timeout=600
)
return output
def test_run_event_as_string(self):
events = [
# second match on 'abc', echo 'def'
('abc\r\n.*GO:', 'echo "def"\n'),
# final match on 'def': exit
('def\r\n.*GO:', 'exit\n'),
# first match on 'GO:' prompt, echo 'abc'
('GO:', 'echo "abc"\n')
]
(data, exitstatus) = pexpect.run(
'bash --rcfile {0}'.format(self.rcfile),
withexitstatus=True,
events=events,
timeout=10)
assert exitstatus == 0
def test_json_output(self):
pexpect.run(f"check50 --dev -o json --output-file foo.json {CHECKS_DIRECTORY}/output")
with open("foo.json", "r") as f:
json_out = json.load(f)
self.assertEqual(json_out["results"][0]["name"], "exists")
def download(full=None):
cmd = "python3 -m deeppavlov.download -test"
if full:
cmd += " -all"
pexpect.run(cmd, timeout=None)
def execute_playbook(self, _logfile=None):
"""
Execute ansible-playbook
:param _logfile: Use this file-like object instead of a LoggerFile for
testing
"""
environ = os.environ
environ['ANSIBLE_SSH_PIPELINING'] = '1'
environ['ANSIBLE_ROLES_PATH'] = "%s/roles" % self.repo_path
args = self._build_args()
command = ' '.join(args)
log.debug("Running %s", command)
out_log = self.log.getChild('out')
out, status = pexpect.run(
command,
logfile=_logfile or LoggerFile(out_log, logging.INFO),
withexitstatus=True,
timeout=None,
)
if status != 0:
raise CommandFailedError(command, status)
def android_adb_over_tcp_disconnect(self):
dev_ip = self.dev_ip
adb_port = self._client.config.android_adb_port
cmd = "adb disconnect %s:%s" % (dev_ip, adb_port)
logging.info("Execute adb command on host: %s" % cmd)
pexpect.run(cmd, timeout=300, logfile=sys.stdout)
def cvs_admin_kb (filename):
"""This uses 'cvs admin' to set the '-kb' sticky option.
"""
s = pexpect.run ('cvs admin -kb %s' % filename)
# There is a timing issue. If I run 'cvs admin' too quickly
# cvs sometimes has trouble obtaining the directory lock.
time.sleep(1)
pexpect_session_manager=PexpectSessionManager(args.l, debug=args.d, timesync=timesync, colors_on=colors_on, replayspeed=args.replayspeed)
if args.replayfile:
replay_file(pexpect_session_manager, args.replayfile[0])
else:
if args.replay:
replay_dir(pexpect_session_manager, args)
if args.commands:
pexpect_session_manager.initialize_commands(args)
main_command_session = None
for session in pexpect_session_manager.pexpect_sessions:
if session.session_number == 0:
main_command_session = session
else:
session.spawn()
assert main_command_session, pexpect_session_manager.quit_autotrace('No main command session set up!')
pexpect.run('kill -CONT ' + str(main_command_session.pid))
while True:
# TODO: doesn't reset if set back on?
#os.system('stty -echo')
try:
while True:
pexpect_session_manager.draw_screen('sessions',quick_help=pexpect_session_manager.get_quick_help())
pexpect_session_manager.handle_sessions()
pexpect_session_manager.handle_input()
except KeyboardInterrupt:
pexpect_session_manager.draw_screen('clearscreen',quick_help=pexpect_session_manager.get_quick_help())
pexpect_session_manager.refresh_window()
pexpect_session_manager.quit_autotrace('Interrupt detected.')
logger.info('Got incoming request')
if 'command' not in request.form:
return jsonify(error="Missing parameter 'command'"), 422
command = request.form['command']
file_name = _extract_filename_from_command(command)
if file_name is not None and not os.path.isfile(file_name):
logger.warn("Couldn't find file %s", file_name)
if not command.startswith('idaw ') and not command.startswith('idaw64 '):
return jsonify(error="'idaw' and 'idaw64' are the only valid commands"), 422
try:
logger.info('Executing %s', command)
timeout = None if 'timeout' not in request.form else int(request.form['timeout'])
_, exit_code = pexpect.run(command, timeout=timeout, withexitstatus=True)
except pexpect.TIMEOUT:
return jsonify(error='request to ida timed out'), 408
finally:
if file_name is not None:
_remove_ida_created_files(file_name)
logger.info('Removed ida leftover files')
if exit_code == 0:
logger.info('Command %s finished executing successfully', command)
else:
logger.warn("Command %s didn't finish correctly, IDA returned exit code %s", command, exit_code)
if exit_code != 0:
return jsonify(error='ida finish with status code %s' % exit_code), 500
else:
return jsonify(message='OK'), 200