1 files changed, 148 insertions, 0 deletions
diff --git a/packages/meshbay-node/tests/test_stream_capacity_config.py b/packages/meshbay-node/tests/test_stream_capacity_config.py
new file mode 100644
index 0000000..7c33a2f
--- /dev/null
+++ b/packages/meshbay-node/tests/test_stream_capacity_config.py
@@ -0,0 +1,148 @@
+"""
+How many people may watch at once, and who decides.
+
+Bounding the client's read-ahead to ninety seconds of film changed what a
+transcode slot is. It used to be a burst — the browser took segments as fast as
+it could append them, so a slot came back within the minute whatever the length
+of the film. Now a slot is held for as long as someone is watching, so the cap
+is a cap on simultaneous viewers, and the right number stopped being a property
+of the code: it depends on the machine the node runs on.
+
+So it belongs to the operator. `max_concurrent_streams` under [node] in
+node.toml, `MESHBAY_MAX_CONCURRENT_STREAMS` in the environment, and the
+constant in the source as the default when neither says anything.
+
+The tests below follow the value along that whole path rather than checking
+that the field parses, because every join in it has been wrong at least once:
+the daemon reads `self._config`, not `self.cfg`, and nothing about the
+attribute that does not exist fails until a video is played.
+"""
+
+import asyncio
+import textwrap
+from pathlib import Path
+
+import pytest
+
+from meshbay_node.config import load_config
+from meshbay_node.transport.webrtc_server import (
+    MAX_CONCURRENT_TRANSCODES,
+    WebRTCPeerSession,
+    WebRTCTransport,
+)
+
+
+def _cfg(tmp_path: Path, body: str):
+    p = tmp_path / "node.toml"
+    p.write_text(textwrap.dedent(body))
+    return load_config(p)
+
+
+# ── What the operator writes ──────────────────────────────────────────────────
+
+def test_the_operator_sets_it(tmp_path):
+    cfg = _cfg(tmp_path, """
+        [node]
+        max_concurrent_streams = 3
+    """)
+    assert cfg.node.max_concurrent_streams == 3
+
+
+def test_saying_nothing_gets_the_default(tmp_path):
+    cfg = _cfg(tmp_path, """
+        [node]
+        quic_port = 19010
+    """)
+    assert cfg.node.max_concurrent_streams == MAX_CONCURRENT_TRANSCODES, (
+        "the config default and the source default disagree, so the number "
+        "depends on whether a node.toml happens to mention it")
+
+
+def test_the_environment_wins_over_the_file(tmp_path, monkeypatch):
+    monkeypatch.setenv("MESHBAY_MAX_CONCURRENT_STREAMS", "5")
+    cfg = _cfg(tmp_path, """
+        [node]
+        max_concurrent_streams = 3
+    """)
+    assert cfg.node.max_concurrent_streams == 5
+
+
+@pytest.mark.parametrize("value", ["0", "-4", '"lots"', "true"])
+def test_a_value_that_would_break_streaming_is_refused(tmp_path, value, caplog):
+    """Zero is the dangerous one.
+
+    `asyncio.Semaphore(0)` is not "no limit". It is a node where every video
+    waits forever, with nothing in the log to say why — so the operator gets
+    the default and a warning naming the setting instead.
+    """
+    cfg = _cfg(tmp_path, f"""
+        [node]
+        max_concurrent_streams = {value}
+    """)
+    assert cfg.node.max_concurrent_streams == MAX_CONCURRENT_TRANSCODES
+    assert "max_concurrent_streams" in caplog.text, (
+        "the value was silently discarded — the operator has no way to learn "
+        "their setting is not in effect")
+
+
+# ── That the number reaches the thing it limits ───────────────────────────────
+
+class _FakePC:
+    def on(self, *a, **k):
+        return lambda f: f
+
+
+def _semaphore_size(n):
+    t = WebRTCTransport(
+        sk_node=None, hub_pk_pem=b"", gek=b"\0" * 32,
+        shared_root=Path("/tmp"), index=None, max_concurrent_streams=n)
+    s = WebRTCPeerSession(_FakePC(), t._ctx, peer_id="p")
+    return s._transcode_semaphore()._value
+
+
+@pytest.mark.parametrize("n,expect", [(None, MAX_CONCURRENT_TRANSCODES), (3, 3), (20, 20)])
+def test_the_configured_number_is_the_semaphore(n, expect):
+    assert asyncio.run(_run(n)) == expect
+
+
+async def _run(n):
+    return _semaphore_size(n)
+
+
+def test_the_budget_is_shared_between_peers():
+    """One budget for the node, not one per browser.
+
+    Building it per call would cap nothing: every viewer would arrive with a
+    full allowance and the node would spawn ffmpeg without limit.
+    """
+    t = WebRTCTransport(
+        sk_node=None, hub_pk_pem=b"", gek=b"\0" * 32,
+        shared_root=Path("/tmp"), index=None, max_concurrent_streams=2)
+
+    async def go():
+        a = WebRTCPeerSession(_FakePC(), t._ctx, peer_id="a")
+        b = WebRTCPeerSession(_FakePC(), t._ctx, peer_id="b")
+        sem_a, sem_b = a._transcode_semaphore(), b._transcode_semaphore()
+        assert sem_a is sem_b, "each peer got its own budget, so there is no cap"
+        await sem_a.acquire()
+        assert sem_b._value == 1, "one peer's stream did not spend the node's budget"
+
+    asyncio.run(go())
+
+
+def test_the_daemon_passes_it(  ):
+    """The join that syntax checking cannot see.
+
+    `self.cfg` parses and imports perfectly well; it raises AttributeError the
+    first time somebody plays a video, which is not where anyone would look.
+    """
+    daemon = (Path(__file__).resolve().parents[1] / "src" / "meshbay_node"
+              / "daemon.py").read_text()
+    i = daemon.index("WebRTCTransport(")
+    call = daemon[i:daemon.index(")", daemon.index("denylist=denylist", i))]
+    assert "max_concurrent_streams=" in call, (
+        "the daemon builds the transport without the operator's setting, so "
+        "node.toml is read and then ignored")
+    assert "self._config.node.max_concurrent_streams" in call, (
+        "the daemon holds its config in _config; any other attribute is an "
+        "AttributeError deferred until someone plays a video")