1 files changed, 285 insertions, 0 deletions
diff --git a/packages/meshbay-cli/tests/test_stream.py b/packages/meshbay-cli/tests/test_stream.py
new file mode 100644
index 0000000..46dc630
--- /dev/null
+++ b/packages/meshbay-cli/tests/test_stream.py
@@ -0,0 +1,285 @@
+"""
+`stream`: a file served on this machine to a media player — byte ranges from
+the node's chunks, read without a lease, only to a request that names this
+address and carries the token; the connection opened again when it dies, and
+the stream stopped by the node's refusal, by Ctrl+C or by closing the player.
+"""
+
+import asyncio
+import json
+import os
+import secrets
+import stat
+import sys
+from urllib.parse import urlsplit
+
+import pytest
+from fake_hub import PASSPHRASE, USERNAME, FakeHub
+from fake_node import FakeNode
+from meshbay_cli import __main__ as cli
+from meshbay_cli.client import relay
+from meshbay_cli.client.relay import byte_range, content_type
+from meshbay_cli.commands import context, stream
+from meshbay_cli.errors import NODE, NOT_FOUND, USAGE, CliError
+
+MiB = 1024 * 1024
+FILM = secrets.token_bytes(3 * MiB + 12345)
+
+
+@pytest.fixture
+def world(monkeypatch, capsys):
+    hub, node = FakeHub(), FakeNode()
+    monkeypatch.setattr(context, "Hub", lambda url: hub)
+    sessions = []
+
+    async def open_group(account, group, **_):
+        sessions.append(node.session())
+        return sessions[-1]
+    monkeypatch.setattr(stream, "open_group", open_group)
+    monkeypatch.setattr(relay, "PING_EVERY", 0.05)
+    monkeypatch.setattr(relay, "RECONNECT_WAITS", (0.01,))
+    node.add("share/films", "Holiday.mkv", FILM)["type"] = "video"
+    node.add("share/docs", "notes.txt", b"hello")
+    assert cli.main(["--hub", hub.url, "-u", USERNAME, "--password", PASSPHRASE,
+                     "login"]) == 0
+    capsys.readouterr()
+
+    def start(*argv):
+        args = cli.build_parser().parse_args(list(argv))
+        return asyncio.ensure_future(stream.run(context.Context(args)))
+    return node, sessions, start
+
+
+def node_film(node) -> dict:
+    return next(e for e in node.files.values() if e["name"] == "Holiday.mkv")
+
+
+async def _until(cond, timeout=5.0):
+    deadline = asyncio.get_running_loop().time() + timeout
+    while not cond():
+        assert asyncio.get_running_loop().time() < deadline, "timed out"
+        await asyncio.sleep(0.02)
+
+
+async def _url(task, capsys, said: list | None = None) -> str:
+    """The address printed on stdout; what was said on stderr meanwhile into `said`."""
+    out = []
+
+    def printed():
+        assert not task.done(), task.exception()
+        got = capsys.readouterr()
+        out.extend(got.out.split())
+        if said is not None:
+            said.append(got.err)
+        return out
+    await _until(printed)
+    return out[0]
+
+
+async def http(url: str, *, method="GET", headers=None, host=None, path=None,
+               keep=False, reader_writer=None):
+    """One request; (status, headers, body). With `keep`, the connection is returned
+    for another request."""
+    u = urlsplit(url)
+    reader, writer = reader_writer or await asyncio.open_connection(u.hostname, u.port)
+    lines = [f"{method} {path or u.path} HTTP/1.1", f"Host: {host or u.netloc}",
+             *(f"{k}: {v}" for k, v in (headers or {}).items())]
+    if not keep:
+        lines.append("Connection: close")
+    writer.write(("\r\n".join(lines) + "\r\n\r\n").encode())
+    await writer.drain()
+    head = (await reader.readuntil(b"\r\n\r\n")).decode("latin-1").split("\r\n")
+    status = int(head[0].split(" ")[1])
+    fields = {k.strip().lower(): v.strip() for k, _, v in
+              (line.partition(":") for line in head[1:] if line)}
+    body = b""
+    if method != "HEAD":
+        body = await reader.readexactly(int(fields.get("content-length", 0)))
+    if keep:
+        return status, fields, body, (reader, writer)
+    writer.close()
+    return status, fields, body
+
+
+# ── Ranges and types ─────────────────────────────────────────────────────────
+
+def test_a_range_is_read_as_players_send_it():
+    assert byte_range(None, 1000) is None
+    assert byte_range("bytes=0-", 1000) == (0, 999)
+    assert byte_range("bytes=10-19", 1000) == (10, 19)
+    assert byte_range("bytes=990-5000", 1000) == (990, 999)
+    assert byte_range("bytes=-100", 1000) == (900, 999)
+    assert byte_range("bytes=-5000", 1000) == (0, 999)
+    assert byte_range("bytes=0-1,5-9", 1000) is None        # several: the whole file
+    assert byte_range("items=0-9", 1000) is None
+    for bad in ("bytes=1000-", "bytes=20-10", "bytes=x-", "bytes=-0"):
+        with pytest.raises(ValueError):
+            byte_range(bad, 1000)
+
+
+def test_a_file_is_served_as_audio_or_video_and_never_as_a_page():
+    assert content_type("Film.MKV", "video") == "video/x-matroska"
+    assert content_type("a.flac", "audio") == "audio/flac"
+    assert content_type("trap.html", "video") == "video/octet-stream"
+    assert content_type("trap.svg", "audio") == "audio/octet-stream"
+
+
+# ── Serving ──────────────────────────────────────────────────────────────────
+
+async def test_the_player_gets_the_file_and_any_range_of_it(world, capsys):
+    node, sessions, start = world
+    task = start("stream", "Family/share/films/Holiday.mkv")
+    try:
+        said = []
+        url = await _url(task, capsys, said)
+        assert url.startswith("http://127.0.0.1:") and url.endswith("/Holiday.mkv")
+        assert "Ctrl+C to stop" in "".join(said)
+
+        status, fields, body = await http(url)
+        assert status == 200 and body == FILM
+        assert fields["content-type"] == "video/x-matroska"
+        assert fields["accept-ranges"] == "bytes"
+        assert fields["x-content-type-options"] == "nosniff"
+
+        # Across a chunk boundary, as a seek lands.
+        a, b = MiB - 100, 2 * MiB + 50
+        status, fields, body = await http(url, headers={"Range": f"bytes={a}-{b}"})
+        assert status == 206 and body == FILM[a:b + 1]
+        assert fields["content-range"] == f"bytes {a}-{b}/{len(FILM)}"
+
+        # The end of the file, as a player probes for an index there; then the
+        # same connection again.
+        status, _, body, conn = await http(url, headers={"Range": "bytes=-64"}, keep=True)
+        assert status == 206 and body == FILM[-64:]
+        status, _, body, conn = await http(url, headers={"Range": "bytes=5-9"}, keep=True,
+                                           reader_writer=conn)
+        assert status == 206 and body == FILM[5:10]
+        conn[1].close()
+
+        status, fields, body = await http(url, method="HEAD")
+        assert status == 200 and fields["content-length"] == str(len(FILM)) and body == b""
+        status, fields, _ = await http(url, headers={"Range": f"bytes={len(FILM)}-"})
+        assert status == 416 and fields["content-range"] == f"bytes */{len(FILM)}"
+
+        # Read as a player reads, without a lease; chunks kept are not asked twice.
+        assert node.leases == {} and node.leaseless_reads
+        assert sorted(node.leaseless_reads) == [(node_film(node)["id"], i) for i in range(4)]
+        assert len(sessions) == 1
+    finally:
+        task.cancel()
+        await asyncio.gather(task, return_exceptions=True)
+
+
+async def test_only_this_address_with_its_token_is_answered(world, capsys):
+    node, _, start = world
+    task = start("stream", "Family/share/films/Holiday.mkv")
+    try:
+        url = await _url(task, capsys)
+        port = urlsplit(url).port
+        # A page that rebinds its own name to 127.0.0.1 sends its own Host.
+        status, _, body = await http(url, host=f"evil.example:{port}")
+        assert status == 403 and body == b""
+        status, _, body = await http(url, path="/not-the-token/Holiday.mkv")
+        assert status == 404 and body == b""
+        status, _, _ = await http(url, method="POST")
+        assert status == 405
+        assert node.leaseless_reads == [], "nothing refused reached the node"
+        status, _, body = await http(url, host=f"localhost:{port}",
+                                     headers={"Range": "bytes=0-3"})
+        assert status == 206 and body == FILM[:4]
+    finally:
+        task.cancel()
+        await asyncio.gather(task, return_exceptions=True)
+
+
+async def test_a_dead_connection_is_opened_again(world, capsys):
+    node, sessions, start = world
+    task = start("stream", "Family/share/films/Holiday.mkv")
+    try:
+        url = await _url(task, capsys)
+        assert (await http(url, headers={"Range": "bytes=0-9"}))[2] == FILM[:10]
+
+        sessions[-1].ch.closed.set()                    # the link goes down
+        await _until(lambda: len(sessions) == 2)
+        a = 3 * MiB
+        assert (await http(url, headers={"Range": f"bytes={a}-"}))[2] == FILM[a:]
+
+        node.unresponsive = True                        # it stops answering pings
+        await _until(lambda: len(sessions) == 3)
+        node.unresponsive = False
+        assert "reconnecting" in capsys.readouterr().err
+    finally:
+        task.cancel()
+        await asyncio.gather(task, return_exceptions=True)
+
+
+async def test_a_refusal_by_the_node_stops_the_stream(world, capsys):
+    node, _, start = world
+    node.fail_chunk = (node_film(node)["id"], 2)
+    task = start("stream", "Family/share/films/Holiday.mkv")
+    url = await _url(task, capsys)
+    status, _, _ = await http(url, headers={"Range": f"bytes={2 * MiB}-"})
+    assert status == 502
+    with pytest.raises(CliError) as e:
+        await asyncio.wait_for(task, 5)
+    assert e.value.code == NODE and "drive stopped answering" in e.value.message
+
+
+async def test_what_cannot_be_streamed_is_said(world, capsys):
+    _, _, start = world
+    for argv, code, says in [
+        (["Family/share/docs/notes.txt"], USAGE, "not a video or a sound file"),
+        (["Family/share/films"], USAGE, "is a folder"),
+        (["Family/share/films/Nope.mkv"], NOT_FOUND, "Nothing"),
+        (["Family"], USAGE, "Name the file"),
+        (["--player", "no-such-player-here", "Family/share/films/Holiday.mkv"], USAGE,
+         "was not found"),
+    ]:
+        with pytest.raises(CliError) as e:
+            await start("stream", *argv)
+        assert e.value.code == code and says in e.value.message, argv
+
+
+# ── --player ─────────────────────────────────────────────────────────────────
+
+def _player(tmp_path, body: str) -> str:
+    script = tmp_path / "player"
+    script.write_text(f"#!{sys.executable}\nimport os, sys, time, urllib.request\n{body}\n")
+    script.chmod(script.stat().st_mode | stat.S_IXUSR)
+    return str(script)
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="a script as a program")
+async def test_closing_the_player_stops_the_stream(world, capsys, tmp_path):
+    out = tmp_path / "played"
+    player = _player(tmp_path, (
+        "opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))\n"
+        f"open({str(out)!r}, 'wb').write(opener.open(sys.argv[1]).read())"))
+    _, _, start = world
+    code = await asyncio.wait_for(
+        start("stream", "--player", player, "Family/share/films/Holiday.mkv"), 10)
+    assert code == 0 and out.read_bytes() == FILM
+    assert "player was closed" in capsys.readouterr().err
+
+
+@pytest.mark.skipif(sys.platform == "win32", reason="a script as a program")
+async def test_stopping_the_stream_closes_the_player_it_started(world, capsys, tmp_path):
+    pid = tmp_path / "pid"
+    player = _player(tmp_path, f"open({str(pid)!r}, 'w').write(str(os.getpid()))\n"
+                               "time.sleep(60)")
+    _, _, start = world
+    task = start("--json", "stream", "--player", player, "Family/share/films/Holiday.mkv")
+    await _until(lambda: pid.exists() and pid.read_text())
+    shown = json.loads(capsys.readouterr().out)
+    assert shown["name"] == "Holiday.mkv" and shown["size"] == len(FILM)
+    task.cancel()                                       # Ctrl+C
+    await asyncio.gather(task, return_exceptions=True)
+
+    def gone():
+        try:
+            os.kill(int(pid.read_text()), 0)
+        except ProcessLookupError:
+            return True
+        with open(f"/proc/{pid.read_text()}/stat") as f:
+            return f.read().split()[2] == "Z"           # ended, not yet reaped
+    await _until(gone)