From 687ef7fece04fb5dffa90b5d784fb1e34066c85a Mon Sep 17 00:00:00 2001 From: Dirk Engling Date: Mon, 26 Oct 2026 23:08:23 +0100 Subject: Add integration test and its runner --- tests/integration/.gitignore | 2 + tests/integration/harness.py | 166 ++++++++++++++++++++++++++++++++++++ tests/integration/test_announce.py | 129 ++++++++++++++++++++++++++++ tests/integration/test_dualstack.py | 89 +++++++++++++++++++ 4 files changed, 386 insertions(+) create mode 100644 tests/integration/.gitignore create mode 100644 tests/integration/harness.py create mode 100644 tests/integration/test_announce.py create mode 100644 tests/integration/test_dualstack.py (limited to 'tests') diff --git a/tests/integration/.gitignore b/tests/integration/.gitignore new file mode 100644 index 0000000..7a60b85 --- /dev/null +++ b/tests/integration/.gitignore @@ -0,0 +1,2 @@ +__pycache__/ +*.pyc diff --git a/tests/integration/harness.py b/tests/integration/harness.py new file mode 100644 index 0000000..c8da1e2 --- /dev/null +++ b/tests/integration/harness.py @@ -0,0 +1,166 @@ +"""Starts a real opentracker and talks to it over the wire. + +Everything here drives the binary rather than the code, so these tests cover +what the unit suite cannot reach: the event loop, the HTTP parser and the +torrent store working together. Run against a sanitizer build and the tracker +reports its own memory errors, which the teardown turns into test failures. +""" + +import os +import socket +import struct +import subprocess +import sys +import tempfile +import time +import unittest +import urllib.error +import urllib.parse +import urllib.request + +TRACKER = os.environ.get("OPENTRACKER", os.path.join(os.path.dirname(__file__), "..", "..", "opentracker.debug")) + + +def bdecode(data): + """Enough bencode for what a tracker answers.""" + + def value(i): + if data[i : i + 1] == b"d": + i, out = i + 1, {} + while data[i : i + 1] != b"e": + i, key = value(i) + i, out[key] = value(i) + return i + 1, out + if data[i : i + 1] == b"l": + i, out = i + 1, [] + while data[i : i + 1] != b"e": + i, item = value(i) + out.append(item) + return i + 1, out + if data[i : i + 1] == b"i": + end = data.index(b"e", i) + return end + 1, int(data[i + 1 : end]) + colon = data.index(b":", i) + length = int(data[i:colon]) + return colon + 1 + length, data[colon + 1 : colon + 1 + length] + + return value(0)[1] + + +def free_port(): + probe = socket.socket() + probe.bind(("127.0.0.1", 0)) + port = probe.getsockname()[1] + probe.close() + return port + + +class Tracker: + """One tracker process, listening on loopback.""" + + def __init__(self, extra_config=""): + self.port = free_port() + self.dir = tempfile.mkdtemp(prefix="ot-integration-") + self.config = os.path.join(self.dir, "opentracker.conf") + 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) + self.log = open(os.path.join(self.dir, "tracker.log"), "w+b") + environment = dict(os.environ, ASAN_OPTIONS="detect_leaks=0:abort_on_error=0") + self.process = subprocess.Popen([TRACKER, "-f", self.config], stdout=self.log, stderr=self.log, env=environment) + self._wait_for_port() + + def _wait_for_port(self): + for _ in range(200): + if self.process.poll() is not None: + raise AssertionError("tracker exited at once:\n" + self.output()) + try: + probe = socket.create_connection(("127.0.0.1", self.port), timeout=0.2) + probe.close() + return + except OSError: + time.sleep(0.05) + raise AssertionError("tracker never listened on %d:\n%s" % (self.port, self.output())) + + def output(self): + self.log.flush() + self.log.seek(0) + return self.log.read().decode("utf-8", "replace") + + def stop(self): + """Returns anything the tracker complained about on the way out.""" + if self.process.poll() is None: + self.process.kill() + self.process.wait() + complaints = [ + line + for line in self.output().splitlines() + if "Sanitizer" in line or "runtime error" in line or line.startswith("ERROR") + ] + self.log.close() + return complaints + + # ----------------------------------------------------------- over http + + def get(self, path, timeout=10): + return urllib.request.urlopen("http://127.0.0.1:%d%s" % (self.port, path), timeout=timeout).read() + + def announce(self, info_hash, peer_id=b"-OT0001-aaaaaaaaaaaa", port=6881, left=0, **extra): + query = { + "info_hash": urllib.parse.quote_from_bytes(info_hash, safe=""), + "peer_id": urllib.parse.quote_from_bytes(peer_id, safe=""), + "port": port, + "left": left, + } + query.update(extra) + return bdecode(self.get("/announce?" + "&".join("%s=%s" % item for item in query.items()))) + + def scrape(self, info_hash): + return bdecode(self.get("/scrape?info_hash=" + urllib.parse.quote_from_bytes(info_hash, safe=""))) + + def stats(self, mode): + return self.get("/stats?mode=" + mode) + + def torrent_count(self): + return int(self.stats("torr").split()[0]) + + # ------------------------------------------------------------ over udp + + def udp_connect(self): + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + sock.settimeout(5) + sock.sendto(struct.pack(">QII", 0x41727101980, 0, 1), ("127.0.0.1", self.port)) + answer = sock.recv(2048) + action, _, connection_id = struct.unpack(">IIQ", answer[:16]) + assert action == 0, "udp connect answered action %d" % action + return sock, connection_id + + def udp_announce(self, sock, connection_id, info_hash, left=0, event=0, port=6881, numwant=-1): + sock.sendto( + struct.pack( + ">QII20s20sQQQIIIiH", connection_id, 1, 2, info_hash, b"p" * 20, 0, left, 0, event, 0, 0, numwant, port + ), + ("127.0.0.1", self.port), + ) + answer = sock.recv(2048) + action, _, interval, leechers, seeders = struct.unpack(">IIIII", answer[:20]) + assert action == 1, "udp announce answered action %d" % action + return {"interval": interval, "leechers": leechers, "seeders": seeders, "peers": answer[20:]} + + +class TrackerTestCase(unittest.TestCase): + """Gives every test its own tracker and fails it on any sanitizer output.""" + + config = "" + + def setUp(self): + self.tracker = Tracker(self.config) + + def tearDown(self): + complaints = self.tracker.stop() + self.assertEqual(complaints, [], "the tracker reported:\n" + "\n".join(complaints)) + + def hash_for(self, name): + """A distinct info hash per test, so tests cannot leak into each other.""" + return (name.encode() + b"\0" * 20)[:20] diff --git a/tests/integration/test_announce.py b/tests/integration/test_announce.py new file mode 100644 index 0000000..ffbc07b --- /dev/null +++ b/tests/integration/test_announce.py @@ -0,0 +1,129 @@ +"""What a peer's announce does to the swarm it joins.""" + +import os +import socket +import sys +import unittest + +sys.path.insert(0, os.path.dirname(__file__)) +from harness import TrackerTestCase + + +def peers_of(answer, key=b"peers"): + """Compact peer list as a list of (address, port).""" + blob, out = answer.get(key, b""), [] + size = 6 if key == b"peers" else 18 + for offset in range(0, len(blob), size): + entry = blob[offset : offset + size] + address = socket.inet_ntop(socket.AF_INET if size == 6 else socket.AF_INET6, entry[:-2]) + out.append((address, entry[-2] << 8 | entry[-1])) + return out + + +class AnnounceLifetime(TrackerTestCase): + config = "" + + def test_a_leecher_is_counted_as_incomplete(self): + info_hash = self.hash_for("leecher") + answer = self.tracker.announce(info_hash, left=100) + self.assertEqual(answer[b"complete"], 0) + self.assertEqual(answer[b"incomplete"], 1) + + def test_no_bytes_left_means_seeding(self): + info_hash = self.hash_for("seeder") + answer = self.tracker.announce(info_hash, left=0) + self.assertEqual(answer[b"complete"], 1) + self.assertEqual(answer[b"incomplete"], 0) + + def test_a_peer_that_finishes_moves_from_one_column_to_the_other(self): + info_hash = self.hash_for("finishes") + peer_id = b"-OT0001-finisher0001" + self.tracker.announce(info_hash, peer_id=peer_id, left=100) + self.assertEqual(self.tracker.announce(info_hash, peer_id=peer_id, left=100)[b"incomplete"], 1) + + answer = self.tracker.announce(info_hash, peer_id=peer_id, left=0, event="completed") + self.assertEqual(answer[b"complete"], 1, "the peer should now count as a seeder") + self.assertEqual(answer[b"incomplete"], 0, "and no longer as a leecher") + + def test_stopped_removes_the_peer(self): + info_hash = self.hash_for("stopped") + peer_id = b"-OT0001-leaves000001" + self.tracker.announce(info_hash, peer_id=peer_id, left=100) + self.assertEqual(self.tracker.scrape(info_hash)[b"files"][info_hash][b"incomplete"], 1) + + self.tracker.announce(info_hash, peer_id=peer_id, left=100, event="stopped") + self.assertEqual( + self.tracker.scrape(info_hash)[b"files"][info_hash][b"incomplete"], 0, "the peer should be gone" + ) + + def test_the_same_peer_announcing_twice_is_still_one_peer(self): + info_hash = self.hash_for("twice") + peer_id = b"-OT0001-repeats00001" + for _ in range(5): + answer = self.tracker.announce(info_hash, peer_id=peer_id, left=100) + self.assertEqual(answer[b"incomplete"], 1, "re-announcing must not duplicate the peer") + + def test_a_peer_is_its_address_and_port_not_its_peer_id(self): + """Two announces from one address and port are one peer, whatever + peer_id they carry, because that pair is the whole identity.""" + info_hash = self.hash_for("identity") + self.tracker.announce(info_hash, peer_id=b"-OT0001-a00000000001", port=6881, left=100) + answer = self.tracker.announce(info_hash, peer_id=b"-OT0001-b00000000002", port=6881, left=100) + self.assertEqual(answer[b"incomplete"], 1, "same address and port, so still one peer") + + answer = self.tracker.announce(info_hash, peer_id=b"-OT0001-c00000000003", port=6882, left=100) + self.assertEqual(answer[b"incomplete"], 2, "a different port makes a different peer") + + def test_a_downloaded_count_is_kept(self): + info_hash = self.hash_for("downloaded") + self.tracker.announce(info_hash, port=7001, left=0, event="completed") + self.tracker.announce(info_hash, port=7002, left=0, event="completed") + self.assertEqual(self.tracker.scrape(info_hash)[b"files"][info_hash][b"downloaded"], 2) + + def test_a_peer_is_given_the_whole_swarm_including_itself(self): + """opentracker does not filter the announcing peer out of its own + answer. Clients are expected to ignore themselves.""" + info_hash = self.hash_for("sees") + self.tracker.announce(info_hash, port=1111, left=100) + answer = self.tracker.announce(info_hash, port=2222, left=100) + + ports = sorted(port for _, port in peers_of(answer)) + self.assertEqual(ports, [1111, 2222]) + + def test_an_unknown_torrent_scrapes_as_empty(self): + answer = self.tracker.scrape(self.hash_for("never-seen")) + self.assertEqual(answer[b"files"], {}, "a torrent nobody announced has no entry") + + +class AnnounceRejects(TrackerTestCase): + def assert_rejected(self, query): + import urllib.error + + with self.assertRaises(urllib.error.HTTPError) as caught: + self.tracker.get("/announce?" + query) + self.assertEqual(caught.exception.code, 400) + + def test_a_short_info_hash_is_rejected(self): + self.assert_rejected("info_hash=tooshort&peer_id=" + "b" * 20 + "&port=6881&left=0") + + def test_a_missing_info_hash_answers_a_failure_reason(self): + """Not a 400: the tracker answers 200 with a bencoded complaint, so + that a client shows the message to its user.""" + from harness import bdecode + + answer = bdecode(self.tracker.get("/announce?peer_id=" + "b" * 20 + "&port=6881&left=0")) + self.assertIn(b"failure reason", answer) + self.assertIn(b"info_hash", answer[b"failure reason"]) + + def test_a_port_out_of_range_is_rejected(self): + self.assert_rejected("info_hash=" + "a" * 20 + "&peer_id=" + "b" * 20 + "&port=65536&left=0") + + def test_a_port_that_is_not_a_number_is_rejected(self): + self.assert_rejected("info_hash=" + "a" * 20 + "&peer_id=" + "b" * 20 + "&port=abc&left=0") + + def test_two_info_hashes_are_rejected(self): + self.assert_rejected("info_hash=" + "a" * 20 + "&info_hash=" + "c" * 20 + "&peer_id=" + "b" * 20 + "&port=6881&left=0") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/integration/test_dualstack.py b/tests/integration/test_dualstack.py new file mode 100644 index 0000000..cd5ba7d --- /dev/null +++ b/tests/integration/test_dualstack.py @@ -0,0 +1,89 @@ +"""A tracker serving both families from one socket. + +These need WANT_IP_FROM_QUERY_STRING so a peer's address can be dictated +rather than being whatever loopback the test happened to come from. +""" + +import os +import socket +import sys +import unittest + +sys.path.insert(0, os.path.dirname(__file__)) +from harness import TrackerTestCase + + +def addresses(answer, key): + blob, out = answer.get(key, b""), [] + size = 6 if key == b"peers" else 18 + for offset in range(0, len(blob), size): + entry = blob[offset : offset + size] + out.append(socket.inet_ntop(socket.AF_INET if size == 6 else socket.AF_INET6, entry[:-2])) + return out + + +class DualStack(TrackerTestCase): + def test_an_ipv4_peer_lands_in_the_v4_list(self): + info_hash = self.hash_for("v4only") + self.tracker.announce(info_hash, ip="10.1.2.3", port=5001, left=100) + answer = self.tracker.announce(info_hash, ip="10.1.2.4", port=5002, left=100) + + self.assertEqual(sorted(addresses(answer, b"peers")), ["10.1.2.3", "10.1.2.4"]) + self.assertEqual(addresses(answer, b"peers6"), [], "no v6 peer was announced") + + def test_an_ipv6_peer_lands_in_the_v6_list(self): + info_hash = self.hash_for("v6only") + self.tracker.announce(info_hash, ip="2001:db8::1", port=5001, left=100) + answer = self.tracker.announce(info_hash, ip="2001:db8::2", port=5002, left=100) + + self.assertEqual(sorted(addresses(answer, b"peers6")), ["2001:db8::1", "2001:db8::2"]) + self.assertEqual(addresses(answer, b"peers"), [], "no v4 peer was announced") + + def test_the_two_families_are_kept_apart_but_both_returned(self): + info_hash = self.hash_for("both") + self.tracker.announce(info_hash, ip="10.9.9.9", port=6001, left=100) + answer = self.tracker.announce(info_hash, ip="2001:db8::9", port=6002, left=100) + + self.assertEqual(addresses(answer, b"peers"), ["10.9.9.9"]) + self.assertEqual(addresses(answer, b"peers6"), ["2001:db8::9"]) + + def test_a_v4_mapped_address_is_the_same_peer_as_the_plain_one(self): + info_hash = self.hash_for("v4mapped") + self.tracker.announce(info_hash, ip="10.5.5.5", port=7001, left=100) + answer = self.tracker.announce(info_hash, ip="::ffff:10.5.5.5", port=7001, left=100) + + self.assertEqual(answer[b"incomplete"], 1, "the same address written two ways is one peer") + self.assertEqual(addresses(answer, b"peers"), ["10.5.5.5"], "and it is held as v4") + + def test_the_two_families_count_into_one_swarm(self): + info_hash = self.hash_for("counts") + self.tracker.announce(info_hash, ip="10.4.4.4", port=8001, left=0) + answer = self.tracker.announce(info_hash, ip="2001:db8::4", port=8002, left=100) + + self.assertEqual(answer[b"complete"], 1) + self.assertEqual(answer[b"incomplete"], 1) + self.assertEqual(self.tracker.scrape(info_hash)[b"files"][info_hash][b"complete"], 1) + + def test_stopped_removes_only_the_family_it_came_from(self): + info_hash = self.hash_for("stopfamily") + self.tracker.announce(info_hash, ip="10.3.3.3", port=9001, left=100) + self.tracker.announce(info_hash, ip="2001:db8::3", port=9002, left=100) + self.tracker.announce(info_hash, ip="10.3.3.3", port=9001, left=100, event="stopped") + + answer = self.tracker.announce(info_hash, ip="2001:db8::33", port=9003, left=100) + self.assertEqual(addresses(answer, b"peers"), [], "the v4 peer left") + self.assertIn("2001:db8::3", addresses(answer, b"peers6"), "the v6 one stayed") + + def test_a_malformed_address_is_rejected(self): + import urllib.error + + # The whole field has to be the address: a prefix that happens to + # parse would quietly record the peer somewhere else + for bad in ("999.1.2.3", "nonsense", "2001:db8:::1", "10.1.2.3.4", "10.1.2.3junk", "2001:db8::1%eth0"): + with self.assertRaises(urllib.error.HTTPError, msg="ip=%s should be refused" % bad) as caught: + self.tracker.announce(self.hash_for("badip"), ip=bad, port=6881, left=0) + self.assertEqual(caught.exception.code, 400) + + +if __name__ == "__main__": + unittest.main() -- cgit v1.3