TapTap
66e82f9384
-c -> --checksum, -m -> --prune-empty-dirs, -M -> --remote-option, -f -> --filter, -s -> --secluded-args, -p -> --perms, -T -> --temp-dir; -a/--archive is now real rsync -rlptgoD (links+metadata+devices+specials). FastSync's own flags moved to long-form-only or new shorts: -j/--threads (multithreading), --preserve (metadata), --sendfile, --chunk-serialization, --timeout, --ssh-port. Client-side only; the wire config fields are unchanged (no PROTOCOL_VERSION bump). The server keeps -p as its port. Docs (README, RSYNC_COMPAT summary 129->132) and unit/integration tests updated. 37/37 unit, 400-pass integration.
250 lines
9.9 KiB
Python
250 lines
9.9 KiB
Python
"""--iconv=CONVERT_SPEC file-NAME charset conversion integration tests.
|
|
|
|
The client converts every source file name from LOCAL to REMOTE before it goes
|
|
on the wire, and the receiver converts it back from REMOTE to LOCAL, so a
|
|
source tree using one charset can be written into a destination tree using
|
|
another (rsync compatibility; content bytes are never touched).
|
|
"""
|
|
import os
|
|
import shutil
|
|
|
|
import pytest
|
|
|
|
from common import TEST_DATA_DIR, run_client, clean_dir, ServerManager
|
|
|
|
LATIN1_NAME = b"caf\xe9.txt"
|
|
UTF8_NAME = "caf\u00e9.txt".encode("utf-8")
|
|
|
|
|
|
def _make(tag):
|
|
source = os.path.join(TEST_DATA_DIR, f"iconv_{tag}_src")
|
|
dest = os.path.join(TEST_DATA_DIR, f"iconv_{tag}_dst")
|
|
clean_dir(source)
|
|
shutil.rmtree(dest, ignore_errors=True)
|
|
# The destination ROOT must pre-exist on the receiver (the --mkpath contract:
|
|
# without --mkpath the server requires the root directory to exist).
|
|
os.makedirs(dest, exist_ok=True)
|
|
return source, dest
|
|
|
|
|
|
def _place_bytes(root, name_bytes, data=b"latin1 payload\n"):
|
|
full = os.path.join(os.fsencode(root), name_bytes)
|
|
os.makedirs(os.path.dirname(full), exist_ok=True)
|
|
with open(full, "wb") as fh:
|
|
fh.write(data)
|
|
return full
|
|
|
|
|
|
def _dest_file(source, dest, name):
|
|
base = os.path.join(dest, os.path.abspath(source).lstrip(os.sep))
|
|
return os.path.join(os.fsencode(base), name)
|
|
|
|
|
|
@pytest.mark.ci
|
|
def test_iconv_latin1_roundtrip(shared_server):
|
|
"""A source file whose name is ISO-8859-1 bytes is transferred with
|
|
--iconv=iso-8859-1,utf-8 and lands on the destination with the ORIGINAL
|
|
latin1 name (the wire carried it as UTF-8)."""
|
|
source, dest = _make("latin1")
|
|
_place_bytes(source, LATIN1_NAME)
|
|
|
|
result, _ = run_client(
|
|
source, dest, flags=["--iconv=iso-8859-1,utf-8"], port=shared_server.port
|
|
)
|
|
assert result.returncode == 0, (result.stderr or result.stdout)[:400]
|
|
|
|
dst = _dest_file(source, dest, LATIN1_NAME)
|
|
assert os.path.exists(dst), f"dest latin1-named file not found under {dest}"
|
|
|
|
|
|
@pytest.mark.ci
|
|
def test_iconv_to_utf8_on_wire(shared_server):
|
|
"""--iconv=utf-8 (single, identity both ways) on an ascii filename transfers
|
|
cleanly with no error."""
|
|
source, dest = _make("utf8")
|
|
src_path = os.path.join(source, "plain.txt")
|
|
with open(src_path, "wb") as fh:
|
|
fh.write(b"identity\n")
|
|
|
|
result, _ = run_client(source, dest, flags=["--iconv=utf-8"], port=shared_server.port)
|
|
assert result.returncode == 0, (result.stderr or result.stdout)[:400]
|
|
|
|
dst = _dest_file(source, dest, os.fsencode("plain.txt"))
|
|
assert os.path.exists(dst)
|
|
|
|
|
|
@pytest.mark.ci
|
|
def test_iconv_passthrough_identity(shared_server):
|
|
"""No --iconv flag: the transfer is unchanged (regression guard -- the common
|
|
path must not go through iconv at all)."""
|
|
source, dest = _make("identity")
|
|
for name, data in (("a.txt", b"aaa\n"), ("sub/b.txt", b"bbb\n")):
|
|
p = os.path.join(source, name)
|
|
os.makedirs(os.path.dirname(p), exist_ok=True)
|
|
with open(p, "wb") as fh:
|
|
fh.write(data)
|
|
|
|
result, _ = run_client(source, dest, port=shared_server.port)
|
|
assert result.returncode == 0, (result.stderr or result.stdout)[:400]
|
|
|
|
for name in ("a.txt", "sub/b.txt"):
|
|
assert os.path.exists(_dest_file(source, dest, os.fsencode(name)))
|
|
|
|
|
|
@pytest.mark.ci
|
|
def test_iconv_receiver_own_charset(shared_server):
|
|
"""A dedicated server started with its OWN --iconv converts received names
|
|
to ITS charset: the source holds a latin1-named file, the wire carries it
|
|
as UTF-8 (from the client's spec), and the receiver re-decodes it to UTF-8
|
|
on disk. This discriminates a real wire conversion from a no-op passthrough
|
|
(a latin1 byte sequence is not valid UTF-8, so the receiver decoding it as
|
|
UTF-8 would fail the transfer)."""
|
|
with ServerManager() as server:
|
|
server.start(extra_args=["--iconv=utf-8"])
|
|
source, dest = _make("recv_charset")
|
|
_place_bytes(source, LATIN1_NAME)
|
|
|
|
result, _ = run_client(
|
|
source, dest, flags=["--iconv=iso-8859-1,utf-8"], port=server.port
|
|
)
|
|
assert result.returncode == 0, (result.stderr or result.stdout)[:400]
|
|
|
|
dst = _dest_file(source, dest, UTF8_NAME)
|
|
assert os.path.exists(dst), f"dest UTF-8-named file not found under {dest}"
|
|
|
|
|
|
@pytest.mark.ci
|
|
def test_iconv_invalid_charset_rejected(shared_server):
|
|
"""An unsupported charset name is rejected at startup with a nonzero exit."""
|
|
source, dest = _make("badcharset")
|
|
src_path = os.path.join(source, "f.txt")
|
|
with open(src_path, "wb") as fh:
|
|
fh.write(b"x")
|
|
|
|
result, _ = run_client(
|
|
source, dest, flags=["--iconv=no-such-charset,utf-8"], port=shared_server.port
|
|
)
|
|
assert result.returncode != 0
|
|
|
|
|
|
@pytest.mark.ci
|
|
def test_iconv_garbage_spec_rejected(shared_server):
|
|
"""A malformed CONVERT_SPEC is rejected at startup with a nonzero exit."""
|
|
source, dest = _make("garbage")
|
|
src_path = os.path.join(source, "f.txt")
|
|
with open(src_path, "wb") as fh:
|
|
fh.write(b"x")
|
|
|
|
result, _ = run_client(source, dest, flags=["--iconv=,,,"], port=shared_server.port)
|
|
assert result.returncode != 0
|
|
|
|
|
|
@pytest.mark.ci
|
|
def test_iconv_expanding_name_growth(shared_server):
|
|
"""A long latin1 name whose UTF-8 encoding expands past the initial output
|
|
buffer exercises the E2BIG growth path in charset_convert (each high-bit
|
|
latin1 byte doubles in UTF-8), and must land unchanged on the destination."""
|
|
source, dest = _make("growth")
|
|
name_bytes = b"a" * 40 + bytes(range(0x80, 0x80 + 40)) + b".txt"
|
|
_place_bytes(source, name_bytes, data=b"growth\n")
|
|
|
|
result, _ = run_client(
|
|
source, dest, flags=["--iconv=iso-8859-1,utf-8"], port=shared_server.port
|
|
)
|
|
assert result.returncode == 0, (result.stderr or result.stdout)[:400]
|
|
|
|
assert os.path.exists(_dest_file(source, dest, name_bytes))
|
|
|
|
|
|
def test_iconv_symlink_path_and_target(shared_server):
|
|
"""A latin1-named symlink pointing at a latin1-named target survives the
|
|
transfer: both the link name and the link target are wire-converted and
|
|
re-decoded on the destination (-l preserves links)."""
|
|
source, dest = _make("symlink")
|
|
target = b"target\xe9.dat"
|
|
_place_bytes(source, target, data=b"t\n")
|
|
os.symlink(target, os.path.join(os.fsencode(source), b"link\xe9"))
|
|
|
|
result, _ = run_client(
|
|
source, dest, flags=["--iconv=iso-8859-1,utf-8", "--links"], port=shared_server.port
|
|
)
|
|
assert result.returncode == 0, (result.stderr or result.stdout)[:400]
|
|
|
|
dst_target = _dest_file(source, dest, target)
|
|
dst_link = _dest_file(source, dest, b"link\xe9")
|
|
assert os.path.exists(dst_target), "dest latin1 target file missing"
|
|
assert os.path.islink(dst_link), "dest latin1 symlink missing"
|
|
assert os.readlink(dst_link) == target, "symlink target not preserved/decoded"
|
|
with open(dst_link, "rb") as fh:
|
|
assert fh.read() == b"t\n"
|
|
|
|
|
|
def test_iconv_hardlink_path_and_target(shared_server):
|
|
"""A latin1-named hard-linked pair is preserved: -H transmits later group
|
|
members as a path+target link to the first member, so both the member name
|
|
and the target wire-convert (the two destination names must stay one
|
|
inode)."""
|
|
source, dest = _make("hardlink")
|
|
a = b"hl_a\xe9.txt"
|
|
b = b"hl_b\xe9.txt"
|
|
src_a = os.path.join(os.fsencode(source), a)
|
|
with open(src_a, "wb") as fh:
|
|
fh.write(b"shared\n")
|
|
os.link(src_a, os.path.join(os.fsencode(source), b))
|
|
|
|
result, _ = run_client(
|
|
source, dest, flags=["--iconv=iso-8859-1,utf-8", "--hard-links"],
|
|
port=shared_server.port,
|
|
)
|
|
assert result.returncode == 0, (result.stderr or result.stdout)[:400]
|
|
|
|
dst_a = _dest_file(source, dest, a)
|
|
dst_b = _dest_file(source, dest, b)
|
|
assert os.path.exists(dst_a) and os.path.exists(dst_b)
|
|
assert os.stat(dst_a).st_ino == os.stat(dst_b).st_ino, \
|
|
"hard-link relationship not preserved across the transfer"
|
|
|
|
|
|
def test_iconv_delete_manifest_consistent(shared_server):
|
|
"""Combining --iconv with --delete: the delete manifest's keep-set paths are
|
|
wire-converted on send and disk-converted on receive, so the receiver's
|
|
delete walker compares like with like and removes exactly the missing
|
|
latin1-named file (never a wrong-named mirror)."""
|
|
source, dest = _make("delete")
|
|
keep = b"keep\xe9.txt"
|
|
gone = b"gone\xe9.txt"
|
|
_place_bytes(source, keep, data=b"k\n")
|
|
_place_bytes(source, gone, data=b"g\n")
|
|
|
|
with ServerManager() as server:
|
|
server.start(extra_args=["--allow-delete"])
|
|
flags = ["--iconv=iso-8859-1,utf-8"]
|
|
result, _ = run_client(source, dest, flags=flags, port=server.port)
|
|
assert result.returncode == 0, (result.stderr or result.stdout)[:400]
|
|
assert os.path.exists(_dest_file(source, dest, keep))
|
|
assert os.path.exists(_dest_file(source, dest, gone))
|
|
|
|
os.remove(os.path.join(os.fsencode(source), gone))
|
|
result, _ = run_client(
|
|
source, dest, flags=flags + ["--delete"], port=server.port
|
|
)
|
|
assert result.returncode == 0, (result.stderr or result.stdout)[:400]
|
|
assert os.path.exists(_dest_file(source, dest, keep)), "kept file deleted"
|
|
assert not os.path.exists(_dest_file(source, dest, gone)), \
|
|
"missing file was not deleted"
|
|
|
|
|
|
def test_iconv_chunk_serialization_blob(shared_server):
|
|
"""-s (chunk serialization) embeds paths and symlink targets inside the
|
|
serialized chunk blob rather than as separate frames; a latin1 name must
|
|
still wire-convert and re-decoded on the destination."""
|
|
source, dest = _make("chunk")
|
|
name = b"\xe9\xe9\xe9\xe9\xe9\xe9\xe9\xe9\xe9\xe9\xe9\xe9\xe9\xe9\xe9\xe9.txt"
|
|
_place_bytes(source, name, data=b"blob\n")
|
|
|
|
result, _ = run_client(
|
|
source, dest, flags=["--iconv=iso-8859-1,utf-8", "--chunk-serialization"], port=shared_server.port
|
|
)
|
|
assert result.returncode == 0, (result.stderr or result.stdout)[:400]
|
|
|
|
assert os.path.exists(_dest_file(source, dest, name)) |