Ssh connection send() and receive()
@@ -26,8 +26,9 @@ def test_can_open_and_close_connection(ssh_connection_class):
from moler.threaded_moler_connection import ThreadedMolerConnection
#moler_conn = ThreadedMolerConnection()
- # connection = ssh_connection_class(moler_connection=moler_conn, port=22, host='localhost', port=22, username='ute', password='ute')
- connection = ssh_connection_class(host='localhost', port=22, username='ute', password='ute')
+ # connection = ssh_connection_class(moler_connection=moler_conn, port=22, host='192.168.44.50', port=22, username='vagrant', password='vagrant')
+ # TODO - let it be runable on any CI machine - need SSH server in test
+ connection = ssh_connection_class(host='192.168.44.50', port=22, username='vagrant', password='vagrant')
assert connection.ssh_client.get_transport() is None
assert connection.shell_channel is None
@@ -37,7 +38,7 @@ def test_can_open_and_close_connection(ssh_connection_class):
assert connection.ssh_client.get_transport() == connection.shell_channel.get_transport()
assert connection.shell_channel.get_transport().is_active()
assert connection.shell_channel.get_transport().is_authenticated()
- assert ('127.0.0.1', 22) == connection.shell_channel.get_transport().getpeername()
+ assert ('192.168.44.50', 22) == connection.shell_channel.get_transport().getpeername()
connection.close()
assert connection.ssh_client.get_transport() is None
@@ -45,7 +46,7 @@ def test_can_open_and_close_connection(ssh_connection_class):
def test_can_open_and_close_connection_as_context_manager(ssh_connection_class):
- connection = ssh_connection_class(host='localhost', port=22, username='ute', password='ute')
+ connection = ssh_connection_class(host='192.168.44.50', port=22, username='vagrant', password='vagrant')
with connection.open():
assert connection.shell_channel.get_transport().is_authenticated()
assert connection.ssh_client.get_transport() is None
@@ -55,6 +56,28 @@ def test_can_open_and_close_connection_as_context_manager(ssh_connection_class):
assert connection.ssh_client.get_transport() is None
+# Note1: different external-IO connection may have different naming for their 'send' method
+# however, they are uniformed via glueing with moler_connection.send()
+# external-IO 'send' method works on bytes; moler_connection performs encoding
+# Note2: we check sending and receiving together - checking send by its result on receive
+def test_can_send_and_receive_binary_data_over_connection(ssh_connection_class):
+
+ connection = ssh_connection_class(host='192.168.44.50', port=22, username='vagrant', password='vagrant')
+ with connection.open():
+ time.sleep(0.1)
+ if connection.shell_channel.recv_ready(): # some banner just after open ssh
+ resp_bytes1 = connection.receive()
+ response1 = resp_bytes1.decode("utf-8")
+ assert 'Last login' in response1
+ #moler_conn.send(data=b'data to be send')
+ request = "pwd\n"
+ bytes2send = request.encode("utf-8")
+ connection.send(bytes2send)
+ time.sleep(0.1)
+ resp_bytes = connection.receive()
+ response = resp_bytes.decode("utf-8")
+ assert '/home/' in response
+
# --------------------------- resources ---------------------------
