Add integration tests for live sync

3 files changed, 172 insertions, 3 deletions

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))