Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
if patterns is None:
return []
if not isinstance(patterns, list):
patterns = [patterns]
# Allow dot to match \n
compile_flags = re.DOTALL
if self.ignorecase:
compile_flags = compile_flags | re.IGNORECASE
compiled_pattern_list = []
for idx, p in enumerate(patterns):
if isinstance(p, self.allowed_string_types):
p = self._coerce_expect_string(p)
compiled_pattern_list.append(re.compile(p, compile_flags))
elif p is EOF:
compiled_pattern_list.append(EOF)
elif p is TIMEOUT:
compiled_pattern_list.append(TIMEOUT)
elif isinstance(p, type(re.compile(''))):
compiled_pattern_list.append(p)
else:
self._pattern_type_err(p)
return compiled_pattern_list
def do_job(self, port, request):
retries = self.retries
while retries > 0:
try:
return self.driver.handle(request, port)
except (OSError, pexpect.exceptions.EOF, Exception): # pylint: disable=broad-except
self.logger.warn(traceback.format_exc())
self.logger.warn("Failed to execute job: %s %s (attempts left %i)", port, request, retries - 1)
if self.driver:
self.driver._bombout() # pylint: disable=W0212,E1101
time.sleep(5)
retries -= 1
continue
return False
project_id, region, instance_name)
instance_flag = '-instances={}=tcp:{}'.format(
instance_connection_string, port)
if cloud_sql_proxy_path is None:
cloud_sql_proxy_path = shutil.which('cloud_sql_proxy')
assert cloud_sql_proxy_path, 'could not find cloud_sql_proxy_path'
process = popen_spawn.PopenSpawn([cloud_sql_proxy_path, instance_flag])
try:
# Make sure cloud sql proxy is started before doing the real work
process.expect('Ready for new connections', timeout=60)
yield
except pexpect.exceptions.TIMEOUT:
raise DatabaseError(
('Cloud SQL Proxy was unable to start after 60 seconds. Output '
'of cloud_sql_proxy: \n{}').format(process.before))
except pexpect.exceptions.EOF:
raise DatabaseError(
('Cloud SQL Proxy exited unexpectedly. Output of '
'cloud_sql_proxy: \n{}').format(process.before))
finally:
process.kill(signal.SIGTERM)
def _connect(self):
"""Connect to the Unifi AP SSH server."""
from pexpect import pxssh, exceptions
self._ssh = pxssh.pxssh()
try:
self._ssh.login(self._host, self._username,
password=self._password, port=self._port)
self._connected = True
except exceptions.EOF:
_LOGGER.error("Connection refused. SSH enabled?")
self._disconnect()
def get_response(self, input_str):
sample = str("\n======================================== SAMPLE 1 ======================================== I'm having some trouble understanding you. Make sure you don't have any special characters in your prompt.").encode('utf-8')
attempts = 0
while attempts < 5:
try:
child = pexpect.spawn('python src/interactive_conditional_samples.py --top_k 40')
child.expect('Model prompt >>> ')
child.sendline(clean_input(input_str))
child.expect('================================================================================')
sample = child.before[len(input_str):]
break
except pexpect.exceptions.EOF:
child.kill(0)
attempts += 1
print("Attempt ", attempts, "failed. Trying again.")
return sample.decode()
def eof(self, err=None):
spawn = self.spawn
spawn.before = spawn.buffer
spawn._buffer = spawn.buffer_type()
spawn._before = spawn.buffer_type()
spawn.after = EOF
index = self.searcher.eof_index
if index >= 0:
spawn.match = EOF
spawn.match_index = index
return index
else:
spawn.match = None
spawn.match_index = None
msg = str(spawn)
msg += '\nsearcher: %s' % self.searcher
if err is not None:
msg = str(err) + '\n' + msg
exc = EOF(msg)
exc.__cause__ = None # in Python 3.x we can use "raise exc from None"
raise exc
if patterns is None:
return []
if not isinstance(patterns, list):
patterns = [patterns]
# Allow dot to match \n
compile_flags = re.DOTALL
if self.ignorecase:
compile_flags = compile_flags | re.IGNORECASE
compiled_pattern_list = []
for idx, p in enumerate(patterns):
if isinstance(p, self.allowed_string_types):
p = self._coerce_expect_string(p)
compiled_pattern_list.append(re.compile(p, compile_flags))
elif p is EOF:
compiled_pattern_list.append(EOF)
elif p is TIMEOUT:
compiled_pattern_list.append(TIMEOUT)
elif isinstance(p, type(re.compile(''))):
compiled_pattern_list.append(p)
else:
self._pattern_type_err(p)
return compiled_pattern_list
spawn._before = spawn.buffer_type()
spawn.after = EOF
index = self.searcher.eof_index
if index >= 0:
spawn.match = EOF
spawn.match_index = index
return index
else:
spawn.match = None
spawn.match_index = None
msg = str(spawn)
msg += '\nsearcher: %s' % self.searcher
if err is not None:
msg = str(err) + '\n' + msg
exc = EOF(msg)
exc.__cause__ = None # in Python 3.x we can use "raise exc from None"
raise exc
def stdin(self, line, prompt=True, timeout=3):
"""
Send line to stdin, optionally expect a prompt.
:param line: line to be send to stdin
:type line: str
:param prompt: boolean indicating whether a prompt is expected, if True absorbs \
all of stdout before inserting line into stdin and raises \
:class:`check50.Failure` if stdout is empty
:type prompt: bool
:param timeout: maximum number of seconds to wait for prompt
:type timeout: int / float
:raises check50.Failure: if ``prompt`` is set to True and no prompt is given
"""
if line == EOF:
log("sending EOF...")
else:
log(_("sending input {}...").format(line))
if prompt:
try:
self.process.expect(".+", timeout=timeout)
except (TIMEOUT, EOF):
raise Failure(_("expected prompt for input, found none"))
except UnicodeDecodeError:
raise Failure(_("output not valid ASCII text"))
# Consume everything on the output buffer
try:
for _i in range(int(timeout * 10)):
self.process.expect(".+", timeout=0.1)
expect(output, timeout=timeout)
except EOF:
result = self.child.before + self.child.buffer
if self.child.after != EOF:
result += self.child.after
raise Error(Mismatch(str_output, result.replace("\r\n", "\n")))
except TIMEOUT:
raise Error("did not find {}".format(Mismatch.raw(str_output)))
except UnicodeDecodeError:
raise Error("output not valid ASCII text")
except Exception:
raise Error("check50 could not verify output")
# If we expected EOF and we still got output, report an error.
if output == EOF and self.child.before:
raise Error(Mismatch(EOF, self.child.before.replace("\r\n", "\n")))
return self