mirrors/kitty
1
0
Fork 0
mirror of https://github.com/kovidgoyal/kitty synced 2026-06-08 14:18:26 +02:00
Code Issues Packages Projects Releases Wiki Activity

More work on file transfer

Browse Source
This commit is contained in:
Kovid Goyal
2021-09-09 12:59:31 +05:30
parent 2178c8e4af
commit f9c99a61d4
6 changed files with 257 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

43
kitty_tests/file_transmission.py
View File

@@ -21,7 +21,7 @@ from kitty.file_transmission import (
from . import BaseTest
def response(id='', msg='', file_id='', name='', action='status', status=''):
def response(id='', msg='', file_id='', name='', action='status', status='', size=-1):
ans = {'action': 'status'}
if id:
ans['id'] = id
@@ -31,6 +31,8 @@ def response(id='', msg='', file_id='', name='', action='status', status=''):
ans['name'] = name
if status:
ans['status'] = status
if size > -1:
ans['size'] = size
return ans
@@ -93,12 +95,12 @@ class TestFileTransmission(BaseTest):
ft.handle_serialized_command(serialized_cmd(action='send', quiet=quiet))
self.assertIn('', ft.active_receives)
self.ae(ft.test_responses, [] if quiet else [response(status='OK')])
ft.handle_serialized_command(serialized_cmd(action='file', name=dest, quiet=quiet))
ft.handle_serialized_command(serialized_cmd(action='file', name=dest))
self.assertPathEqual(ft.active_file().name, dest)
self.assertIsNone(ft.active_file().actual_file)
self.ae(ft.test_responses, [] if quiet else [response(status='OK'), response(status='STARTED', name=dest)])
ft.handle_serialized_command(serialized_cmd(action='data', data='abcd'))
self.assertPathEqual(ft.active_file().actual_file.name, dest)
self.assertPathEqual(ft.active_file().name, dest)
ft.handle_serialized_command(serialized_cmd(action='end_data', data='123'))
self.ae(ft.test_responses, [] if quiet else [response(status='OK'), response(status='STARTED', name=dest), response(status='OK', name=dest)])
self.assertTrue(ft.active_receives)
@@ -137,6 +139,41 @@ class TestFileTransmission(BaseTest):
del odata
del data
# overwriting
self.clean_tdir()
ft = FileTransmission()
one = os.path.join(self.tdir, '1')
two = os.path.join(self.tdir, '2')
three = os.path.join(self.tdir, '3')
open(two, 'w').close()
os.symlink(two, one)
ft.handle_serialized_command(serialized_cmd(action='send'))
ft.handle_serialized_command(serialized_cmd(action='file', name=one))
ft.handle_serialized_command(serialized_cmd(action='end_data', data='abcd'))
ft.handle_serialized_command(serialized_cmd(action='finish'))
self.assertFalse(os.path.islink(one))
with open(one) as f:
self.ae(f.read(), 'abcd')
self.assertTrue(os.path.isfile(two))
ft = FileTransmission()
ft.handle_serialized_command(serialized_cmd(action='send'))
ft.handle_serialized_command(serialized_cmd(action='file', name=two, ftype='symlink'))
ft.handle_serialized_command(serialized_cmd(action='end_data', data='path:/abcd'))
ft.handle_serialized_command(serialized_cmd(action='finish'))
self.ae(os.readlink(two), '/abcd')
with open(three, 'w') as f:
f.write('abcd')
self.responses = []
ft = FileTransmission()
ft.handle_serialized_command(serialized_cmd(action='send'))
self.assertResponses(ft, status='OK')
ft.handle_serialized_command(serialized_cmd(action='file', name=three))
self.assertResponses(ft, status='STARTED', name=three, size=4)
ft.handle_serialized_command(serialized_cmd(action='end_data', data='11'))
ft.handle_serialized_command(serialized_cmd(action='finish'))
with open(three) as f:
self.ae(f.read(), '11')
# multi file send
self.clean_tdir()
ft = FileTransmission()