""" Transfer leases over the session, rather than over `TransferSlots` alone. test_transfer_slots.py proves the decisions; this proves the seam. Both exist because the seam is where this repo's defects have actually lived — a reply routed by arrival order, a session popped from a dict without its work being stopped, a slot released by a `finally` nobody reached. Three things can only be checked here: - the handlers answer under the right shape, and refuse another connection's transfer id; - **losing the connection gives everything back.** That is the primary reclaim, and it is a hook (`shutdown_tasks`) rather than a timer, so a test of the pool alone would never touch it; - a slot freed by one peer is *announced* to the peer waiting on it. A grant nobody hears about is precisely the "stuck at waiting" report the design exists to prevent, and it would look correct in the pool. """ import pytest # Every test drives a message handler, and in the node a message handler always # runs inside the event loop: `_do_transfer_open` starts the sweeper task there. # Calling these synchronously tested a situation that cannot happen and failed # on "no current event loop" the moment the sweeper stopped being faked. pytestmark = pytest.mark.asyncio from meshbay_common.protocol import MNP from meshbay_node.transfers import DOWNLOAD, UPLOAD from meshbay_node.transport.webrtc_server import WebRTCPeerSession class _Session(WebRTCPeerSession): """A session with the DataChannel replaced by a list, and nothing else.""" def __init__(self, ctx, *, key, user, group="g1"): self._ctx = ctx self._registry_key = key self._user_id = user self._group_id = group self.sent: list[dict] = [] def _send(self, msg): self.sent.append(msg) def _spawn(self, coro): # pragma: no cover - not used by these tests coro.close() return None def last(self, mtype=MNP.TRANSFER_STATE): return next(m for m in reversed(self.sent) if m.get("type") == mtype) @pytest.fixture def ctx(): """A transport context, with the sweeper stopped on the way out. A task left running past the end of its test is a warning in the next one and a hang in the worst case; the sweeper is started on demand by design, so tearing it down is the test's job. """ c: dict = {"_peers": {}} yield c task = c.get("_transfer_sweeper") if task is not None: task.cancel() def _join(ctx, key, user, group="g1") -> _Session: s = _Session(ctx, key=key, user=user, group=group) ctx["_peers"][key] = s return s async def test_a_granted_transfer_is_answered_as_granted(ctx): peer = _join(ctx, "s1", "alice") peer._do_transfer_open({"tr": "t1", "kind": DOWNLOAD, "bytes": 10}) reply = peer.last() assert reply["state"] == "granted" assert reply["tr"] == "t1" assert reply["kind"] == DOWNLOAD assert reply["used"] == 1 and reply["cap"] >= 1 async def test_a_queued_transfer_is_told_how_many_are_ahead(ctx): peer = _join(ctx, "s1", "alice") peer._slots().per_member[DOWNLOAD] = 1 peer._do_transfer_open({"tr": "t1"}) peer._do_transfer_open({"tr": "t2"}) peer._do_transfer_open({"tr": "t3"}) assert [m["state"] for m in peer.sent] == ["granted", "queued", "queued"] assert peer.sent[-1]["ahead"] == 1 async def test_the_reply_carries_no_name_and_no_path(ctx): """A lease holds neither, and `transfer_state` stays in clear — so this is the message where a filename would quietly become metadata on the wire.""" peer = _join(ctx, "s1", "alice") peer._do_transfer_open({"tr": "t1", "name": "Some Saga.mkv", "path": "/srv/films"}) assert set(peer.last()) <= { "type", "v", "tr", "state", "kind", "used", "cap", "node_used", "node_cap", "ahead", "reason"} async def test_closing_frees_the_slot(ctx): peer = _join(ctx, "s1", "alice") peer._do_transfer_open({"tr": "t1"}) peer._do_transfer_close({"tr": "t1", "reason": "done"}) assert peer.last()["state"] == "closed" assert peer._slots().in_use(DOWNLOAD) == 0 async def test_one_peer_cannot_close_anothers_transfer(ctx): """A denial of service one random id away, otherwise.""" alice = _join(ctx, "s1", "alice") bob = _join(ctx, "s2", "bob") alice._do_transfer_open({"tr": "t1"}) bob._do_transfer_close({"tr": "t1"}) assert bob.last("error")["code"] == "not_your_transfer" assert "t1" in alice._slots().leases async def test_one_peer_cannot_open_on_anothers_id(ctx): alice = _join(ctx, "s1", "alice") bob = _join(ctx, "s2", "bob") alice._do_transfer_open({"tr": "t1"}) bob._do_transfer_open({"tr": "t1"}) assert bob.last("error")["code"] == "not_your_transfer" async def test_a_transfer_with_no_id_is_refused(ctx): peer = _join(ctx, "s1", "alice") peer._do_transfer_open({"kind": DOWNLOAD}) assert peer.last("error")["code"] == "bad_transfer_id" # ── the reclaim that matters ──────────────────────────────────────────────── async def test_losing_the_connection_gives_everything_back(ctx): peer = _join(ctx, "s1", "alice") peer._do_transfer_open({"tr": "t1"}) peer._do_transfer_open({"tr": "t2", "kind": UPLOAD}) peer._release_transfers() slots = peer._slots() assert slots.leases == {} assert slots.in_use(DOWNLOAD) == 0 and slots.in_use(UPLOAD) == 0 async def test_the_freed_slot_reaches_the_peer_that_was_waiting(ctx): """ The seam this file exists for. In the pool, granting is correct; if the grant is not pushed, the waiting client sits on "waiting" for ever with a node that believes it is streaming — and every unit test still passes. """ alice = _join(ctx, "s1", "alice") bob = _join(ctx, "s2", "bob") alice._slots().caps[DOWNLOAD] = 1 alice._do_transfer_open({"tr": "a1"}) bob._do_transfer_open({"tr": "b1"}) assert bob.last()["state"] == "queued" alice._release_transfers() assert bob.last()["state"] == "granted", ( "bob was granted the slot and never told") assert bob.last()["tr"] == "b1" async def test_a_grant_crossing_groups_still_reaches_its_peer(ctx): """The pools are node-wide and `_peer_registry` is per group (finding H1), so the peer to notify is not necessarily in the notifier's own registry.""" groups = {"g1": {"_peers": {}}, "g2": {"_peers": {}}} ctx = {"groups": groups} alice = _Session(ctx, key="s1", user="alice", group="g1") groups["g1"]["_peers"]["s1"] = alice bob = _Session(ctx, key="s2", user="bob", group="g2") groups["g2"]["_peers"]["s2"] = bob alice._slots().caps[DOWNLOAD] = 1 alice._do_transfer_open({"tr": "a1"}) bob._do_transfer_open({"tr": "b1"}) assert bob.last()["state"] == "queued" alice._release_transfers() assert bob.last()["state"] == "granted", ( "a slot freed in one group never reached the peer waiting in another") async def test_raising_the_cap_notifies_who_it_starts(ctx): """`set_capacity` arrives from the loopback API, with no session behind it — the grants it produces still have to be pushed.""" from meshbay_node.transport.webrtc_server import WebRTCTransport transport = WebRTCTransport.__new__(WebRTCTransport) transport._ctx = ctx peer = _join(ctx, "s1", "alice") peer._slots().caps[DOWNLOAD] = 1 peer._do_transfer_open({"tr": "t1"}) peer._do_transfer_open({"tr": "t2"}) assert peer.last()["state"] == "queued" transport.set_capacity(max_concurrent_downloads=4) assert peer.last()["state"] == "granted" and peer.last()["tr"] == "t2" async def test_the_operator_can_see_the_queue(ctx): """`GET /api/transfers` is the answer to "was this peer ever queued", which a log line cannot give when the symptom is that nothing is happening.""" from meshbay_node import ops peer = _join(ctx, "s1", "alice") peer._slots().caps[DOWNLOAD] = 1 peer._do_transfer_open({"tr": "t1", "bytes": 5}) peer._do_transfer_open({"tr": "t2", "bytes": 7}) class _T: _ctx = ctx snapshot = await ops.list_transfers({"webrtc": _T()}) assert snapshot["pools"][DOWNLOAD]["in_use"] == 1 assert snapshot["pools"][DOWNLOAD]["queued"] == 1 assert {x["state"] for x in snapshot["leases"]} == {"granted", "queued"} assert all("name" not in x and "path" not in x for x in snapshot["leases"]) async def test_asking_before_anything_has_transferred_is_not_an_error(): from meshbay_node import ops class _T: _ctx: dict = {} snapshot = await ops.list_transfers({"webrtc": _T()}) assert snapshot["leases"] == [] assert snapshot["pools"][DOWNLOAD]["in_use"] == 0 @pytest.mark.asyncio async def test_the_caps_shown_are_the_operators_before_anything_transfers(): """ `transfers set 2 2` answers "applied now"; `transfers show` said 0/8 — because the no-pool branch reported the module defaults rather than what the operator had just set. Found by running it against a real node. The previous test asserted the defaults, so it agreed with the bug: an operator would have read that as the hot-swap doing nothing all over again. """ from meshbay_node import ops class _T: _ctx = {"max_concurrent_downloads": 2, "max_concurrent_uploads": 3} snapshot = await ops.list_transfers({"webrtc": _T()}) assert snapshot["pools"][DOWNLOAD]["cap"] == 2 assert snapshot["pools"][UPLOAD]["cap"] == 3 # ── the sweeper's lifetime ────────────────────────────────────────────────── async def test_the_sweeper_outlives_the_session_that_started_it(ctx): """ It was started with `self._spawn`, which ties a task to one session's set — so it was cancelled the moment that peer left, and every other peer's abandoned lease stopped being reclaimed. Nothing else would have noticed: the node simply fills up over days. """ import asyncio from meshbay_node.transport import webrtc_server as ws alice = _join(ctx, "s1", "alice") bob = _join(ctx, "s2", "bob") # The real _spawn, so the session genuinely owns what it starts. alice._tasks = set() alice._spawn = ws.WebRTCPeerSession._spawn.__get__(alice) bob._tasks = set() bob._spawn = ws.WebRTCPeerSession._spawn.__get__(bob) alice._do_transfer_open({"tr": "a1"}) bob._do_transfer_open({"tr": "b1"}) sweeper = ctx["_transfer_sweeper"] assert sweeper is not None and not sweeper.done() # Alice leaves, exactly as shutdown_tasks does it. alice._release_transfers() for task in list(alice._tasks): task.cancel() await asyncio.gather(*alice._tasks, return_exceptions=True) await asyncio.sleep(0) assert not sweeper.done(), ( "the sweeper died with the session that happened to start it; bob's " "lease would never be reclaimed") sweeper.cancel() async def test_the_sweeper_stops_when_the_last_lease_goes(ctx): """An idle node must run no timer — the reason this is started on demand rather than at boot.""" import asyncio from meshbay_node.transport import webrtc_server as ws peer = _join(ctx, "s1", "alice") peer._tasks = set() peer._spawn = ws.WebRTCPeerSession._spawn.__get__(peer) original = ws.TRANSFER_SWEEP_SECS ws.TRANSFER_SWEEP_SECS = 0.01 try: peer._do_transfer_open({"tr": "t1"}) peer._do_transfer_close({"tr": "t1"}) sweeper = ctx["_transfer_sweeper"] await asyncio.wait_for(sweeper, timeout=2) assert ctx.get("_transfer_sweeper") is None finally: ws.TRANSFER_SWEEP_SECS = original async def test_a_chunk_request_keeps_its_lease_alive(ctx): """ The seam that cost an afternoon. `TransferSlots.touch` existed, was tested, and **nothing ever called it**: the node ignored `tr` on `file_req` entirely, so `used` stayed False for every download ever made and the sweeper revoked each grant 30 s in, while the file was transferring. Neither side's tests could see it — the pool was correct, the handlers were correct, and the call between them was missing. Only the node's own log showed it, repeating the same reclaim every 30 s. """ peer = _join(ctx, "s1", "alice") peer._do_transfer_open({"tr": "t1", "bytes": 1024}) lease = peer._slots().leases["t1"] assert lease.used is False # A chunk request for a file that does not exist still counts: what marks # the lease is the peer asking, not the node succeeding. peer._group_ctx()["index"] = None try: await peer._do_file_request({"file_id": "nope", "chunk_index": 0, "tr": "t1"}) except Exception: pass assert peer._slots().leases["t1"].used is True, ( "a chunk request under this lease did not mark it alive; the node will " "revoke the grant in 30 seconds")