From 0fcf290ba2db320287163f142a83da0f9fe48b77 Mon Sep 17 00:00:00 2001 From: Dirk Engling Date: Tue, 3 Nov 2026 02:40:38 +0100 Subject: Add integration tests for live sync --- Mk/tests.mk | 20 +++++- tests/integration/harness.py | 34 ++++++++++- tests/integration/test_livesync.py | 121 +++++++++++++++++++++++++++++++++++++ 3 files changed, 172 insertions(+), 3 deletions(-) create mode 100644 tests/integration/test_livesync.py diff --git a/Mk/tests.mk b/Mk/tests.mk index 46f308b..a7d4fbc 100644 --- a/Mk/tests.mk +++ b/Mk/tests.mk @@ -29,7 +29,13 @@ LDFLAGS_integration=$(LINK) -fsanitize=address,undefined CFLAGS_tsan=$(OPTS_debug) $(CPPFLAGS) $(OT_CFLAGS) $(FEATURES_integration) -fsanitize=thread -fno-omit-frame-pointer LDFLAGS_tsan=$(LINK) -fsanitize=thread -BINARIES+=$(TESTBIN) $(TESTBIN).poll $(TESTBIN).sanitize opentracker.integration opentracker.tsan +# Live sync is a compile time feature, so reaching it from a black box test +# needs a binary of its own. tests/integration/test_livesync.py skips itself +# against a build without it, which is what -v is for. +FEATURES_livesync=$(FEATURES_integration) -DWANT_SYNC_LIVE +CFLAGS_livesync=$(OPTS_debug) $(CPPFLAGS) $(OT_CFLAGS) $(FEATURES_livesync) -fsanitize=address,undefined -fno-omit-frame-pointer + +BINARIES+=$(TESTBIN) $(TESTBIN).poll $(TESTBIN).sanitize opentracker.integration opentracker.tsan opentracker.livesync $(TESTBIN): $(TEST_SOURCES) $(TEST_UNITS) $(TEST_HEADERS) $(BUILDDEPS) $(CC) -o $@ $(CFLAGS_test) $(TEST_SOURCES) $(TEST_UNITS) @@ -46,6 +52,9 @@ opentracker.integration: $(SOURCES) $(HEADERS) $(BUILDDEPS) opentracker.tsan: $(SOURCES) $(HEADERS) $(BUILDDEPS) $(CC) -o $@ $(CFLAGS_tsan) $(SOURCES) $(LDFLAGS_tsan) +opentracker.livesync: $(SOURCES) $(HEADERS) $(BUILDDEPS) + $(CC) -o $@ $(CFLAGS_livesync) $(SOURCES) $(LDFLAGS_integration) + check: $(TESTBIN) $(BINARY) ./$(TESTBIN) @@ -61,12 +70,19 @@ check-integration: opentracker.integration check-tsan: opentracker.tsan OPENTRACKER=./opentracker.tsan $(PYTHON) -m unittest discover -s tests/integration +# Only the live sync tests: a build with WANT_SYNC_LIVE refuses to start +# without a live sync address, so it cannot stand in for the plain binary in +# the rest of the suite. +check-livesync: opentracker.livesync + OPENTRACKER=./opentracker.livesync $(PYTHON) -m unittest discover -s tests/integration -p 'test_livesync.py' + # Cheapest first, so a quick failure comes back quickly. check-all: $(MAKE) check $(MAKE) check-poll $(MAKE) check-sanitize $(MAKE) check-integration + $(MAKE) check-livesync $(MAKE) check-tsan -.PHONY: check check-poll check-sanitize check-integration check-tsan check-all +.PHONY: check check-poll check-sanitize check-integration check-livesync check-tsan check-all diff --git a/tests/integration/harness.py b/tests/integration/harness.py index b98ce83..a9d38ca 100644 --- a/tests/integration/harness.py +++ b/tests/integration/harness.py @@ -20,6 +20,9 @@ import urllib.error import urllib.parse import urllib.request +# groupip_1 in src/ot_livesync.c +LIVESYNC_GROUP = "224.0.23.5" + TRACKER = os.environ.get("OPENTRACKER", os.path.join(os.path.dirname(__file__), "..", "..", "opentracker.debug")) @@ -76,6 +79,9 @@ class Tracker: def __init__(self, extra_config="", files=None): self.port = free_port() + # Only used by a config that asks for live sync, but allocated either + # way so {livesync_port} always resolves. + self.livesync_port = free_port() self.dir = tempfile.mkdtemp(prefix="ot-integration-") self.log = None self.process = None @@ -87,7 +93,7 @@ class Tracker: with open(self.config, "w") as handle: handle.write("listen.tcp 127.0.0.1:%d\n" % self.port) handle.write("listen.udp 127.0.0.1:%d\n" % self.port) - handle.write(extra_config.replace("{dir}", self.dir)) + handle.write(extra_config.replace("{dir}", self.dir).replace("{livesync_port}", str(self.livesync_port))) self.log = open(os.path.join(self.dir, "tracker.log"), "w+b") # Leak detection is left at the sanitizer's own default, which is on # where LeakSanitizer exists and off where it does not. Asking for it @@ -186,6 +192,32 @@ class Tracker: def torrent_count(self): return int(self.stats("torr").split()[0]) + def livesync_send(self, payload): + """Put a packet on the live sync group. + + It has to go to the group rather than straight at the port: the + tracker binds two sockets there, the reader on the wildcard address + and the writer on the configured one, and the kernel gives a unicast + datagram to the more specific of the two, which nobody reads.""" + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + try: + sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_IF, socket.inet_aton("127.0.0.1")) + sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_LOOP, 1) + sock.setsockopt(socket.IPPROTO_IP, socket.IP_MULTICAST_TTL, 1) + sock.sendto(payload, (LIVESYNC_GROUP, self.livesync_port)) + finally: + sock.close() + + def wait_until(self, predicate, timeout=5): + """A live sync packet is handled by another thread, so an effect of + one arrives some time after the send.""" + deadline = time.time() + timeout + while time.time() < deadline: + if predicate(): + return True + time.sleep(0.02) + return predicate() + # ------------------------------------------------------------ over udp def udp_connect(self): diff --git a/tests/integration/test_livesync.py b/tests/integration/test_livesync.py new file mode 100644 index 0000000..8dd754b --- /dev/null +++ b/tests/integration/test_livesync.py @@ -0,0 +1,121 @@ +"""Live sync, driven over the multicast group a cluster would use. + +Live sync sits behind WANT_SYNC_LIVE, so these only run against a binary +built with it: make check-livesync. Against any other build they skip, which +is what the feature list in -v is for. +""" + +import socket +import struct +import subprocess +import unittest + +from harness import TRACKER, TrackerTestCase + +# enum { OT_SYNC_PEER4, OT_SYNC_PEER6 } in src/ot_livesync.c +SYNC_PEER4 = 0 +SYNC_PEER6 = 1 + +# src/trackerlogic.h +FLAG_SEEDING = 0x80 +FLAG_COMPLETED = 0x40 +FLAG_STOPPED = 0x20 + + +def has_livesync(): + done = subprocess.run([TRACKER, "-v"], capture_output=True, timeout=30) + return b"WANT_SYNC_LIVE" in done.stderr + + +def peer4(address, port, flags): + """Four bytes of address, two of port, one of flags, one the tracker + overwrites with its own idea of the time.""" + return socket.inet_aton(address) + struct.pack("!H", port) + bytes([flags, 0]) + + +def peer6(address, port, flags): + return socket.inet_pton(socket.AF_INET6, address) + struct.pack("!H", port) + bytes([flags, 0]) + + +def packet(kind, entries, tracker_id=0xDEADBEEF): + body = b"".join(info_hash + peer for info_hash, peer in entries) + return struct.pack("!I", tracker_id) + struct.pack("!I", kind) + body + + +@unittest.skipUnless(has_livesync(), "this build has no WANT_SYNC_LIVE") +class LiveSyncIn(TrackerTestCase): + config = "livesync.cluster.listen 127.0.0.1:{livesync_port}\nlivesync.cluster.node_ip 127.0.0.1\n" + + def counts(self, info_hash): + """Seeders and leechers a scrape reports for one torrent.""" + files = self.tracker.scrape(info_hash)[b"files"] + if info_hash not in files: + return None + entry = files[info_hash] + return entry[b"complete"], entry[b"incomplete"] + + def test_a_synced_seeder_arrives(self): + info_hash = self.hash_for("sync-seed") + self.tracker.livesync_send(packet(SYNC_PEER4, [(info_hash, peer4("10.1.2.3", 6881, FLAG_SEEDING))])) + self.assertTrue(self.tracker.wait_until(lambda: self.counts(info_hash) == (1, 0)), + "scrape says %r" % (self.counts(info_hash),)) + + def test_a_synced_leecher_arrives(self): + info_hash = self.hash_for("sync-leech") + self.tracker.livesync_send(packet(SYNC_PEER4, [(info_hash, peer4("10.1.2.4", 6881, 0))])) + self.assertTrue(self.tracker.wait_until(lambda: self.counts(info_hash) == (0, 1)), + "scrape says %r" % (self.counts(info_hash),)) + + def test_a_v6_peer_arrives(self): + info_hash = self.hash_for("sync-v6") + self.tracker.livesync_send(packet(SYNC_PEER6, [(info_hash, peer6("2001:db8::1", 6881, FLAG_SEEDING))])) + self.assertTrue(self.tracker.wait_until(lambda: self.counts(info_hash) == (1, 0)), + "scrape says %r" % (self.counts(info_hash),)) + + def test_one_packet_carries_several_torrents(self): + hashes = [self.hash_for("sync-many-%d" % i) for i in range(4)] + self.tracker.livesync_send(packet(SYNC_PEER4, [(h, peer4("10.1.2.%d" % (5 + i), 6881, FLAG_SEEDING)) + for i, h in enumerate(hashes)])) + self.assertTrue(self.tracker.wait_until(lambda: all(self.counts(h) == (1, 0) for h in hashes))) + + def test_a_stopped_peer_is_taken_away_again(self): + info_hash = self.hash_for("sync-stop") + peer = peer4("10.1.2.9", 6881, FLAG_SEEDING) + self.tracker.livesync_send(packet(SYNC_PEER4, [(info_hash, peer)])) + self.assertTrue(self.tracker.wait_until(lambda: self.counts(info_hash) == (1, 0))) + + self.tracker.livesync_send(packet(SYNC_PEER4, [(info_hash, peer4("10.1.2.9", 6881, FLAG_STOPPED))])) + self.assertTrue(self.tracker.wait_until(lambda: self.counts(info_hash) in (None, (0, 0))), + "scrape still says %r" % (self.counts(info_hash),)) + + def test_a_truncated_entry_is_dropped(self): + """The loop takes whole hash and peer pairs and leaves a short tail + where it is, so the complete entry before it still lands.""" + good, short = self.hash_for("sync-good"), self.hash_for("sync-short") + payload = packet(SYNC_PEER4, [(good, peer4("10.1.2.10", 6881, FLAG_SEEDING))]) + short + b"\x0a\x01" + self.tracker.livesync_send(payload) + self.assertTrue(self.tracker.wait_until(lambda: self.counts(good) == (1, 0))) + self.assertIsNone(self.counts(short)) + + def test_an_unknown_packet_type_is_ignored(self): + info_hash = self.hash_for("sync-unknown") + self.tracker.livesync_send(packet(99, [(info_hash, peer4("10.1.2.11", 6881, FLAG_SEEDING))])) + self.assertFalse(self.tracker.wait_until(lambda: self.counts(info_hash) is not None, timeout=1)) + + def test_a_packet_with_nothing_but_a_header_is_ignored(self): + before = self.tracker.torrent_count() + self.tracker.livesync_send(struct.pack("!I", 0xDEADBEEF) + struct.pack("!I", SYNC_PEER4)) + self.assertFalse(self.tracker.wait_until(lambda: self.tracker.torrent_count() != before, timeout=1)) + + +@unittest.skipUnless(has_livesync(), "this build has no WANT_SYNC_LIVE") +class LiveSyncAccess(TrackerTestCase): + """Nothing blesses 127.0.0.1 here, so the tracker should take nothing + from it. Source address checking is all live sync has.""" + + config = "livesync.cluster.listen 127.0.0.1:{livesync_port}\nlivesync.cluster.node_ip 10.0.0.1\n" + + def test_an_unblessed_sender_is_ignored(self): + info_hash = self.hash_for("sync-unblessed") + self.tracker.livesync_send(packet(SYNC_PEER4, [(info_hash, peer4("10.1.2.12", 6881, FLAG_SEEDING))])) + self.assertFalse(self.tracker.wait_until(lambda: self.tracker.torrent_count() != 0, timeout=1)) -- cgit v1.3