Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_string_to_bytes():
"""Convert string to bytes"""
assert utilities.write_bytes("test") == b"test"
def test_bytes_to_bytes():
"""Convert bytes to bytes"""
result = b"hello world"
assert utilities.write_bytes(result) == result
def test_invalid_data_to_bytes():
"""Convert an invalid data type to bytes"""
try:
utilities.write_bytes(456779)
except ValueError as exc:
assert isinstance(exc, ValueError)
return
assert False
:param out_data: data to be written to the channel
:type out_data: str (can be either unicode/byte string)
"""
if self.protocol == "ssh":
self.remote_conn.sendall(write_bytes(out_data, encoding=self.encoding))
elif self.protocol == "telnet":
self.remote_conn.write(write_bytes(out_data, encoding=self.encoding))
elif self.protocol == "serial":
self.remote_conn.write(write_bytes(out_data, encoding=self.encoding))
self.remote_conn.flush()
else:
raise ValueError("Invalid protocol specified")
try:
log.debug(
"write_channel: {}".format(
write_bytes(out_data, encoding=self.encoding)
)
)
if self._session_log_fin or self.session_log_record_writes:
self._write_session_log(out_data)
except UnicodeDecodeError:
# Don't log non-ASCII characters; this is null characters and telnet IAC (PY2)
pass
def _write_session_log(self, data):
if self.session_log is not None and len(data) > 0:
# Hide the password and secret in the session_log
if self.password:
data = data.replace(self.password, "********")
if self.secret:
data = data.replace(self.secret, "********")
if isinstance(self.session_log, io.BufferedIOBase):
data = self.normalize_linefeeds(data)
self.session_log.write(write_bytes(data, encoding=self.encoding))
else:
self.session_log.write(self.normalize_linefeeds(data))
self.session_log.flush()
def _write_channel(self, out_data):
"""Generic handler that will write to both SSH and telnet channel.
:param out_data: data to be written to the channel
:type out_data: str (can be either unicode/byte string)
"""
if self.protocol == "ssh":
self.remote_conn.sendall(write_bytes(out_data, encoding=self.encoding))
elif self.protocol == "telnet":
self.remote_conn.write(write_bytes(out_data, encoding=self.encoding))
elif self.protocol == "serial":
self.remote_conn.write(write_bytes(out_data, encoding=self.encoding))
self.remote_conn.flush()
else:
raise ValueError("Invalid protocol specified")
try:
log.debug(
"write_channel: {}".format(
write_bytes(out_data, encoding=self.encoding)
)
)
if self._session_log_fin or self.session_log_record_writes:
self._write_session_log(out_data)
except UnicodeDecodeError:
def _write_channel(self, out_data):
"""Generic handler that will write to both SSH and telnet channel.
:param out_data: data to be written to the channel
:type out_data: str (can be either unicode/byte string)
"""
if self.protocol == "ssh":
self.remote_conn.sendall(write_bytes(out_data, encoding=self.encoding))
elif self.protocol == "telnet":
self.remote_conn.write(write_bytes(out_data, encoding=self.encoding))
elif self.protocol == "serial":
self.remote_conn.write(write_bytes(out_data, encoding=self.encoding))
self.remote_conn.flush()
else:
raise ValueError("Invalid protocol specified")
try:
log.debug(
"write_channel: {}".format(
write_bytes(out_data, encoding=self.encoding)
)
)
if self._session_log_fin or self.session_log_record_writes:
self._write_session_log(out_data)
except UnicodeDecodeError:
# Don't log non-ASCII characters; this is null characters and telnet IAC (PY2)
pass