#!/usr/bin/env python3

"""
Test suite for blocksync.py.

Run all tests:
./test_blocksync.py

Run one group:
./test_blocksync.py -v TestErrorPaths

Include the latency benchmark, which takes a few seconds:
BLOCKSYNC_BENCH=1 ./test_blocksync.py

Test a different version:
BLOCKSYNC_PATH=./blocksync.py.orig ./test_blocksync.py

Simulate a laggy link, e.g. for the latency benchmark above (seconds of one-way delay per read):
FAKE_SSH_DELAY=0.05 BLOCKSYNC_BENCH=1 ./test_blocksync.py

Simulate a chatty ssh login that floods stderr, to check nothing hangs on it (bytes of noise):
FAKE_SSH_STDERR_NOISE=5000000 ./test_blocksync.py -v TestRemote

Notes
-----
The only dependency is Python3, same as blocksync itself. Test data is generated on the fly in a
temporary directory, so no binary fixtures are stored alongside the code and every run starts from
a known state.

Remote paths are covered by a stand-in ssh that is written into a temporary bin directory and put
in front of PATH. It runs the command it is handed through a real shell, so the worker source has
to survive shell quoting exactly as it would on a remote host. Two environment variables shape its
behaviour, both read fresh by the stand-in itself on every invocation:

- FAKE_SSH_DELAY: one-way propagation delay in seconds, applied to every chunk of data flowing in
  either direction. Used by the latency benchmark (TestLatency) to show the round-trip cost of
  comparing one chunk at a time; ordinary correctness tests leave it unset (delay 0).
- FAKE_SSH_STDERR_NOISE: number of bytes of "X" the stand-in writes to stderr immediately on
  startup, before running anything, to imitate a verbose ssh login banner or MOTD. Used by
  test_chatty_ssh_does_not_hang to make sure that output alone cannot deadlock a transfer; a value
  bigger than the OS pipe buffer (typically 64 KB) is needed to actually exercise that path, so the
  tests that set it use a few megabytes.

Both are read by fake_ssh_env() as **extra keyword arguments, so an individual test can set either
without disturbing the rest of the environment, for example fake_ssh_env(FAKE_SSH_DELAY = 0.01).

Written 2026 by Claude.ai as part of a code audit. CC0, same as the code it tests.
"""


import unittest
import subprocess
import sys
import os
import stat
import hashlib
import random
import re
import tempfile
import shutil


MB = 1024 * 1024
# Which blocksync to test. The override makes it possible to point the same suite at another
# version, which is how these tests were checked against the code as it was before the audit.
BLOCKSYNC = os.environ.get("BLOCKSYNC_PATH", os.path.join(os.path.dirname(os.path.abspath(__file__)), "blocksync.py"))


# Stand-in for ssh. Strips the "--" and the hostname, then runs the rest through a shell, so the
# quoting of the worker source is exercised for real.
#
# FAKE_SSH_DELAY (seconds, default 0): one-way propagation delay. Every piece of data gets a
# release time and a separate thread writes it out then, so the link is delayed without being
# throttled: this changes latency, not bandwidth.
#
# FAKE_SSH_STDERR_NOISE (byte count, default 0): writes that many "X" bytes to stderr before
# doing anything else, standing in for a chatty ssh login banner. See the module docstring for
# how each is meant to be used.
FAKE_SSH = '''#!/usr/bin/env python3
import sys, os, subprocess, threading, time, queue
DELAY = float(os.environ.get("FAKE_SSH_DELAY", "0"))
NOISE = int(os.environ.get("FAKE_SSH_STDERR_NOISE", "0"))
if NOISE:
\tsys.stderr.write("X" * NOISE + "\\n")
\tsys.stderr.flush()
a = sys.argv[1:]
if a and a[0] == "--":
\ta = a[1:]
a = a[1:] # drop hostname
p = subprocess.Popen(["/bin/sh", "-c", " ".join(a)], stdin = subprocess.PIPE,
\t\tstdout = subprocess.PIPE, stderr = sys.stderr.buffer)
def link(src, dst):
\tq = queue.Queue()
\tdef reader():
\t\ttry:
\t\t\twhile True:
\t\t\t\tb = src.read1(262144)
\t\t\t\tq.put((time.monotonic() + DELAY, b))
\t\t\t\tif not b:
\t\t\t\t\tbreak
\t\texcept Exception:
\t\t\tq.put((time.monotonic() + DELAY, b""))
\tdef writer():
\t\ttry:
\t\t\twhile True:
\t\t\t\trel, b = q.get()
\t\t\t\tnow = time.monotonic()
\t\t\t\tif rel > now:
\t\t\t\t\ttime.sleep(rel - now)
\t\t\t\tif not b:
\t\t\t\t\tbreak
\t\t\t\tdst.write(b)
\t\t\t\tdst.flush()
\t\texcept Exception:
\t\t\tpass
\t\ttry:
\t\t\tdst.close()
\t\texcept Exception:
\t\t\tpass
\tthreading.Thread(target = reader, daemon = True).start()
\tt = threading.Thread(target = writer, daemon = True)
\tt.start()
\treturn t
link(sys.stdin.buffer, p.stdin)
t2 = link(p.stdout, sys.stdout.buffer)
rc = p.wait()
t2.join(timeout = 10)
try:
\tsys.stdout.buffer.flush()
except Exception:
\tpass
os._exit(rc) # skip interpreter shutdown, the stdin reader is parked on a read
'''


def chunk_bytes(index, variant, size):
	"""
	Deterministic pseudo-random contents for one chunk.

	Both index and variant go into the seed, so chunk 3 variant "a" differs from chunk 3 variant "b"
	and from chunk 4 variant "a". Seeding on the index alone would make chunk 0 identical across
	variants whenever the seeds are combined by multiplication, which quietly turns a difference
	test into a no-op.
	"""

	return random.Random(index * 1000003 + variant).randbytes(size)


def make_image(path, variants, chunk_size = MB, last_chunk_size = None):
	"""
	Write a file of len(variants) chunks, each filled according to its variant number.

	last_chunk_size, if given, shortens only the final chunk to that many bytes, so callers can
	build a source whose size is not an exact multiple of chunk_size. It has no effect on an empty
	variants list. Content for the shortened chunk is still seeded from its index and variant, so a
	"same last chunk" and a "different last chunk" file still agree or disagree only on that chunk,
	same as with full-size chunks.
	"""

	with open(path, "wb") as file_handle:
		last_index = len(variants) - 1
		for index, variant in enumerate(variants):
			size = chunk_size
			if last_chunk_size is not None and index == last_index:
				size = last_chunk_size
			file_handle.write(chunk_bytes(index, variant, size))


def read_bytes(path):
	with open(path, "rb") as file_handle:
		return file_handle.read()


def digest(path):
	hasher = hashlib.sha256()
	with open(path, "rb") as file_handle:
		for block in iter(lambda: file_handle.read(1 << 20), b""):
			hasher.update(block)
	return hasher.hexdigest()


class BlocksyncTestCase(unittest.TestCase):
	"""
	Shared scaffolding: a temporary directory per test and a helper to run blocksync.
	"""

	def setUp(self):
		self.dir = tempfile.mkdtemp(prefix = "blocksync-test-")
		self.src = os.path.join(self.dir, "src.img")
		self.dst = os.path.join(self.dir, "dst.img")

	def tearDown(self):
		# chmod back, a read-only file would otherwise survive the cleanup
		for name in os.listdir(self.dir):
			try:
				os.chmod(os.path.join(self.dir, name), 0o644)
			except OSError:
				pass
		shutil.rmtree(self.dir, ignore_errors = True)

	def fake_ssh_env(self, **extra):
		"""
		Return an environment with the stand-in ssh in front of PATH.

		extra is passed straight through to the environment (values stringified), which is how
		individual tests configure the stand-in: fake_ssh_env(FAKE_SSH_DELAY = 0.01) or
		fake_ssh_env(FAKE_SSH_STDERR_NOISE = 5 * MB). See the module docstring and the comment
		above FAKE_SSH for what each variable does; unset ones default to off (0).
		"""

		bin_dir = os.path.join(self.dir, "bin")
		os.makedirs(bin_dir, exist_ok = True)
		ssh_path = os.path.join(bin_dir, "ssh")
		with open(ssh_path, "w") as file_handle:
			file_handle.write(FAKE_SSH)
		os.chmod(ssh_path, os.stat(ssh_path).st_mode | stat.S_IEXEC | stat.S_IXGRP | stat.S_IXOTH)

		env = dict(os.environ)
		env["PATH"] = bin_dir + os.pathsep + env.get("PATH", "")
		env.update({key: str(value) for key, value in extra.items()})
		return env

	def run_blocksync(self, *args, env = None, timeout = 120):
		"""
		Run blocksync and return the completed process. Never raises on a non-zero exit, the tests
		assert on the status themselves.
		"""

		return subprocess.run([sys.executable, BLOCKSYNC, *args],
				capture_output = True, text = True, env = env, timeout = timeout)

	def assert_no_traceback(self, result):
		self.assertNotIn("Traceback", result.stderr,
				f"blocksync crashed instead of reporting an error:\n{result.stderr}")

	def assert_synced(self):
		self.assertEqual(digest(self.dst), digest(self.src), "destination does not match source")


class TestLocalSync(BlocksyncTestCase):
	"""
	The core job: make the destination match the source, and touch nothing else.
	"""

	def test_identical_files_stay_identical(self):
		make_image(self.src, [0, 1, 2, 3])
		make_image(self.dst, [0, 1, 2, 3])
		before = digest(self.dst)
		result = self.run_blocksync("-q", self.src, self.dst)
		self.assertEqual(result.returncode, 0, result.stderr)
		self.assertEqual(digest(self.dst), before)

	def test_single_chunk_differs(self):
		make_image(self.src, [0, 1, 2, 3])
		make_image(self.dst, [0, 1, 9, 3])
		result = self.run_blocksync("-q", self.src, self.dst)
		self.assertEqual(result.returncode, 0, result.stderr)
		self.assert_synced()

	def test_first_chunk_differs(self):
		# chunk 0 deserves its own test, off-by-one and falsy-index mistakes land here
		make_image(self.src, [0, 1, 2, 3])
		make_image(self.dst, [9, 1, 2, 3])
		self.run_blocksync("-q", self.src, self.dst)
		self.assert_synced()

	def test_last_chunk_differs(self):
		# the transfer of the final chunk is followed by shutdown, so a missing flush shows up here
		make_image(self.src, [0, 1, 2, 3])
		make_image(self.dst, [0, 1, 2, 9])
		self.run_blocksync("-q", self.src, self.dst)
		self.assert_synced()

	def test_every_chunk_differs(self):
		make_image(self.src, [0, 1, 2, 3])
		make_image(self.dst, [9, 8, 7, 6])
		self.run_blocksync("-q", self.src, self.dst)
		self.assert_synced()

	def test_tail_beyond_source_is_left_alone(self):
		# a destination may be bigger than the source, anything past the source must survive
		make_image(self.src, [0, 1])
		make_image(self.dst, [9, 9, 5, 6])
		tail_before = read_bytes(self.dst)[2 * MB:]
		result = self.run_blocksync("-q", self.src, self.dst)
		self.assertEqual(result.returncode, 0, result.stderr)
		body = read_bytes(self.dst)
		self.assertEqual(body[:2 * MB], read_bytes(self.src))
		self.assertEqual(body[2 * MB:], tail_before, "data past the source length was modified")

	def test_second_run_copies_nothing(self):
		make_image(self.src, [0, 1, 2, 3])
		make_image(self.dst, [9, 1, 9, 3])
		self.run_blocksync("-q", self.src, self.dst)
		result = self.run_blocksync("-v", self.src, self.dst)
		self.assertEqual(result.returncode, 0, result.stderr)
		self.assertIn("different: 0", result.stdout, "a repeat run should find no differences")

	def test_empty_source(self):
		# zero chunks used to divide by zero in the progress line
		open(self.src, "wb").close()
		make_image(self.dst, [0, 1])
		result = self.run_blocksync("-q", self.src, self.dst)
		self.assert_no_traceback(result)
		self.assertEqual(result.returncode, 0, result.stderr)


class TestChunkSize(BlocksyncTestCase):
	"""
	--chunk-size has to produce an identical result at every setting.
	"""

	def test_all_sizes_sync_correctly(self):
		for size_mb in (1, 2, 4, 8, 16):
			with self.subTest(chunk_size_mb = size_mb):
				make_image(self.src, [0] * 16)
				make_image(self.dst, [0, 0, 9, 0, 0, 0, 0, 9, 0, 0, 0, 0, 0, 0, 0, 9])
				result = self.run_blocksync("-q", "-c", str(size_mb), self.src, self.dst)
				self.assertEqual(result.returncode, 0, result.stderr)
				self.assert_synced()

	def test_source_need_not_be_multiple_of_chunk_size(self):
		# a source size that is not an exact multiple of chunk_size is a normal case, not an error.
		# See TestPartialLastChunk for the detailed matrix this shorthand stands in for.
		make_image(self.src, [0, 1, 2], last_chunk_size = 512 * 1024)
		make_image(self.dst, [0, 1, 9], last_chunk_size = 512 * 1024)
		result = self.run_blocksync("-q", self.src, self.dst)
		self.assertEqual(result.returncode, 0, result.stderr)
		self.assert_synced()

	def test_zero_chunk_size_is_rejected(self):
		make_image(self.src, [0])
		make_image(self.dst, [0])
		result = self.run_blocksync("-c", "0", self.src, self.dst)
		self.assert_no_traceback(result)
		self.assertEqual(result.returncode, 1)


class TestPartialLastChunk(BlocksyncTestCase):
	"""
	The source size need not be a multiple of chunk_size. The worker is told src_size so it can
	compute the length of that last, short chunk instead of always reading/sending/receiving a
	full chunk_size; these tests cover the size relationships where that arithmetic can go wrong.

	Full chunks before the last one always use variant 0 on both sides (identical), so a failure
	can only come from how the last, partial chunk is handled, not from unrelated differences
	earlier in the file.

	Sizes below use a 1 MB chunk_size and a 512 KB final chunk throughout, so "last chunk" always
	means "the second, half-size chunk" and every scenario is the same shape: two chunks, chunk 0
	full and identical, chunk 1 half-size and either identical or different.
	"""

	LAST = 512 * 1024 # last chunk size in bytes: half of the default 1 MB chunk_size

	def test_src_dst_same_size_last_chunk_identical(self):
		# both files end exactly at 1.5 MB, last chunk already matches
		make_image(self.src, [0, 1], last_chunk_size = self.LAST)
		make_image(self.dst, [0, 1], last_chunk_size = self.LAST)
		before = digest(self.dst)
		result = self.run_blocksync("-v", self.src, self.dst)
		self.assertEqual(result.returncode, 0, result.stderr)
		self.assertEqual(digest(self.dst), before, "an already-synced partial last chunk was rewritten")
		self.assertIn("different: 0", result.stdout)

	def test_src_dst_same_size_last_chunk_different(self):
		# both files end exactly at 1.5 MB, last chunk needs transferring
		make_image(self.src, [0, 1], last_chunk_size = self.LAST)
		make_image(self.dst, [0, 9], last_chunk_size = self.LAST)
		result = self.run_blocksync("-q", self.src, self.dst)
		self.assertEqual(result.returncode, 0, result.stderr)
		self.assert_synced()
		# destination file length itself must not change, only content
		self.assertEqual(os.path.getsize(self.dst), MB + self.LAST)

	def test_dst_bigger_multiple_of_chunk_size_last_chunk_identical(self):
		# src ends at 1.5 MB, dst is a clean 2 MB (2 full chunks); src's last chunk already matches
		# the corresponding bytes at the front of dst's second chunk
		make_image(self.src, [0, 1], last_chunk_size = self.LAST)
		with open(self.dst, "wb") as file_handle:
			file_handle.write(chunk_bytes(0, 0, MB))
			file_handle.write(chunk_bytes(1, 1, self.LAST)) # matches src's partial last chunk...
			file_handle.write(b"\xAA" * (MB - self.LAST))   # ...padded out to a full 2nd chunk
		tail_before = read_bytes(self.dst)[MB + self.LAST:]
		result = self.run_blocksync("-v", self.src, self.dst)
		self.assertEqual(result.returncode, 0, result.stderr)
		body = read_bytes(self.dst)
		self.assertEqual(body[:MB + self.LAST], read_bytes(self.src), "matching partial chunk was rewritten")
		self.assertEqual(body[MB + self.LAST:], tail_before, "bytes past source length were touched")
		self.assertIn("different: 0", result.stdout)

	def test_dst_bigger_multiple_of_chunk_size_last_chunk_different(self):
		# src ends at 1.5 MB, dst is a clean 2 MB; src's last chunk differs from dst's corresponding
		# bytes, so only the first LAST bytes of dst's second chunk should change
		make_image(self.src, [0, 1], last_chunk_size = self.LAST)
		with open(self.dst, "wb") as file_handle:
			file_handle.write(chunk_bytes(0, 0, MB))
			file_handle.write(chunk_bytes(1, 9, self.LAST))  # differs from src's partial last chunk
			file_handle.write(b"\xAA" * (MB - self.LAST))    # padding, must survive untouched
		tail_before = read_bytes(self.dst)[MB + self.LAST:]
		result = self.run_blocksync("-q", self.src, self.dst)
		self.assertEqual(result.returncode, 0, result.stderr)
		body = read_bytes(self.dst)
		self.assertEqual(body[:MB + self.LAST], read_bytes(self.src))
		self.assertEqual(body[MB + self.LAST:], tail_before, "bytes past source length were touched")

	def test_dst_bigger_non_multiple_of_chunk_size_last_chunk_identical(self):
		# dst is bigger than src by an amount that is itself not a multiple of chunk_size
		make_image(self.src, [0, 1], last_chunk_size = self.LAST)
		with open(self.dst, "wb") as file_handle:
			file_handle.write(chunk_bytes(0, 0, MB))
			file_handle.write(chunk_bytes(1, 1, self.LAST)) # matches src
			file_handle.write(b"\xBB" * 12345)               # odd-sized tail past the source
		tail_before = read_bytes(self.dst)[MB + self.LAST:]
		result = self.run_blocksync("-v", self.src, self.dst)
		self.assertEqual(result.returncode, 0, result.stderr)
		body = read_bytes(self.dst)
		self.assertEqual(body[:MB + self.LAST], read_bytes(self.src))
		self.assertEqual(body[MB + self.LAST:], tail_before)
		self.assertIn("different: 0", result.stdout)

	def test_dst_bigger_non_multiple_of_chunk_size_last_chunk_different(self):
		make_image(self.src, [0, 1], last_chunk_size = self.LAST)
		with open(self.dst, "wb") as file_handle:
			file_handle.write(chunk_bytes(0, 0, MB))
			file_handle.write(chunk_bytes(1, 9, self.LAST)) # differs from src
			file_handle.write(b"\xBB" * 12345)               # odd-sized tail past the source
		tail_before = read_bytes(self.dst)[MB + self.LAST:]
		result = self.run_blocksync("-q", self.src, self.dst)
		self.assertEqual(result.returncode, 0, result.stderr)
		body = read_bytes(self.dst)
		self.assertEqual(body[:MB + self.LAST], read_bytes(self.src))
		self.assertEqual(body[MB + self.LAST:], tail_before, "bytes past source length were touched")

	def test_last_chunk_smaller_than_one_disk_sector(self):
		# a tiny tail, well under a typical 512-byte or 4096-byte sector, still has to round-trip
		make_image(self.src, [0, 1], last_chunk_size = 100)
		make_image(self.dst, [0, 9], last_chunk_size = 100)
		result = self.run_blocksync("-q", self.src, self.dst)
		self.assertEqual(result.returncode, 0, result.stderr)
		self.assert_synced()
		self.assertEqual(os.path.getsize(self.dst), MB + 100)

	def test_second_run_copies_nothing(self):
		# idempotency has to hold for the partial chunk too, not just full ones (TestLocalSync
		# already covers the all-full-chunks case)
		make_image(self.src, [0, 1, 2], last_chunk_size = self.LAST)
		make_image(self.dst, [9, 1, 9], last_chunk_size = self.LAST)
		self.run_blocksync("-q", self.src, self.dst)
		result = self.run_blocksync("-v", self.src, self.dst)
		self.assertEqual(result.returncode, 0, result.stderr)
		self.assertIn("different: 0", result.stdout, "a repeat run should find no differences, including the last chunk")

	def test_dry_run_reports_partial_chunk_size_not_full_chunk_size(self):
		# only the last chunk differs and it is half a chunk; "Would copy" must reflect that, not
		# a rounded-up full chunk_size
		make_image(self.src, [0, 1], last_chunk_size = self.LAST)
		make_image(self.dst, [0, 9], last_chunk_size = self.LAST)
		before = digest(self.dst)
		result = self.run_blocksync("--dry-run", self.src, self.dst)
		self.assertEqual(result.returncode, 0, result.stderr)
		self.assertEqual(digest(self.dst), before, "--dry-run modified the destination")
		self.assertIn("Would copy 0 MB", result.stdout) # 512 KB rounds down to 0 MB, not up to 1

	def test_source_smaller_than_one_chunk(self):
		# the whole source is a single partial chunk, chunk 0 is simultaneously first and last;
		# make_image with a one-element variants list and last_chunk_size exercises exactly that
		make_image(self.src, [0], last_chunk_size = self.LAST)
		make_image(self.dst, [9], last_chunk_size = self.LAST)
		result = self.run_blocksync("-v", "-v", self.src, self.dst) # -v -v: debug, for "chunks total"
		self.assertEqual(result.returncode, 0, result.stderr)
		self.assert_synced()
		self.assertEqual(os.path.getsize(self.src), self.LAST)
		self.assertIn("src chunks total: 1", result.stdout)

	def test_source_smaller_than_one_chunk_already_matches(self):
		make_image(self.src, [0], last_chunk_size = self.LAST)
		make_image(self.dst, [0], last_chunk_size = self.LAST)
		before = digest(self.dst)
		result = self.run_blocksync("-v", self.src, self.dst)
		self.assertEqual(result.returncode, 0, result.stderr)
		self.assertEqual(digest(self.dst), before)
		self.assertIn("different: 0", result.stdout)

	def test_various_chunk_sizes_with_partial_last_chunk(self):
		for chunk_mb in (1, 2, 4):
			with self.subTest(chunk_size_mb = chunk_mb):
				chunk_size = chunk_mb * MB
				last = chunk_size // 3 # deliberately not a clean fraction of MB either
				make_image(self.src, [0, 1, 2], chunk_size, last_chunk_size = last)
				make_image(self.dst, [9, 1, 9], chunk_size, last_chunk_size = last)
				result = self.run_blocksync("-q", "-c", str(chunk_mb), self.src, self.dst)
				self.assertEqual(result.returncode, 0, result.stderr)
				self.assert_synced()


class TestDryRun(BlocksyncTestCase):
	"""
	--dry-run has to report differences without writing a single byte.
	"""

	def test_destination_is_untouched(self):
		make_image(self.src, [0, 1, 2, 3])
		make_image(self.dst, [9, 1, 9, 3])
		before = digest(self.dst)
		result = self.run_blocksync("-q", "-n", self.src, self.dst)
		self.assertEqual(result.returncode, 0, result.stderr)
		self.assertEqual(digest(self.dst), before, "--dry-run modified the destination")

	def test_reports_what_would_be_copied(self):
		make_image(self.src, [0, 1, 2, 3])
		make_image(self.dst, [9, 1, 9, 3])
		result = self.run_blocksync("--dry-run", self.src, self.dst)
		self.assertIn("Would copy 2 MB", result.stdout)

	@unittest.skipIf(os.geteuid() == 0, "running as root, permissions are not enforced")
	def test_works_on_a_read_only_destination(self):
		make_image(self.src, [0, 1])
		make_image(self.dst, [9, 1])
		os.chmod(self.dst, 0o444)
		result = self.run_blocksync("-q", "-n", self.src, self.dst)
		self.assertEqual(result.returncode, 0, result.stderr)


class TestErrorPaths(BlocksyncTestCase):
	"""
	Every failure has to be one readable line and a non-zero exit, never a stack trace.
	"""

	def test_missing_source(self):
		make_image(self.dst, [0])
		result = self.run_blocksync(os.path.join(self.dir, "nope.img"), self.dst)
		self.assert_no_traceback(result)
		self.assertEqual(result.returncode, 1)
		self.assertIn("No such file", result.stderr)

	def test_missing_destination(self):
		make_image(self.src, [0])
		result = self.run_blocksync(self.src, os.path.join(self.dir, "nope.img"))
		self.assert_no_traceback(result)
		self.assertEqual(result.returncode, 1)
		self.assertIn("No such file", result.stderr)

	@unittest.skipIf(os.geteuid() == 0, "running as root, permissions are not enforced")
	def test_unwritable_destination(self):
		make_image(self.src, [0])
		make_image(self.dst, [9])
		os.chmod(self.dst, 0o444)
		result = self.run_blocksync(self.src, self.dst)
		self.assert_no_traceback(result)
		self.assertEqual(result.returncode, 1)
		self.assertIn("Permission denied", result.stderr)

	def test_destination_smaller_than_source(self):
		make_image(self.src, [0, 1, 2])
		make_image(self.dst, [0])
		result = self.run_blocksync(self.src, self.dst)
		self.assert_no_traceback(result)
		self.assertEqual(result.returncode, 1)
		self.assertIn("bigger than destination", result.stderr)

	def test_target_with_several_colons_does_not_crash(self):
		# an IPv6 address used to hit "too many values to unpack"
		make_image(self.dst, [0])
		result = self.run_blocksync("user@2001:db8::1:/dev/sdb", self.dst,
				env = self.fake_ssh_env())
		self.assert_no_traceback(result)
		self.assertEqual(result.returncode, 1)

	def test_usage_line_is_rendered(self):
		result = self.run_blocksync("--help")
		self.assertEqual(result.returncode, 0)
		self.assertNotIn("%prog", result.stdout, "optparse placeholder leaked into the help text")
		self.assertNotIn("%(prog)s", result.stdout)


class TestRemote(BlocksyncTestCase):
	"""
	The remote paths, driven through the stand-in ssh.
	"""

	def setUp(self):
		super().setUp()
		self.env = self.fake_ssh_env()

	def test_pull_from_remote(self):
		make_image(self.src, [0, 1, 2, 3])
		make_image(self.dst, [9, 1, 2, 9])
		result = self.run_blocksync("-q", f"host:{self.src}", self.dst, env = self.env)
		self.assertEqual(result.returncode, 0, result.stderr)
		self.assert_synced()

	def test_push_to_remote(self):
		make_image(self.src, [0, 1, 2, 3])
		make_image(self.dst, [9, 1, 2, 9])
		result = self.run_blocksync("-q", self.src, f"host:{self.dst}", env = self.env)
		self.assertEqual(result.returncode, 0, result.stderr)
		self.assert_synced()

	def test_remote_to_remote(self):
		make_image(self.src, [0, 1, 2, 3])
		make_image(self.dst, [9, 1, 2, 9])
		result = self.run_blocksync("-q", f"host:{self.src}", f"host:{self.dst}", env = self.env)
		self.assertEqual(result.returncode, 0, result.stderr)
		self.assert_synced()

	def test_local_path_containing_a_colon(self):
		# a slash before the colon means local, as in rsync and scp
		local = os.path.join(self.dir, "back-up:1.img")
		make_image(local, [0, 1])
		make_image(self.dst, [9, 1])
		result = self.run_blocksync("-q", local, self.dst, env = self.env)
		self.assertEqual(result.returncode, 0, result.stderr)
		self.assertEqual(digest(self.dst), digest(local))

	def test_apostrophe_in_worker_source_survives_quoting(self):
		"""
		The worker is shipped as source text. It used to be wrapped in triple single quotes, so one
		apostrophe anywhere in the function broke remote synchronization with a shell error.
		"""

		patched = os.path.join(self.dir, "patched-blocksync.py")
		source = read_bytes(BLOCKSYNC).decode()
		marker = "\t\t# advance to next chunk"
		self.assertIn(marker, source, "the comment this test patches has moved")
		with open(patched, "w") as file_handle:
			file_handle.write(source.replace(marker, marker + ", don't skip any"))

		make_image(self.src, [0, 1])
		make_image(self.dst, [9, 1])
		result = subprocess.run([sys.executable, patched, "-q", f"host:{self.src}", self.dst],
				capture_output = True, text = True, env = self.env, timeout = 120)
		self.assertEqual(result.returncode, 0, result.stderr)
		self.assert_synced()

	def test_chatty_ssh_does_not_hang(self):
		"""
		Nothing reads stderr during a transfer, so a worker or ssh that fills the pipe buffer used to
		block on its own write and hang the run for good. A login banner is enough.
		"""

		make_image(self.src, [0, 1, 2, 3])
		make_image(self.dst, [9, 1, 2, 9])
		env = self.fake_ssh_env(FAKE_SSH_STDERR_NOISE = 5 * MB)
		try:
			result = self.run_blocksync("-q", f"host:{self.src}", self.dst, env = env, timeout = 60)
		except subprocess.TimeoutExpired:
			self.fail("blocksync deadlocked on a full stderr pipe")
		self.assertEqual(result.returncode, 0, result.stderr[:400])
		self.assert_synced()


class TestTopologies(BlocksyncTestCase):
	"""
	Source and destination can each independently be local or remote, four combinations in total.
	TestRemote already checks pull / push / remote-to-remote error-free happy paths; this class
	instead runs the *same* handful of scenarios, including a partial last chunk, across all four
	combinations, so a bug that only shows up in one direction (for example, in whichever side of a
	transfer sends versus receives) cannot hide behind the other three passing.
	"""

	def setUp(self):
		super().setUp()
		self.env = self.fake_ssh_env()

	def sync(self, src_remote, dst_remote):
		"""
		Run blocksync with src and/or dst prefixed with "host:" as requested, using the stand-in
		ssh. Every combination goes through the same env, so a local leg in a mixed pair still has
		the fake ssh on PATH even though it will not be invoked for that leg.
		"""

		src = f"host:{self.src}" if src_remote else self.src
		dst = f"host:{self.dst}" if dst_remote else self.dst
		return self.run_blocksync("-q", src, dst, env = self.env)

	def test_matrix_full_chunks_differ(self):
		for src_remote, dst_remote in ((False, False), (False, True), (True, False), (True, True)):
			with self.subTest(src_remote = src_remote, dst_remote = dst_remote):
				make_image(self.src, [0, 1, 2, 3])
				make_image(self.dst, [9, 1, 9, 3])
				result = self.sync(src_remote, dst_remote)
				self.assertEqual(result.returncode, 0, result.stderr)
				self.assert_synced()

	def test_matrix_partial_last_chunk(self):
		# the same partial-last-chunk scenario as TestPartialLastChunk.
		# test_src_dst_same_size_last_chunk_different, run over all four topologies, so the worker
		# side of the src_size fix is exercised through the ssh path too, not only locally
		last = 512 * 1024
		for src_remote, dst_remote in ((False, False), (False, True), (True, False), (True, True)):
			with self.subTest(src_remote = src_remote, dst_remote = dst_remote):
				make_image(self.src, [0, 1], last_chunk_size = last)
				make_image(self.dst, [0, 9], last_chunk_size = last)
				result = self.sync(src_remote, dst_remote)
				self.assertEqual(result.returncode, 0, result.stderr)
				self.assert_synced()
				self.assertEqual(os.path.getsize(self.dst), MB + last)

	def test_matrix_dst_bigger_than_src(self):
		# a destination bigger than the source, checked in all four topologies, since only the dst
		# side needs to tolerate reading/writing less than its own full size
		for src_remote, dst_remote in ((False, False), (False, True), (True, False), (True, True)):
			with self.subTest(src_remote = src_remote, dst_remote = dst_remote):
				make_image(self.src, [0, 1])
				make_image(self.dst, [9, 9, 5, 6])
				tail_before = read_bytes(self.dst)[2 * MB:]
				result = self.sync(src_remote, dst_remote)
				self.assertEqual(result.returncode, 0, result.stderr)
				body = read_bytes(self.dst)
				self.assertEqual(body[:2 * MB], read_bytes(self.src))
				self.assertEqual(body[2 * MB:], tail_before, "data past the source length was modified")


class TestDifferential(BlocksyncTestCase):
	"""
	Randomised comparison against the obvious ground truth: after a run the destination must equal
	the source, byte for byte, with everything past the source length untouched.
	"""

	def test_random_layouts(self):
		rnd = random.Random(20260822)
		for trial in range(25):
			with self.subTest(trial = trial):
				size_mb = rnd.choice([1, 1, 1, 2, 4])
				chunk_size = size_mb * MB
				count = rnd.randint(0, 6)
				extra = rnd.choice([0, 0, 0, 1])
				# variant 0 means "same as source", anything else differs
				variants = [rnd.choice([0, 0, 7]) for _ in range(count)]
				# sometimes shorten the source's last chunk, so the fuzz test also covers a source
				# size that is not an exact multiple of chunk_size, not just full-chunk layouts.
				# Meaningless with an empty source, so only applied when there is a last chunk.
				last = rnd.choice([None, None, None, chunk_size // 3]) if count else None
				last_size = last if last is not None else chunk_size
				src_size = (count - 1) * chunk_size + last_size if count else 0
				make_image(self.src, [0] * count, chunk_size, last_chunk_size = last)

				# built by hand rather than through make_image's own last_chunk_size, because dst
				# can have more chunks than src (extra > 0): the chunk that must be shortened to
				# line up with src's last chunk is at index count - 1, not dst's own last index.
				with open(self.dst, "wb") as file_handle:
					for index in range(count):
						size = last_size if index == count - 1 else chunk_size
						file_handle.write(chunk_bytes(index, variants[index], size))
					for index in range(extra):
						file_handle.write(chunk_bytes(count + index, 5, chunk_size))

				before = read_bytes(self.dst)
				tail = before[src_size:]

				result = self.run_blocksync("-q", "-n", "-c", str(size_mb), self.src, self.dst)
				self.assertEqual(result.returncode, 0, result.stderr)
				self.assertEqual(read_bytes(self.dst), before, "--dry-run wrote to the destination")

				result = self.run_blocksync("-q", "-c", str(size_mb), self.src, self.dst)
				self.assertEqual(result.returncode, 0, result.stderr)
				body = read_bytes(self.dst)
				self.assertEqual(body[:src_size], read_bytes(self.src))
				self.assertEqual(body[src_size:], tail)


@unittest.skipUnless(os.environ.get("BLOCKSYNC_BENCH"), "set BLOCKSYNC_BENCH=1 to run the benchmark")
class TestLatency(BlocksyncTestCase):
	"""
	Not a pass or fail test, it prints the round trip cost. Chunks are compared one at a time and a
	worker cannot hash chunk N+1 before the command for chunk N arrives, so every chunk costs a full
	round trip. Bigger chunks buy that back, at the price of transferring more per change.
	"""

	def test_report_round_trip_cost(self):
		make_image(self.src, [0] * 64)
		print("\n  64 MB scanned, 2 chunks different\n")
		print(f"  {'round trip':>12} {'1 MB':>10} {'4 MB':>10} {'16 MB':>10}")
		for delay in (0.0005, 0.005):
			row = []
			for size_mb in (1, 4, 16):
				make_image(self.dst, [0] * 5 + [9] + [0] * 3 + [9] + [0] * 54)
				env = self.fake_ssh_env(FAKE_SSH_DELAY = delay)
				# read blocksync's own figure rather than wall clock, so interpreter startup does
				# not get counted and the numbers line up with what the tool reports in the field
				result = self.run_blocksync("-c", str(size_mb),
						f"host:{self.src}", self.dst, env = env)
				self.assertEqual(result.returncode, 0, result.stderr)
				self.assert_synced()
				reported = re.search(r" in (\d+):(\d\d):(\d\d(?:\.\d+)?),", result.stdout)
				self.assertIsNotNone(reported, f"could not read the duration from: {result.stdout!r}")
				hours, minutes, seconds = reported.groups()
				elapsed = int(hours) * 3600 + int(minutes) * 60 + float(seconds)
				row.append(f"{elapsed:.3f}s")
			print(f"  {int(delay * 2000):>9} ms {row[0]:>10} {row[1]:>10} {row[2]:>10}")
		print()


if __name__ == "__main__":
	if not os.path.exists(BLOCKSYNC):
		sys.exit(f"cannot find {BLOCKSYNC}")
	unittest.main(verbosity = 2)