diff options
Diffstat (limited to 'packages/meshbay-node/tests')
7 files changed, 408 insertions, 87 deletions
diff --git a/packages/meshbay-node/tests/test_indexer.py b/packages/meshbay-node/tests/test_indexer.py index 68ccc4c..3eac39e 100644 --- a/packages/meshbay-node/tests/test_indexer.py +++ b/packages/meshbay-node/tests/test_indexer.py @@ -356,6 +356,82 @@ async def test_progress_stops_even_when_hashing_raises(tmp_path, sk_node, gek): "an exception mid-scan must not leave the scanning flag stuck on" +@pytest.mark.asyncio +async def test_realtime_watchdog_add_reports_progress_like_a_bulk_scan(tmp_path, sk_node, gek): + """ + Found live: dropping a whole season into an already-watched folder gave + no scanning indicator and no progress bar at all — only the initial scan + and the periodic reconcile backstop ever touched `progress`, never the + real-time per-file watchdog path (_schedule_update/_debounce/ + _update_entry). Drives that path directly (as _WatchdogHandler would), + without a real filesystem observer, exactly like test_root_availability.py + already does for _update_entry alone. + """ + d = tmp_path / "shared" + d.mkdir() + paths = [] + for i in range(3): + p = d / f"ep{i}.mkv" + p.write_bytes(os.urandom(1024 * (i + 1))) + paths.append(p) + total_size = sum(p.stat().st_size for p in paths) + + indexer = DirectoryIndexer(roots=one_root(d), group_id="g", sk_node=sk_node, + gek=gek, debounce_secs=0.01) + indexer._loop = asyncio.get_running_loop() + assert indexer.progress.scanning is False + + for p in paths: + indexer._schedule_update(p) + # _schedule_update only posts _debounce via call_soon_threadsafe (it is + # written to be called from the watchdog thread) — give the loop one + # turn to actually run the three posted calls before asserting on them. + await asyncio.sleep(0) + + # All three scheduled near-simultaneously (as a burst of watchdog events + # for one `mv` would arrive) — the indicator must flip on immediately, + # before any single file has actually finished hashing. + assert indexer.progress.scanning is True + assert indexer.progress.total_bytes == total_size + assert indexer.progress.scanned_bytes == 0 + + await asyncio.sleep(0.05) # past debounce_secs; lets all three fire and finish + + assert indexer.progress.scanning is False, "must end idle, not stuck scanning" + assert indexer.progress.scanned_bytes == total_size + assert indexer.progress.total_bytes == total_size + assert len(indexer.index.entries) == 3 + + +@pytest.mark.asyncio +async def test_realtime_watchdog_rapid_rewrite_does_not_double_count(tmp_path, sk_node, gek): + """A file rewritten during its own debounce window (on_modified firing + again before the first timer fires) must count its size once, not once + per event — the old timer is cancelled, and its accounting must transfer + to whichever fire() actually runs rather than being counted twice.""" + d = tmp_path / "shared" + d.mkdir() + p = d / "ep0.mkv" + p.write_bytes(os.urandom(2048)) + + indexer = DirectoryIndexer(roots=one_root(d), group_id="g", sk_node=sk_node, + gek=gek, debounce_secs=0.05) + indexer._loop = asyncio.get_running_loop() + + indexer._schedule_update(p) + indexer._schedule_update(p) # re-triggered before the first timer fires + indexer._schedule_update(p) + await asyncio.sleep(0) # let the three posted _debounce calls actually run + + assert indexer.progress.total_bytes == p.stat().st_size, \ + "one file re-triggered must count its size once, not three times" + + await asyncio.sleep(0.1) + + assert indexer.progress.scanning is False + assert indexer.progress.scanned_bytes == p.stat().st_size + + # ── Off-loop directory walks, reconcile backoff ───────────────────────────── @pytest.mark.asyncio diff --git a/packages/meshbay-node/tests/test_season_and_search_requests.py b/packages/meshbay-node/tests/test_season_and_search_requests.py index 8ba55fb..856797e 100644 --- a/packages/meshbay-node/tests/test_season_and_search_requests.py +++ b/packages/meshbay-node/tests/test_season_and_search_requests.py @@ -20,6 +20,7 @@ pytestmark = pytest.mark.asyncio def _session(media_cache=None, tmdb_client=None) -> WebRTCPeerSession: session = WebRTCPeerSession.__new__(WebRTCPeerSession) session._ctx = {"media_cache": media_cache, "tmdb_client": tmdb_client} + session._group_id = None session.sent = [] session._send = session.sent.append return session diff --git a/packages/meshbay-node/tests/test_stream_audio_transcode.py b/packages/meshbay-node/tests/test_stream_audio_transcode.py index dde0df4..aaf1595 100644 --- a/packages/meshbay-node/tests/test_stream_audio_transcode.py +++ b/packages/meshbay-node/tests/test_stream_audio_transcode.py @@ -149,7 +149,7 @@ async def test_probe_video_reports_aac_regardless_of_source_audio_codec(tmp_path clip = tmp_path / "clip.mkv" _make_clip(clip, acodec="eac3", channels=6) - codec, duration, has_audio, width, height = await _probe_video(str(clip)) + codec, duration, has_audio, width, height, raw_codec = await _probe_video(str(clip)) assert has_audio is True assert duration > 0 @@ -157,6 +157,7 @@ async def test_probe_video_reports_aac_regardless_of_source_audio_codec(tmp_path assert "eac3" not in codec and "ec-3" not in codec assert "mp4a.40.2" in codec assert (width, height) == (320, 240) + assert raw_codec == "h264" async def test_probe_video_handles_no_audio_track(tmp_path): @@ -168,9 +169,10 @@ async def test_probe_video_handles_no_audio_track(tmp_path): check=True, capture_output=True, ) - codec, duration, has_audio, width, height = await _probe_video(str(clip)) + codec, duration, has_audio, width, height, raw_codec = await _probe_video(str(clip)) assert has_audio is False assert codec is not None and "," not in codec, \ "no audio track must not produce a dangling ',' or a fake audio codec" assert (width, height) == (320, 240) + assert raw_codec == "h264" diff --git a/packages/meshbay-node/tests/test_stream_hevc_transcode.py b/packages/meshbay-node/tests/test_stream_hevc_transcode.py new file mode 100644 index 0000000..b3f0474 --- /dev/null +++ b/packages/meshbay-node/tests/test_stream_hevc_transcode.py @@ -0,0 +1,154 @@ +""" +HEVC video is transcoded to H264 for streaming, never copied — unlike the +codecs media_probe.py's BROWSER_INCOMPATIBLE_VIDEO_CODECS excludes. + +Found live: a real HEVC/EAC3 WEB-DL streamed fine over MNP (ffprobe/VLC play +it) but the browser reported "Codec not supported for streaming: +hev1.1.6.L93.B0,mp4a.40.2" from MediaSource.isTypeSupported — Chrome has no +HEVC decoder on most non-Apple platforms. "-c:v copy" on an incompatible +codec is not a mux failure the way EAC3 audio is (test_stream_audio_ +transcode.py); ffmpeg happily remuxes it, and the browser is the one that +then refuses it, silently, at playback rather than at stream_init. + +These tests spawn real ffmpeg/ffprobe against small synthetic files (lavfi +test sources, ~1s), the same style as test_stream_audio_transcode.py. +""" + +import shutil +import subprocess +from pathlib import Path + +import pytest +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + +from meshbay_common.crypto import generate_gek +from meshbay_common.webcrypto import chunk_key_aes, decrypt_chunk_aes +from meshbay_node.indexer.group_index import GroupIndex +from meshbay_node.transport.webrtc_server import WebRTCPeerSession, _probe_video + +from conftest import one_root + +_HAVE_FFMPEG = shutil.which("ffmpeg") and shutil.which("ffprobe") +_HAVE_HEVC_ENCODER = _HAVE_FFMPEG and b"libx265" in subprocess.run( + ["ffmpeg", "-hide_banner", "-encoders"], capture_output=True).stdout +pytestmark = [ + pytest.mark.asyncio, + pytest.mark.skipif(not _HAVE_HEVC_ENCODER, reason="ffmpeg/libx265 not installed"), +] + + +def _make_hevc_clip(path: Path) -> None: + """~1s of HEVC video + AAC audio — a minimal stand-in for a real HEVC WEB-DL.""" + subprocess.run( + ["ffmpeg", "-hide_banner", "-loglevel", "error", "-y", + "-f", "lavfi", "-i", "testsrc=size=320x240:rate=25:duration=1", + "-f", "lavfi", "-i", "sine=frequency=440:duration=1:sample_rate=48000", + "-c:v", "libx265", "-preset", "ultrafast", "-c:a", "aac", + str(path)], + check=True, capture_output=True, + ) + + +def _session(video_path: Path, gek: bytes, *, transcode_incompatible_video: bool = True): + import blake3 + file_bytes = video_path.read_bytes() + file_id = blake3.blake3(file_bytes).hexdigest() + + sk_node = Ed25519PrivateKey.generate() + index = GroupIndex(group_id="g" * 32, sk_node=sk_node, gek=gek) + from meshbay_common.protocol import IndexEntry + index.add_entry(IndexEntry( + id=file_id, name=video_path.name, path=video_path.parent.name, + size=len(file_bytes), type="video", added_at=0)) + + session = WebRTCPeerSession.__new__(WebRTCPeerSession) + session._ctx = { + "roots": one_root(video_path.parent), + "index": index, + "gek": gek, + "sk_node": sk_node, + "max_concurrent_streams": 4, + "transcode_incompatible_video": transcode_incompatible_video, + } + session._group_id = None + session._user_id = "tester" + session._stream_stopped = False + session._stream_keepalives = 0 + session.sent = [] + session._send = session.sent.append + session._audit = lambda *a, **k: None + return session, file_id + + +def _reassemble(sent: list[dict], gek: bytes, file_id: str) -> bytes: + file_hash = bytes.fromhex(file_id) + segments = sorted( + (m for m in sent if m.get("type") == "stream_data"), + key=lambda m: m["segment_index"]) + out = b"" + for m in segments: + key = chunk_key_aes(gek, file_hash, m["segment_index"]) + out += decrypt_chunk_aes(key, m["nonce"], m["ct"]) + return out + + +def _output_video_codec(path: Path) -> str: + probe = subprocess.run( + ["ffprobe", "-v", "error", "-select_streams", "v:0", + "-show_entries", "stream=codec_name", "-of", "csv=p=0", str(path)], + check=True, capture_output=True, text=True) + return probe.stdout.strip() + + +async def test_hevc_video_is_transcoded_to_h264_by_default(tmp_path): + clip = tmp_path / "clip.mkv" + _make_hevc_clip(clip) + gek = generate_gek() + session, file_id = _session(clip, gek) + + await session._stream_video_inner({"file_id": file_id, "start": 0, "credits": 0}) + + errors = [m for m in session.sent if m.get("type") == "error"] + assert not errors, f"streaming must not fail: {errors}" + + init = next(m for m in session.sent if m.get("type") == "stream_init") + assert init["codec"].startswith("avc1."), ( + "the reported codec must be the transcoded H264 string, never the " + f"source's hev1 string a browser cannot decode: {init['codec']}") + + remuxed = _reassemble(session.sent, gek, file_id) + out_path = tmp_path / "out.mp4" + out_path.write_bytes(remuxed) + assert _output_video_codec(out_path) == "h264", \ + "the bytes on the wire must actually be H264, not just the reported label" + + +async def test_hevc_transcode_can_be_disabled_by_the_operator(tmp_path): + clip = tmp_path / "clip.mkv" + _make_hevc_clip(clip) + gek = generate_gek() + session, file_id = _session(clip, gek, transcode_incompatible_video=False) + + await session._stream_video_inner({"file_id": file_id, "start": 0, "credits": 0}) + + assert not [m for m in session.sent if m.get("type") == "error"] + init = next(m for m in session.sent if m.get("type") == "stream_init") + assert init["codec"].startswith("hev1."), ( + "with the fallback disabled, the source is copied as-is and the " + f"original HEVC codec string must be reported unchanged: {init['codec']}") + + remuxed = _reassemble(session.sent, gek, file_id) + out_path = tmp_path / "out.mp4" + out_path.write_bytes(remuxed) + assert _output_video_codec(out_path) == "hevc", \ + "with the fallback disabled, the wire bytes must still be copied HEVC" + + +async def test_probe_video_reports_raw_codec_name_for_hevc(tmp_path): + clip = tmp_path / "clip.mkv" + _make_hevc_clip(clip) + + codec, duration, has_audio, width, height, raw_codec = await _probe_video(str(clip)) + + assert raw_codec == "hevc" + assert codec is not None and codec.startswith("hev1.") diff --git a/packages/meshbay-node/tests/test_tmdb.py b/packages/meshbay-node/tests/test_tmdb.py index 4500feb..fcbc2a2 100644 --- a/packages/meshbay-node/tests/test_tmdb.py +++ b/packages/meshbay-node/tests/test_tmdb.py @@ -7,14 +7,12 @@ from meshbay_node.tmdb import TmdbClient class FakeRoster: - def __init__(self, enabled: bool = True, token: str | None = "fake-token", - language: str | None = None): - self._enabled = enabled + def __init__(self, token: str | None = "fake-token", language: str | None = None): self._token = token self._language = language async def tmdb_config(self): - return self._enabled, self._token, self._language + return self._token, self._language def _handler(response_map): @@ -106,25 +104,6 @@ async def test_no_results_returns_none_and_zero_confidence(): @pytest.mark.asyncio -async def test_disabled_via_roster_setting_makes_no_request(): - calls = [] - - def handle(request: httpx.Request) -> httpx.Response: - calls.append(request) - return httpx.Response(200, json={"results": []}) - - client = TmdbClient( - roster=FakeRoster(enabled=False), - transport=httpx.MockTransport(handle), - ) - result, ratio = await client.search_movie("Anything") - - assert result is None - assert calls == [] # confirms the disabled check short-circuits before any request - await client.close() - - -@pytest.mark.asyncio async def test_no_token_resolvable_makes_no_request(monkeypatch): monkeypatch.delenv("MESHBAY_TMDB_DEFAULT_TOKEN", raising=False) calls = [] @@ -134,7 +113,7 @@ async def test_no_token_resolvable_makes_no_request(monkeypatch): return httpx.Response(200, json={"results": []}) client = TmdbClient( - roster=FakeRoster(enabled=True, token=None), + roster=FakeRoster(token=None), transport=httpx.MockTransport(handle), ) result, ratio = await client.search_movie("Anything") diff --git a/packages/meshbay-node/tests/test_tmdb_config_policy.py b/packages/meshbay-node/tests/test_tmdb_config_policy.py index 29ef54c..c1781e3 100644 --- a/packages/meshbay-node/tests/test_tmdb_config_policy.py +++ b/packages/meshbay-node/tests/test_tmdb_config_policy.py @@ -1,9 +1,14 @@ """ -The operator decides whether the node calls TMDB at all, and whether it uses -a custom API token — docs/mediacenter.md §5.5. Same shape as -test_apps_enabled_policy.py/test_scan_settings_policy.py: a signed operator -instruction, node-wide (group_id="") rather than per-group, stored via -roster.py's group_settings table. +The operator's custom TMDB API token and query language — docs/mediacenter.md +§5.5. Same shape as test_apps_enabled_policy.py/test_scan_settings_policy.py: +a signed operator instruction, node-wide (group_id="") rather than per-group, +stored via roster.py's group_settings table. + +Whether TMDB is used *at all* used to live in this same op — moved to its +own per-group op (test_tmdb_enabled_policy.py, 2026-08-24): a real +media-library group and a test/demo group on the same node need not share +that decision, while the token and query language stay one operator's +shared credential/cache. Specific to this one: the subject signed/audited must never contain the token itself (it would end up in the audit log in plaintext) — only whether @@ -55,51 +60,39 @@ def _fake_challenge(issued: list): # ── Refused before a challenge is even issued ─────────────────────────────── -async def test_missing_enabled_is_refused(tmp_path): +async def test_non_string_token_is_refused(tmp_path): session = _session(tmp_path, "op", operator="op") session._has_admin_authority = lambda: True issued = [] session._issue_admin_challenge = _fake_challenge(issued) - session._do_tmdb_config({}) + session._do_tmdb_config({"token": 12345}) assert not issued assert [m for m in session.sent if m.get("type") == "error"] -async def test_non_bool_enabled_is_refused(tmp_path): - session = _session(tmp_path, "op", operator="op") - session._has_admin_authority = lambda: True - issued = [] - session._issue_admin_challenge = _fake_challenge(issued) +async def test_a_request_with_nobody_to_authorize_it_is_refused(tmp_path): + session = _session(tmp_path, "member-1", operator="the-operator") + session._has_admin_authority = lambda: False - session._do_tmdb_config({"enabled": "yes"}) + session._do_tmdb_config({"token": "x"}) - assert not issued assert [m for m in session.sent if m.get("type") == "error"] -async def test_non_string_token_is_refused(tmp_path): +async def test_non_string_language_is_refused(tmp_path): session = _session(tmp_path, "op", operator="op") session._has_admin_authority = lambda: True issued = [] session._issue_admin_challenge = _fake_challenge(issued) - session._do_tmdb_config({"enabled": True, "token": 12345}) + session._do_tmdb_config({"language": 42}) assert not issued assert [m for m in session.sent if m.get("type") == "error"] -async def test_a_request_with_nobody_to_authorize_it_is_refused(tmp_path): - session = _session(tmp_path, "member-1", operator="the-operator") - session._has_admin_authority = lambda: False - - session._do_tmdb_config({"enabled": False}) - - assert [m for m in session.sent if m.get("type") == "error"] - - # ── Who may change it, and what gets signed ───────────────────────────────── async def test_changing_it_needs_a_signature(tmp_path): @@ -108,7 +101,7 @@ async def test_changing_it_needs_a_signature(tmp_path): issued = [] session._issue_admin_challenge = _fake_challenge(issued) - session._do_tmdb_config({"enabled": True}) + session._do_tmdb_config({}) assert len(issued) == 1 op, subject, payload, group_id = issued[0] @@ -125,23 +118,23 @@ async def test_the_token_itself_never_appears_in_the_signed_subject(tmp_path): session._issue_admin_challenge = _fake_challenge(issued) secret = "sk-super-secret-tmdb-token" - session._do_tmdb_config({"enabled": True, "token": secret}) + session._do_tmdb_config({"token": secret}) _, subject, payload, _ = issued[0] assert secret not in subject assert payload["token"] == secret, "the real value still has to reach the exec step somehow" -async def test_subject_reflects_enabled_and_whether_a_token_was_supplied(tmp_path): +async def test_subject_reflects_whether_a_token_was_supplied(tmp_path): session = _session(tmp_path, "op", operator="op") session._has_admin_authority = lambda: True issued = [] session._issue_admin_challenge = _fake_challenge(issued) - session._do_tmdb_config({"enabled": False, "token": "x"}) + session._do_tmdb_config({"token": "x"}) _, subject, _, _ = issued[0] - assert subject == "enabled=False,custom_token=yes,language=default" + assert subject == "custom_token=yes,language=default" async def test_subject_says_no_custom_token_when_none_given(tmp_path): @@ -150,10 +143,10 @@ async def test_subject_says_no_custom_token_when_none_given(tmp_path): issued = [] session._issue_admin_challenge = _fake_challenge(issued) - session._do_tmdb_config({"enabled": True}) + session._do_tmdb_config({}) _, subject, _, _ = issued[0] - assert subject == "enabled=True,custom_token=no,language=default" + assert subject == "custom_token=no,language=default" async def test_subject_reflects_a_configured_language(tmp_path): @@ -162,44 +155,32 @@ async def test_subject_reflects_a_configured_language(tmp_path): issued = [] session._issue_admin_challenge = _fake_challenge(issued) - session._do_tmdb_config({"enabled": True, "language": "fr-FR"}) + session._do_tmdb_config({"language": "fr-FR"}) _, subject, payload, _ = issued[0] - assert subject == "enabled=True,custom_token=no,language=fr-FR" + assert subject == "custom_token=no,language=fr-FR" assert payload["language"] == "fr-FR" -async def test_non_string_language_is_refused(tmp_path): - session = _session(tmp_path, "op", operator="op") - session._has_admin_authority = lambda: True - issued = [] - session._issue_admin_challenge = _fake_challenge(issued) - - session._do_tmdb_config({"enabled": True, "language": 42}) - - assert not issued - assert [m for m in session.sent if m.get("type") == "error"] - - # ── Where it is stored ────────────────────────────────────────────────────── async def test_the_setting_lives_on_the_node_and_survives_a_restart(tmp_path): roster = Roster(db_path=tmp_path / "roster.db") await roster.open() try: - enabled, token, language = await roster.tmdb_config() - assert (enabled, token, language) == (True, None, None), ( - "absent must mean on, with the shipped default token, TMDB's own default language") - await roster.set_tmdb_config(True, "my-custom-token", "fr-FR", set_by="op") - enabled, token, language = await roster.tmdb_config() - assert (enabled, token, language) == (True, "my-custom-token", "fr-FR") + token, language = await roster.tmdb_config() + assert (token, language) == (None, None), ( + "absent must mean the shipped default token, TMDB's own default language") + await roster.set_tmdb_config("my-custom-token", "fr-FR", set_by="op") + token, language = await roster.tmdb_config() + assert (token, language) == ("my-custom-token", "fr-FR") finally: await roster.close() reopened = Roster(db_path=tmp_path / "roster.db") await reopened.open() try: - assert await reopened.tmdb_config() == (True, "my-custom-token", "fr-FR") + assert await reopened.tmdb_config() == ("my-custom-token", "fr-FR") finally: await reopened.close() @@ -208,11 +189,11 @@ async def test_clearing_the_token_reverts_to_the_default(tmp_path): roster = Roster(db_path=tmp_path / "roster.db") await roster.open() try: - await roster.set_tmdb_config(True, "a-token", set_by="op") - assert (await roster.tmdb_config())[1] == "a-token" + await roster.set_tmdb_config("a-token", set_by="op") + assert (await roster.tmdb_config())[0] == "a-token" - await roster.set_tmdb_config(True, "", set_by="op") - enabled, token, language = await roster.tmdb_config() + await roster.set_tmdb_config("", set_by="op") + token, language = await roster.tmdb_config() assert token is None, "an explicit empty string clears the custom token" finally: await roster.close() @@ -222,9 +203,9 @@ async def test_omitting_the_token_leaves_it_unchanged(tmp_path): roster = Roster(db_path=tmp_path / "roster.db") await roster.open() try: - await roster.set_tmdb_config(True, "a-token", set_by="op") - await roster.set_tmdb_config(False, None, set_by="op") - enabled, token, language = await roster.tmdb_config() - assert (enabled, token) == (False, "a-token") + await roster.set_tmdb_config("a-token", set_by="op") + await roster.set_tmdb_config(set_by="op") + token, language = await roster.tmdb_config() + assert token == "a-token" finally: await roster.close() diff --git a/packages/meshbay-node/tests/test_tmdb_enabled_policy.py b/packages/meshbay-node/tests/test_tmdb_enabled_policy.py new file mode 100644 index 0000000..8e945ad --- /dev/null +++ b/packages/meshbay-node/tests/test_tmdb_enabled_policy.py @@ -0,0 +1,128 @@ +""" +Whether TMDB lookups run *at all* for a group — docs/mediacenter.md §5.5. +Per-group (2026-08-24 — used to be node-wide, folded into tmdb_config): a +real media-library group and a test/demo group on the same node need not +share the decision to spend TMDB quota and make outbound requests. Same +shape as test_video_root_policy.py: a signed operator instruction, scoped to +self._group_id (not passed explicitly on the wire), stored via roster.py's +group_settings table under the real group_id. + +The custom API token and query language stay node-wide — see +test_tmdb_config_policy.py for those. +""" + +from pathlib import Path + +import pytest + +from meshbay_common.adminop import OP_TMDB_ENABLED +from meshbay_node.indexer.group_index import GroupIndex +from meshbay_node.roster import Roster +from meshbay_node.transport.webrtc_server import WebRTCPeerSession +from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey + +from conftest import one_root + +pytestmark = pytest.mark.asyncio + + +def _session(tmp_path: Path, user_id: str, *, operator: str | None = None) -> WebRTCPeerSession: + shared_root = tmp_path / "shared" + shared_root.mkdir(exist_ok=True) + index = GroupIndex(group_id="g" * 32, sk_node=Ed25519PrivateKey.generate()) + ctx = { + "roots": one_root(shared_root), + "index": index, + "sk_node": index.sk_node, + "node_user_id": operator, + } + session = WebRTCPeerSession.__new__(WebRTCPeerSession) + session._ctx = ctx + session._group_id = "g" * 32 + session._user_id = user_id + session._pk_user = "" + session.sent = [] + session._send = session.sent.append + session._audit = lambda *a, **k: None + return session + + +# ── Refused before a challenge is even issued ─────────────────────────────── + +async def test_missing_enabled_is_refused(tmp_path): + session = _session(tmp_path, "op", operator="op") + session._has_admin_authority = lambda: True + issued = [] + session._issue_admin_challenge = lambda op, subject: issued.append((op, subject)) + + session._do_tmdb_enabled({}) + + assert not issued + assert [m for m in session.sent if m.get("type") == "error"] + + +async def test_non_bool_enabled_is_refused(tmp_path): + session = _session(tmp_path, "op", operator="op") + session._has_admin_authority = lambda: True + issued = [] + session._issue_admin_challenge = lambda op, subject: issued.append((op, subject)) + + session._do_tmdb_enabled({"enabled": "yes"}) + + assert not issued + assert [m for m in session.sent if m.get("type") == "error"] + + +async def test_a_request_with_nobody_to_authorize_it_is_refused(tmp_path): + session = _session(tmp_path, "member-1", operator="the-operator") + session._has_admin_authority = lambda: False + + session._do_tmdb_enabled({"enabled": False}) + + assert [m for m in session.sent if m.get("type") == "error"] + + +# ── Accepted cases ─────────────────────────────────────────────────────────── + +async def test_a_valid_request_is_signed_against_this_groups_id(tmp_path): + session = _session(tmp_path, "op", operator="op") + session._has_admin_authority = lambda: True + issued = [] + session._issue_admin_challenge = lambda op, subject: issued.append((op, subject)) + + session._do_tmdb_enabled({"enabled": True}) + + assert issued == [(OP_TMDB_ENABLED, "True")] + + +async def test_disabling_is_signed_too(tmp_path): + session = _session(tmp_path, "op", operator="op") + session._has_admin_authority = lambda: True + issued = [] + session._issue_admin_challenge = lambda op, subject: issued.append((op, subject)) + + session._do_tmdb_enabled({"enabled": False}) + + assert issued == [(OP_TMDB_ENABLED, "False")] + + +# ── Where it is stored ────────────────────────────────────────────────────── + +async def test_the_setting_lives_on_the_node_and_survives_a_restart(tmp_path): + roster = Roster(db_path=tmp_path / "roster.db") + await roster.open() + try: + assert await roster.tmdb_enabled("g1") is True, "absent must mean on" + await roster.set_tmdb_enabled("g1", False, set_by="op") + assert await roster.tmdb_enabled("g1") is False + finally: + await roster.close() + + reopened = Roster(db_path=tmp_path / "roster.db") + await reopened.open() + try: + assert await reopened.tmdb_enabled("g1") is False + assert await reopened.tmdb_enabled("g2") is True, \ + "one group's setting must not answer for another" + finally: + await reopened.close() |