1 files changed, 105 insertions, 0 deletions
diff --git a/packages/meshbay-hub/tests/test_download_lanes.py b/packages/meshbay-hub/tests/test_download_lanes.py
new file mode 100644
index 0000000..f133f5c
--- /dev/null
+++ b/packages/meshbay-hub/tests/test_download_lanes.py
@@ -0,0 +1,105 @@
+"""
+A large download is spread over several connections to its node.
+
+One WebRTC connection carries 5 to 6 MB/s at 22 ms on a line that carries 47:
+the browser's UDP socket has a 64 KB kernel buffer, a burst overflows it, and
+the node's SCTP sender halves its window at every loss. The ceiling is per
+connection, so `MeshBayTransport.lanes` opens more of them and
+`pipelinedDownload` sends chunk i over lane i modulo their number.
+
+These run the shipped lanes (transport.js, executed whole) under the shipped
+pipeline (lifted out of file-utils.js as text) against a fake clock and a fake
+node. See harness/download_lanes_harness.mjs.
+"""
+
+import json
+import os
+import shutil
+import subprocess
+from pathlib import Path
+
+import pytest
+from spa_source import classic_scripts
+
+STATIC = Path(__file__).resolve().parents[1] / "src" / "meshbay_hub" / "static"
+FILE_UTILS = STATIC / "file-utils.js"
+HARNESS = Path(__file__).parent / "harness" / "download_lanes_harness.mjs"
+
+pytestmark = pytest.mark.skipif(
+    shutil.which("node") is None or not FILE_UTILS.exists(),
+    reason="node or the SPA sources are not available")
+
+
+def _run(**cfg) -> dict:
+    transport = os.pathsep.join(
+        str(STATIC / name) for name in classic_scripts()
+        if name.startswith("transport"))
+    proc = subprocess.run(
+        ["node", str(HARNESS), str(FILE_UTILS), transport, json.dumps(cfg)],
+        capture_output=True, text=True, encoding="utf-8")
+    assert proc.returncode == 0, proc.stderr
+    return json.loads(proc.stdout)
+
+
+def test_a_large_download_runs_over_eight_connections_in_order():
+    out = _run(chunks=200)
+    assert out["outcome"] == "done"
+    assert out["written"] == 200 and out["inOrder"]
+    assert out["offers"] == 7, "seven lanes beside the connection already open"
+    lanes = {k: v for k, v in out["carried"].items() if k != "main"}
+    assert len(lanes) == 7 and min(lanes.values()) >= 15, out["carried"]
+    # It starts on the main connection without waiting for the lanes.
+    assert out["carried"]["main"] > max(lanes.values())
+    assert out["peakInFlight"] == 16, "two chunks in flight per connection"
+
+
+def test_a_lane_is_the_main_connection_again_and_never_a_join():
+    args = _run(chunks=200)["laneArgs"]
+    assert args == {"groupId": "group", "sameGek": True, "sameIdentity": True,
+                    "joinCode": None, "nodePk": "pk"}
+
+
+def test_a_small_download_opens_nothing():
+    out = _run(chunks=10)
+    assert out["outcome"] == "done" and out["offers"] == 0
+    assert out["peakInFlight"] == 8
+
+
+def test_a_read_without_a_lease_opens_nothing():
+    """A preview or a track is not a transfer, and the node would count each
+    lane's read against that lane's own leaseless ceiling."""
+    out = _run(chunks=200, tr="")
+    assert out["outcome"] == "done" and out["offers"] == 0
+    assert out["carried"] == {"main": 200}
+
+
+def test_a_lane_that_fails_is_dropped_and_its_chunks_asked_again():
+    out = _run(chunks=200, failLaneAfter=3)
+    assert out["outcome"] == "done"
+    assert out["written"] == 200 and out["inOrder"]
+    assert out["retried"] >= 1
+    assert out["carried"]["lane1"] <= 3
+    assert out["offers"] == 8, "the dropped lane is replaced, once"
+
+
+def test_a_node_that_refuses_lanes_costs_one_round_of_offers():
+    """`lanes` is asked at every chunk: without the retry delay a refusing node
+    would be sent an offer per chunk."""
+    out = _run(chunks=200, refuseLanes=True)
+    assert out["outcome"] == "done"
+    assert out["carried"] == {"main": 200}
+    assert out["offers"] == 7
+
+
+def test_idle_lanes_are_closed():
+    out = _run(chunks=200)
+    assert out["lanesAfterIdle"] == 0 and out["lanesLeftOpen"] == 0
+    assert out["timersLeft"] == 0, "the idle timer outlived the lanes"
+
+
+@pytest.mark.parametrize("close_at", [400, 1500])
+def test_closing_the_connection_closes_its_lanes(close_at):
+    """At 400 ms the lanes are still negotiating, at 1500 they are carrying."""
+    out = _run(chunks=2000, closeAt=close_at)
+    assert out["lanesLeftOpenByClose"] == 0
+    assert out["offers"] == 7, "a closed connection opened more lanes"