1 files changed, 121 insertions, 0 deletions

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