diff options
Diffstat (limited to 'community/kitty/remove-failing-test.patch')
-rw-r--r-- | community/kitty/remove-failing-test.patch | 486 |
1 files changed, 0 insertions, 486 deletions
diff --git a/community/kitty/remove-failing-test.patch b/community/kitty/remove-failing-test.patch deleted file mode 100644 index a369cb9bed7..00000000000 --- a/community/kitty/remove-failing-test.patch +++ /dev/null @@ -1,486 +0,0 @@ -diff --git a/kitty_tests/file_transmission.py b/kitty_tests/file_transmission.py -deleted file mode 100644 -index f7f5df5..0000000 ---- a/kitty_tests/file_transmission.py -+++ /dev/null -@@ -1,480 +0,0 @@ --#!/usr/bin/env python --# License: GPLv3 Copyright: 2021, Kovid Goyal <kovid at kovidgoyal.net> -- -- --import os --import shutil --import stat --import tempfile --import zlib --from pathlib import Path -- --from kittens.transfer.librsync import ( -- LoadSignature, PatchFile, delta_for_file, signature_of_file --) --from kittens.transfer.main import parse_transfer_args --from kittens.transfer.receive import File, files_for_receive --from kittens.transfer.rsync import decode_utf8_buffer, parse_ftc --from kittens.transfer.send import files_for_send --from kittens.transfer.utils import cwd_path, expand_home, home_path, set_paths --from kitty.file_transmission import ( -- Action, Compression, FileTransmissionCommand, FileType, -- TestFileTransmission as FileTransmission, TransmissionType, -- ZlibDecompressor, iter_file_metadata --) -- --from . import BaseTest -- -- --def response(id='test', msg='', file_id='', name='', action='status', status='', size=-1): -- ans = {'action': 'status'} -- if id: -- ans['id'] = id -- if file_id: -- ans['file_id'] = file_id -- if name: -- ans['name'] = name -- if status: -- ans['status'] = status -- if size > -1: -- ans['size'] = size -- return ans -- -- --def names_in(path): -- for dirpath, dirnames, filenames in os.walk(path): -- for d in dirnames + filenames: -- yield os.path.relpath(os.path.join(dirpath, d), path) -- -- --def serialized_cmd(**fields) -> str: -- if 'id' not in fields: -- fields['id'] = 'test' -- for k, A in (('action', Action), ('ftype', FileType), ('ttype', TransmissionType), ('compression', Compression)): -- if k in fields: -- fields[k] = A[fields[k]] -- if isinstance(fields.get('data'), str): -- fields['data'] = fields['data'].encode('utf-8') -- ans = FileTransmissionCommand(**fields) -- return ans.serialize() -- -- --class TestFileTransmission(BaseTest): -- -- def setUp(self): -- self.tdir = os.path.realpath(tempfile.mkdtemp()) -- self.responses = [] -- self.orig_home = os.environ.get('HOME') -- -- def tearDown(self): -- shutil.rmtree(self.tdir) -- self.responses = [] -- if self.orig_home is None: -- os.environ.pop('HOME', None) -- else: -- os.environ['HOME'] = self.orig_home -- -- def clean_tdir(self): -- shutil.rmtree(self.tdir) -- self.tdir = os.path.realpath(tempfile.mkdtemp()) -- self.responses = [] -- -- def cr(self, a, b): -- def f(r): -- r.pop('size', None) -- return r -- a = tuple(f(r) for r in a if r.get('status') != 'PROGRESS') -- b = tuple(f(r) for r in b if r.get('status') != 'PROGRESS') -- self.ae(a, b) -- -- def assertResponses(self, ft, limit=1024, **kw): -- self.responses.append(response(**kw)) -- self.cr(ft.test_responses[:limit], self.responses[:limit]) -- -- def assertPathEqual(self, a, b): -- a = os.path.abspath(os.path.realpath(a)) -- b = os.path.abspath(os.path.realpath(b)) -- self.ae(a, b) -- -- def test_rsync_roundtrip(self): -- a_path = os.path.join(self.tdir, 'a') -- b_path = os.path.join(self.tdir, 'b') -- c_path = os.path.join(self.tdir, 'c') -- -- def files_equal(a_path, c_path): -- self.ae(os.path.getsize(a_path), os.path.getsize(c_path)) -- with open(c_path, 'rb') as b, open(c_path, 'rb') as c: -- self.ae(b.read(), c.read()) -- -- def patch(old_path, new_path, output_path, max_delta_len=0): -- sig_loader = LoadSignature() -- for chunk in signature_of_file(old_path): -- sig_loader.add_chunk(chunk) -- sig_loader.commit() -- self.assertTrue(sig_loader.finished) -- delta_len = 0 -- with PatchFile(old_path, output_path) as patcher: -- for chunk in delta_for_file(new_path, sig_loader.signature): -- self.assertFalse(patcher.finished) -- patcher.write(chunk) -- delta_len += len(chunk) -- self.assertTrue(patcher.finished) -- if max_delta_len: -- self.assertLessEqual(delta_len, max_delta_len) -- files_equal(output_path, new_path) -- -- sz = 1024 * 1024 + 37 -- with open(a_path, 'wb') as f: -- f.write(os.urandom(sz)) -- with open(b_path, 'wb') as f: -- f.write(os.urandom(sz)) -- -- patch(a_path, b_path, c_path) -- # test size of delta -- patch(a_path, a_path, c_path, max_delta_len=256) -- -- def test_file_get(self): -- # send refusal -- for quiet in (0, 1, 2): -- ft = FileTransmission(allow=False) -- ft.handle_serialized_command(serialized_cmd(action='receive', id='x', quiet=quiet)) -- self.cr(ft.test_responses, [] if quiet == 2 else [response(id='x', status='EPERM:User refused the transfer')]) -- self.assertFalse(ft.active_sends) -- # reading metadata for specs -- cwd = os.path.join(self.tdir, 'cwd') -- home = os.path.join(self.tdir, 'home') -- os.mkdir(cwd), os.mkdir(home) -- with set_paths(cwd=cwd, home=home): -- ft = FileTransmission() -- self.responses = [] -- ft.handle_serialized_command(serialized_cmd(action='receive', size=1)) -- self.assertResponses(ft, status='OK') -- ft.handle_serialized_command(serialized_cmd(action='file', file_id='missing', name='XXX')) -- self.responses.append(response(status='ENOENT:Failed to read spec', file_id='missing')) -- self.assertResponses(ft, status='OK', name=home) -- ft = FileTransmission() -- self.responses = [] -- ft.handle_serialized_command(serialized_cmd(action='receive', size=2)) -- self.assertResponses(ft, status='OK') -- with open(os.path.join(home, 'a'), 'w') as f: -- f.write('a') -- os.mkdir(f.name + 'd') -- with open(os.path.join(f.name + 'd', 'b'), 'w') as f2: -- f2.write('bbb') -- os.symlink(f.name, f.name + 'd/s') -- os.link(f.name, f.name + 'd/h') -- os.symlink('XXX', f.name + 'd/q') -- ft.handle_serialized_command(serialized_cmd(action='file', file_id='a', name='a')) -- ft.handle_serialized_command(serialized_cmd(action='file', file_id='b', name='ad')) -- files = {r['name']: r for r in ft.test_responses if r['action'] == 'file'} -- self.ae(len(files), 6) -- q = files[f.name] -- tgt = q['status'].encode('ascii') -- self.ae(q['size'], 1), self.assertNotIn('ftype', q) -- q = files[f.name + 'd'] -- self.ae(q['ftype'], 'directory') -- q = files[f.name + 'd/b'] -- self.ae(q['size'], 3) -- q = files[f.name + 'd/s'] -- self.ae(q['ftype'], 'symlink') -- self.ae(q['data'], tgt) -- q = files[f.name + 'd/h'] -- self.ae(q['ftype'], 'link') -- self.ae(q['data'], tgt) -- q = files[f.name + 'd/q'] -- self.ae(q['ftype'], 'symlink') -- self.assertNotIn('data', q) -- base = os.path.join(self.tdir, 'base') -- os.mkdir(base) -- src = os.path.join(base, 'src.bin') -- data = os.urandom(16 * 1024) -- with open(src, 'wb') as f: -- f.write(data) -- sl = os.path.join(base, 'src.link') -- os.symlink(src, sl) -- for compress in ('none', 'zlib'): -- ft = FileTransmission() -- self.responses = [] -- ft.handle_serialized_command(serialized_cmd(action='receive', size=1)) -- self.assertResponses(ft, status='OK') -- ft.handle_serialized_command(serialized_cmd(action='file', file_id='src', name=src)) -- ft.active_sends['test'].metadata_sent = True -- ft.test_responses = [] -- ft.handle_serialized_command(serialized_cmd(action='file', file_id='src', name=src, compression=compress)) -- received = b''.join(x['data'] for x in ft.test_responses) -- if compress == 'zlib': -- received = ZlibDecompressor()(received, True) -- self.ae(data, received) -- ft.test_responses = [] -- ft.handle_serialized_command(serialized_cmd(action='file', file_id='sl', name=sl, compression=compress)) -- received = b''.join(x['data'] for x in ft.test_responses) -- self.ae(received.decode('utf-8'), src) -- -- def test_file_put(self): -- # send refusal -- for quiet in (0, 1, 2): -- ft = FileTransmission(allow=False) -- ft.handle_serialized_command(serialized_cmd(action='send', id='x', quiet=quiet)) -- self.cr(ft.test_responses, [] if quiet == 2 else [response(id='x', status='EPERM:User refused the transfer')]) -- self.assertFalse(ft.active_receives) -- # simple single file send -- for quiet in (0, 1, 2): -- ft = FileTransmission() -- dest = os.path.join(self.tdir, '1.bin') -- ft.handle_serialized_command(serialized_cmd(action='send', quiet=quiet)) -- self.assertIn('test', ft.active_receives) -- self.cr(ft.test_responses, [] if quiet else [response(status='OK')]) -- ft.handle_serialized_command(serialized_cmd(action='file', name=dest)) -- self.assertPathEqual(ft.active_file('test').name, dest) -- self.assertIsNone(ft.active_file('test').actual_file) -- self.cr(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('test').name, dest) -- ft.handle_serialized_command(serialized_cmd(action='end_data', data='123')) -- self.cr(ft.test_responses, [] if quiet else [response(status='OK'), response(status='STARTED', name=dest), response(status='OK', name=dest)]) -- self.assertTrue(ft.active_receives) -- ft.handle_serialized_command(serialized_cmd(action='finish')) -- self.assertFalse(ft.active_receives) -- with open(dest) as f: -- self.ae(f.read(), 'abcd123') -- # cancel a send -- ft = FileTransmission() -- dest = os.path.join(self.tdir, '2.bin') -- ft.handle_serialized_command(serialized_cmd(action='send')) -- self.cr(ft.test_responses, [response(status='OK')]) -- ft.handle_serialized_command(serialized_cmd(action='file', name=dest)) -- self.assertPathEqual(ft.active_file('test').name, dest) -- ft.handle_serialized_command(serialized_cmd(action='data', data='abcd')) -- self.assertTrue(os.path.exists(dest)) -- ft.handle_serialized_command(serialized_cmd(action='cancel')) -- self.cr(ft.test_responses, [response(status='OK'), response(status='STARTED', name=dest), response(status='CANCELED')]) -- self.assertFalse(ft.active_receives) -- # compress with zlib -- ft = FileTransmission() -- dest = os.path.join(self.tdir, '3.bin') -- ft.handle_serialized_command(serialized_cmd(action='send')) -- self.cr(ft.test_responses, [response(status='OK')]) -- ft.handle_serialized_command(serialized_cmd(action='file', name=dest, compression='zlib')) -- self.assertPathEqual(ft.active_file('test').name, dest) -- odata = b'abcd' * 1024 + b'xyz' -- c = zlib.compressobj() -- ft.handle_serialized_command(serialized_cmd(action='data', data=c.compress(odata))) -- self.assertTrue(os.path.exists(dest)) -- ft.handle_serialized_command(serialized_cmd(action='end_data', data=c.flush())) -- self.cr(ft.test_responses, [response(status='OK'), response(status='STARTED', name=dest), response(status='OK', name=dest)]) -- ft.handle_serialized_command(serialized_cmd(action='finish')) -- with open(dest, 'rb') as f: -- self.ae(f.read(), odata) -- del odata -- -- # 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() -- dest = os.path.join(self.tdir, '2.bin') -- ft.handle_serialized_command(serialized_cmd(action='send')) -- self.assertResponses(ft, status='OK') -- fid = 0 -- -- def send(name, data, **kw): -- nonlocal fid -- fid += 1 -- kw['action'] = 'file' -- kw['file_id'] = str(fid) -- kw['name'] = name -- ft.handle_serialized_command(serialized_cmd(**kw)) -- self.assertResponses(ft, status='OK' if not data else 'STARTED', name=name, file_id=str(fid)) -- if data: -- ft.handle_serialized_command(serialized_cmd(action='end_data', file_id=str(fid), data=data)) -- self.assertResponses(ft, status='OK', name=name, file_id=str(fid)) -- -- send(dest, b'xyz', permissions=0o777, mtime=13000) -- st = os.stat(dest) -- self.ae(st.st_nlink, 1) -- self.ae(stat.S_IMODE(st.st_mode), 0o777) -- self.ae(st.st_mtime_ns, 13000) -- send(dest + 's1', 'path:' + os.path.basename(dest), permissions=0o777, mtime=17000, ftype='symlink') -- st = os.stat(dest + 's1', follow_symlinks=False) -- self.ae(stat.S_IMODE(st.st_mode), 0o777) -- self.ae(st.st_mtime_ns, 17000) -- self.ae(os.readlink(dest + 's1'), os.path.basename(dest)) -- send(dest + 's2', 'fid:1', ftype='symlink') -- self.ae(os.readlink(dest + 's2'), os.path.basename(dest)) -- send(dest + 's3', 'fid_abs:1', ftype='symlink') -- self.assertPathEqual(os.readlink(dest + 's3'), dest) -- send(dest + 'l1', 'path:' + os.path.basename(dest), ftype='link') -- self.ae(os.stat(dest).st_nlink, 2) -- send(dest + 'l2', 'fid:1', ftype='link') -- self.ae(os.stat(dest).st_nlink, 3) -- send(dest + 'd1/1', 'in_dir') -- send(dest + 'd1', '', ftype='directory', mtime=29000) -- send(dest + 'd2', '', ftype='directory', mtime=29000) -- with open(dest + 'd1/1') as f: -- self.ae(f.read(), 'in_dir') -- self.assertTrue(os.path.isdir(dest + 'd1')) -- self.assertTrue(os.path.isdir(dest + 'd2')) -- -- ft.handle_serialized_command(serialized_cmd(action='finish')) -- self.ae(os.stat(dest + 'd1').st_mtime_ns, 29000) -- self.ae(os.stat(dest + 'd2').st_mtime_ns, 29000) -- self.assertFalse(ft.active_receives) -- -- def test_parse_ftc(self): -- def t(raw, *expected): -- a = [] -- -- def c(k, v, has_semicolons): -- a.append(decode_utf8_buffer(k)) -- if has_semicolons: -- v = bytes(v).replace(b';;', b';') -- a.append(decode_utf8_buffer(v)) -- -- parse_ftc(raw, c) -- self.ae(tuple(a), expected) -- -- t('a=b', 'a', 'b') -- t('a=b;', 'a', 'b') -- t('a1=b1;c=d;;', 'a1', 'b1', 'c', 'd;') -- t('a1=b1;c=d;;e', 'a1', 'b1', 'c', 'd;e') -- t('a1=b1;c=d;;;1=1', 'a1', 'b1', 'c', 'd;', '1', '1') -- -- def test_path_mapping_receive(self): -- opts = parse_transfer_args([])[0] -- b = Path(os.path.join(self.tdir, 'b')) -- os.makedirs(b) -- open(b / 'r', 'w').close() -- os.mkdir(b / 'd') -- open(b / 'd' / 'r', 'w').close() -- -- def am(files, kw): -- m = {f.remote_path: f.expanded_local_path for f in files} -- kw = {str(k): expand_home(str(v)) for k, v in kw.items()} -- self.ae(kw, m) -- -- def gm(all_specs): -- specs = list((str(i), str(s)) for i, s in enumerate(all_specs)) -- files = [] -- for x in iter_file_metadata(specs): -- if isinstance(x, Exception): -- raise x -- files.append(File(x)) -- return files, specs -- -- def tf(args, expected, different_home=''): -- if opts.mode == 'mirror': -- all_specs = args -- dest = '' -- else: -- all_specs = args[:-1] -- dest = args[-1] -- files, specs = gm(all_specs) -- orig_home = home_path() -- with set_paths(cwd_path(), different_home or orig_home): -- files = list(files_for_receive(opts, dest, files, orig_home, specs)) -- self.ae(len(files), len(expected)) -- am(files, expected) -- -- opts.mode = 'mirror' -- with set_paths(cwd=b, home='/foo/bar'): -- tf([b/'r', b/'d'], {b/'r': b/'r', b/'d': b/'d', b/'d'/'r': b/'d'/'r'}) -- tf([b/'r', b/'d/r'], {b/'r': b/'r', b/'d'/'r': b/'d'/'r'}) -- with set_paths(cwd=b, home=self.tdir): -- tf([b/'r', b/'d'], {b/'r': '~/b/r', b/'d': '~/b/d', b/'d'/'r': '~/b/d/r'}, different_home='/foo/bar') -- opts.mode = 'normal' -- with set_paths(cwd='/some/else', home='/foo/bar'): -- tf([b/'r', b/'d', '/dest'], {b/'r': '/dest/r', b/'d': '/dest/d', b/'d'/'r': '/dest/d/r'}) -- tf([b/'r', b/'d', '~/dest'], {b/'r': '~/dest/r', b/'d': '~/dest/d', b/'d'/'r': '~/dest/d/r'}) -- with set_paths(cwd=b, home='/foo/bar'): -- tf([b/'r', b/'d', '/dest'], {b/'r': '/dest/r', b/'d': '/dest/d', b/'d'/'r': '/dest/d/r'}) -- os.symlink('/foo/b', b / 'e') -- os.symlink('r', b / 's') -- os.link(b / 'r', b / 'h') -- with set_paths(cwd='/some/else', home='/foo/bar'): -- files = gm((b/'e', b/'s', b/'r', b / 'h'))[0] -- self.assertEqual(files[0].ftype, FileType.symlink) -- self.assertEqual(files[1].ftype, FileType.symlink) -- self.assertEqual(files[1].remote_target, files[2].remote_id) -- self.assertEqual(files[3].ftype, FileType.link) -- self.assertEqual(files[3].remote_target, files[2].remote_id) -- -- def test_path_mapping_send(self): -- opts = parse_transfer_args([])[0] -- b = Path(os.path.join(self.tdir, 'b')) -- os.makedirs(b) -- open(b / 'r', 'w').close() -- os.mkdir(b / 'd') -- open(b / 'd' / 'r', 'w').close() -- -- def gm(*args): -- return files_for_send(opts, list(map(str, args))) -- -- def am(files, kw): -- m = {f.expanded_local_path: f.remote_path for f in files} -- kw = {str(k): str(v) for k, v in kw.items()} -- self.ae(m, kw) -- -- def tf(args, expected): -- files = gm(*args) -- self.ae(len(files), len(expected)) -- am(files, expected) -- -- opts.mode = 'mirror' -- with set_paths(cwd=b, home='/foo/bar'): -- tf(['r', 'd'], {b/'r': b/'r', b/'d': b/'d', b/'d'/'r': b/'d'/'r'}) -- tf(['r', 'd/r'], {b/'r': b/'r', b/'d'/'r': b/'d'/'r'}) -- with set_paths(cwd=b, home=self.tdir): -- tf(['r', 'd'], {b/'r': '~/b/r', b/'d': '~/b/d', b/'d'/'r': '~/b/d/r'}) -- opts.mode = 'normal' -- with set_paths(cwd='/some/else', home='/foo/bar'): -- tf([b/'r', b/'d', '/dest'], {b/'r': '/dest/r', b/'d': '/dest/d', b/'d'/'r': '/dest/d/r'}) -- tf([b/'r', b/'d', '~/dest'], {b/'r': '~/dest/r', b/'d': '~/dest/d', b/'d'/'r': '~/dest/d/r'}) -- with set_paths(cwd=b, home='/foo/bar'): -- tf(['r', 'd', '/dest'], {b/'r': '/dest/r', b/'d': '/dest/d', b/'d'/'r': '/dest/d/r'}) -- -- os.symlink('/foo/b', b / 'e') -- os.symlink('r', b / 's') -- os.link(b / 'r', b / 'h') -- with set_paths(cwd='/some/else', home='/foo/bar'): -- files = gm(b / 'e', 'dest') -- self.ae(files[0].symbolic_link_target, 'path:/foo/b') -- files = gm(b / 's', b / 'r', 'dest') -- self.ae(files[0].symbolic_link_target, 'fid:2') -- files = gm(b / 'h', 'dest') -- self.ae(files[0].file_type, FileType.regular) -- files = gm(b / 'h', b / 'r', 'dest') -- self.ae(files[1].file_type, FileType.link) -- self.ae(files[1].hard_link_target, '1') |