11from unittest .mock import MagicMock
22import os
33import socket
4- from unittest import skipUnless
54
65from ssh2 .channel import Channel
76from ssh2 .error_codes import LIBSSH2_ERROR_EAGAIN
@@ -104,8 +103,6 @@ def test_failed_pkey_auth(self):
104103 self .user , 'EVEN MORE FAKE FILE' ,
105104 publickey = 'FAKE FILE' )
106105
107- @skipUnless (hasattr (Session , 'scp_recv2' ),
108- "Function not supported by libssh2" )
109106 def test_scp_recv2 (self ):
110107 self .assertEqual (self ._auth (), 0 )
111108 test_data = b"data"
@@ -132,66 +129,6 @@ def test_scp_recv2(self):
132129 os .unlink (remote_filename )
133130 self .assertRaises (SCPProtocolError , self .session .scp_recv2 , remote_filename )
134131
135- def test_scp_recv (self ):
136- self .assertEqual (self ._auth (), 0 )
137- test_data = b"data"
138- remote_filename = os .sep .join ([os .path .dirname (__file__ ),
139- "remote_test_file" ])
140- with open (remote_filename , 'wb' ) as fh :
141- fh .write (test_data )
142- try :
143- (file_chan , fileinfo ) = self .session .scp_recv (remote_filename )
144- except TypeError :
145- os .unlink (remote_filename )
146- raise
147- try :
148- total = 0
149- size , data = file_chan .read (size = fileinfo .st_size )
150- total += size
151- while total < fileinfo .st_size :
152- total += size
153- size , data = file_chan .read ()
154- self .assertEqual (total , fileinfo .st_size )
155- except Exception :
156- raise
157- finally :
158- os .unlink (remote_filename )
159- self .assertRaises (SCPProtocolError , self .session .scp_recv , remote_filename )
160-
161- def test_scp_send (self ):
162- self .assertEqual (self ._auth (), 0 )
163- test_data = b"data"
164- remote_filename = os .sep .join ([os .path .dirname (__file__ ),
165- "remote_test_file" ])
166- to_copy = os .sep .join ([os .path .dirname (__file__ ),
167- "copied" ])
168- with open (remote_filename , 'wb' ) as fh :
169- fh .write (test_data )
170- fileinfo = os .stat (remote_filename )
171- try :
172- chan = self .session .scp_send (
173- to_copy , fileinfo .st_mode & 777 , fileinfo .st_size )
174- with open (remote_filename , 'rb' ) as local_fh :
175- for data in local_fh :
176- chan .write (data )
177- chan .send_eof ()
178- chan .wait_eof ()
179- chan .wait_closed ()
180- self .assertEqual (os .stat (to_copy ).st_size ,
181- os .stat (remote_filename ).st_size )
182- except Exception :
183- raise
184- finally :
185- os .unlink (remote_filename )
186- try :
187- os .unlink (to_copy )
188- except OSError :
189- pass
190- self .assertRaises (SCPProtocolError , self .session .scp_send ,
191- '/cannot_write' , 1 & 777 , 1 )
192-
193- @skipUnless (hasattr (Session , 'scp_send64' ),
194- "Function not supported by libssh2" )
195132 def test_scp_send64 (self ):
196133 self .assertEqual (self ._auth (), 0 )
197134 test_data = b"data"
0 commit comments