diff options
Diffstat (limited to 'packages/meshbay-node/tests/test_task_lifetime.py')
| -rw-r--r-- | packages/meshbay-node/tests/test_task_lifetime.py | 51 |
1 files changed, 16 insertions, 35 deletions
diff --git a/packages/meshbay-node/tests/test_task_lifetime.py b/packages/meshbay-node/tests/test_task_lifetime.py index 133c030..a8445fa 100644 --- a/packages/meshbay-node/tests/test_task_lifetime.py +++ b/packages/meshbay-node/tests/test_task_lifetime.py @@ -25,33 +25,28 @@ file. """ import re -from pathlib import Path import pytest - -SERVER = (Path(__file__).resolve().parents[1] / "src" / "meshbay_node" - / "transport" / "webrtc_server.py") +from node_source import TRANSPORT, session_method, session_source, webrtc_source pytestmark = pytest.mark.skipif( - not SERVER.exists(), reason="the node sources are not available") + not TRANSPORT.exists(), reason="the node sources are not available") @pytest.fixture(scope="module") def source(): - return SERVER.read_text(encoding="utf-8") + return webrtc_source() @pytest.fixture(scope="module") -def session(source): - """The body of WebRTCPeerSession.""" - i = source.index("class WebRTCPeerSession") - return source[i:source.index("\nclass WebRTCTransport")] +def session(): + """The body of WebRTCPeerSession, and of every class it is built from.""" + return session_source() def test_the_session_keeps_a_reference_to_what_it_starts(session): assert "self._tasks: set[asyncio.Task] = set()" in session - spawn = session[session.index("def _spawn("):] - spawn = spawn[:spawn.index("\n def ", 1)] + spawn = session_method("_spawn") assert "self._tasks.add(task)" in spawn, "the reference is what keeps it alive" assert "self._tasks.discard(" in spawn and "add_done_callback(" in spawn, ( "without this the set grows for the life of the session") @@ -72,8 +67,7 @@ def test_nothing_in_the_session_is_fired_and_forgotten(session): def test_the_stream_still_takes_a_slot_for_its_whole_life(session): """The leak is only interesting because the slot is held this way.""" - body = session[session.index("async def _stream_video(self"):] - body = body[:body.index("\n async def ", 1)] + body = session_method("_stream_video") assert "async with sem:" in body @@ -83,13 +77,11 @@ def test_closing_a_session_releases_its_tasks(session): The cancelling itself lives in shutdown_tasks, which the state handler also uses; close() is the variant that additionally shuts the peer connection. """ - close = session[session.index("async def close(self)"):] - close = close[:close.index("\n\n")] if "\n\n" in close else close + close = session_method("close") assert "shutdown_tasks()" in close assert "self._pc.close()" in close - fn = session[session.index("async def shutdown_tasks(self)"):] - fn = fn[:fn.index("\n async def ", 1)] + fn = session_method("shutdown_tasks") assert "_stop_stream()" in fn assert "task.cancel()" in fn assert "gather" in fn, "cancelling without awaiting does not run the exits" @@ -129,8 +121,7 @@ def test_a_new_request_retires_the_previous_stream(session): assert "self._spawn(self._replace_stream(msg))" in session, ( "the stream request must go through the path that retires the old one") - fn = session[session.index("async def _replace_stream(self"):] - fn = fn[:fn.index("\n async def ", 1)] + fn = session_method("_replace_stream") assert "self._stop_stream()" in fn assert "await asyncio.wait_for(asyncio.shield(prev)" in fn, ( "the slot comes back when the old task exits its `async with sem` — " @@ -173,8 +164,7 @@ def test_losing_the_peer_stops_its_work(source): def test_shutdown_is_separate_from_closing_the_connection(session): """The state handler runs while aiortc is already tearing pc down.""" - fn = session[session.index("async def shutdown_tasks(self)"):] - fn = fn[:fn.index("\n async def ", 1)] + fn = session_method("shutdown_tasks") assert "self._pc.close()" not in fn, ( "calling pc.close() from the state handler re-enters the teardown") assert "task.cancel()" in fn and "gather" in fn @@ -186,8 +176,7 @@ def test_a_dead_channel_is_noticed_while_waiting_not_after(source): Waiting the whole budget on a channel that is already shut is the slot being held for nothing, which is what the log above shows. """ - fn = source[source.index("async def _await_stream_credit"):] - fn = fn[:fn.index("\n async def ", 1)] + fn = session_method("_await_stream_credit") before = fn[:fn.index("wait_for")] assert 'readyState != "open"' in before, ( "the channel must be checked before the wait, not only after it") @@ -221,8 +210,7 @@ def test_the_pipes_are_drained_before_waiting_on_ffmpeg(source): Which is the reported failure exactly: one video fine, the next hanging, the one after refused. """ - fn = source[source.index("async def _stream_video_inner"):] - fn = fn[:fn.index("\n def _send(")] + fn = session_method("_stream_video_inner") finally_block = fn[fn.index("finally:"):] assert "proc.kill()" in finally_block @@ -234,8 +222,7 @@ def test_the_pipes_are_drained_before_waiting_on_ffmpeg(source): def test_releasing_the_slot_does_not_depend_on_ffmpeg_behaving(source): """SIGKILL has already been sent; the OS will reap it either way.""" - fn = source[source.index("async def _stream_video_inner"):] - fn = fn[:fn.index("\n def _send(")] + fn = session_method("_stream_video_inner") tail = fn[fn.index("wait_for(proc.wait()"):] assert "except Exception" in tail, ( "a timeout waiting for ffmpeg must not stop the slot coming back") @@ -253,13 +240,7 @@ def test_chunks_wait_for_room_on_the_channel(session): work through the rest — which is what "stuck at 1 MB" looks like, one chunk being exactly one megabyte. """ - start = session.index("async def _do_file_request") - # Up to whatever the next member is. This used to end at - # "\n def _do_stream_segment" — a neighbour removed in MNP 2.0 — and an - # `index()` on a name that no longer exists fails the test for a reason - # that has nothing to do with what it is about. - nxt = re.search(r"\n (?:@|(?:async )?def )", session[start:]) - fn = session[start:start + nxt.start()] if nxt else session[start:] + fn = session_method("_do_file_request") assert "DOWNLOAD_BUFFER_HIGH" in fn, "the send buffer has to be watched" assert "await asyncio.sleep" in fn, "waiting for room is the point" assert 'readyState != "open"' in fn, ( |