diff options
Diffstat (limited to 'packages/meshbay-node/tests')
| -rw-r--r-- | packages/meshbay-node/tests/test_enrich_audio.py | 150 | ||||
| -rw-r--r-- | packages/meshbay-node/tests/test_media_cache.py | 53 | ||||
| -rw-r--r-- | packages/meshbay-node/tests/test_musicbrainz.py | 163 | ||||
| -rw-r--r-- | packages/meshbay-node/tests/test_musicbrainz_config_policy.py | 179 | ||||
| -rw-r--r-- | packages/meshbay-node/tests/test_musicbrainz_enabled_policy.py | 126 |
5 files changed, 669 insertions, 2 deletions
diff --git a/packages/meshbay-node/tests/test_enrich_audio.py b/packages/meshbay-node/tests/test_enrich_audio.py new file mode 100644 index 0000000..62d0a0a --- /dev/null +++ b/packages/meshbay-node/tests/test_enrich_audio.py @@ -0,0 +1,150 @@ +"""Tests for indexer/enrich_audio.py — tag/cover extraction and the end-to-end pool.""" + +import asyncio +import shutil +import subprocess +from pathlib import Path + +import pytest + +from meshbay_common.protocol import IndexEntry +from meshbay_node.indexer.enrich_audio import ( + AudioEnricher, _artist_album_from_ancestors, _extract_cover, +) +from meshbay_node.indexer.title_parse import parse_track_filename +from meshbay_node.media_cache import MediaCache + +_HAVE_FFMPEG = shutil.which("ffmpeg") and shutil.which("ffprobe") + + +# ── pure helpers, no ffmpeg/mutagen file needed ────────────────────────────── + +def test_parse_track_filename_splits_leading_track_number(): + parsed = parse_track_filename("01 - Venus As A Boy (Edited Lp Version).mp3") + assert parsed.track_no == 1 + assert parsed.title == "Venus As A Boy (Edited Lp Version)" + + +def test_parse_track_filename_handles_dot_separated(): + parsed = parse_track_filename("03. Human Behaviour.mp3") + assert parsed.track_no == 3 + assert parsed.title == "Human Behaviour" + + +def test_parse_track_filename_handles_underscore_separated(): + parsed = parse_track_filename("12_Some_Title.mp3") + assert parsed.track_no == 12 + assert parsed.title == "Some Title" + + +def test_parse_track_filename_no_prefix_leaves_track_no_none(): + parsed = parse_track_filename("Some Title.mp3") + assert parsed.track_no is None + assert parsed.title == "Some Title" + + +def test_parse_track_filename_does_not_mistake_a_leading_year_for_a_track_number(): + parsed = parse_track_filename("1999 - Some Title.mp3") + assert parsed.track_no is None, "a 4-digit prefix is capped out, not read as track 199" + + +def test_artist_album_from_ancestors_reads_artist_album_track_layout(tmp_path): + folder = tmp_path / "Some Artist" / "Some Album" + folder.mkdir(parents=True) + track = folder / "01 - A Track.mp3" + track.touch() + + artist, album = _artist_album_from_ancestors(track) + + assert artist == "Some Artist" + assert album == "Some Album" + + +# ── end-to-end against a real (tiny, synthetic) MP3 file ──────────────────── + +pytestmark_ffmpeg = pytest.mark.skipif(not _HAVE_FFMPEG, reason="ffmpeg/ffprobe not installed") + + +def _make_clip(path: Path, *, title=None, artist=None, album=None, track=None) -> None: + subprocess.run( + ["ffmpeg", "-hide_banner", "-loglevel", "error", "-y", + "-f", "lavfi", "-i", "sine=frequency=440:duration=1", + "-c:a", "libmp3lame", "-b:a", "64k", + *(["-metadata", f"title={title}"] if title else []), + *(["-metadata", f"artist={artist}"] if artist else []), + *(["-metadata", f"album={album}"] if album else []), + *(["-metadata", f"track={track}"] if track else []), + str(path)], + check=True, capture_output=True, + ) + + +@pytest.fixture +async def media_cache(tmp_path): + c = MediaCache(db_path=tmp_path / "media_cache.db") + await c.open() + yield c + await c.close() + + +@pytestmark_ffmpeg +@pytest.mark.asyncio +async def test_enricher_prefers_tags_over_filename_parse(tmp_path, media_cache): + clip = tmp_path / "99 - wrong title.mp3" + _make_clip(clip, title="Real Title", artist="Real Artist", album="Real Album", track=3) + entry = IndexEntry(id="fileid1", name=clip.name, path=clip.name, + size=clip.stat().st_size, type="audio", added_at=0) + + enricher = AudioEnricher(media_cache) + done = asyncio.get_event_loop().create_future() + + async def on_done(file_id, fields): + done.set_result((file_id, fields)) + + enricher.spawn(entry, clip, on_done) + file_id, fields = await asyncio.wait_for(done, timeout=30) + + assert file_id == "fileid1" + assert fields["display_title"] == "Real Title" + assert fields["artist"] == "Real Artist" + assert fields["album"] == "Real Album" + assert fields["track_no"] == 3 + assert fields["duration"] == 1 + assert fields.get("thumb_hash") is None, "no embedded cover was written in this clip" + + +@pytestmark_ffmpeg +@pytest.mark.asyncio +async def test_enricher_falls_back_to_filename_and_folder_when_tags_absent(tmp_path, media_cache): + folder = tmp_path / "Folder Artist" / "Folder Album" + folder.mkdir(parents=True) + clip = folder / "05 - Filename Title.mp3" + _make_clip(clip) # no metadata tags at all + entry = IndexEntry(id="fileid2", name=clip.name, + path=str(clip.relative_to(tmp_path)), + size=clip.stat().st_size, type="audio", added_at=0) + + enricher = AudioEnricher(media_cache) + done = asyncio.get_event_loop().create_future() + + async def on_done(file_id, fields): + done.set_result((file_id, fields)) + + enricher.spawn(entry, clip, on_done) + _, fields = await asyncio.wait_for(done, timeout=30) + + assert fields["display_title"] == "Filename Title" + assert fields["track_no"] == 5 + assert fields["artist"] == "Folder Artist" + assert fields["album"] == "Folder Album" + + +@pytestmark_ffmpeg +def test_extract_cover_returns_none_when_no_apic_frame(tmp_path): + from mutagen import File as MutagenFile + + clip = tmp_path / "plain.mp3" + _make_clip(clip) + + mf = MutagenFile(str(clip)) + assert _extract_cover(mf) is None diff --git a/packages/meshbay-node/tests/test_media_cache.py b/packages/meshbay-node/tests/test_media_cache.py index b66c448..e70ef43 100644 --- a/packages/meshbay-node/tests/test_media_cache.py +++ b/packages/meshbay-node/tests/test_media_cache.py @@ -1,10 +1,10 @@ -"""Tests for media_cache.py — TMDB/thumbnail cache and its pruning obligation.""" +"""Tests for media_cache.py — TMDB/MusicBrainz/thumbnail cache and its pruning obligation.""" import time import pytest -from meshbay_node.media_cache import MediaCache, TMDB_META_TTL_SECS +from meshbay_node.media_cache import MediaCache, TMDB_META_TTL_SECS, MUSICBRAINZ_META_TTL_SECS @pytest.fixture @@ -69,3 +69,52 @@ async def test_prune_file_removes_thumb_and_mapping_but_not_shared_meta(cache): # ep2's own mapping and the shared show metadata both survive assert await cache.get_file_tmdb("ep2") == ("555", "tv") assert await cache.get_tmdb_meta("555", "tv") == {"name": "A Show"} + + +# ── Music app (docs/musicbay.md §6) — file_mbid/mbid_meta ──────────────────── + +@pytest.mark.asyncio +async def test_file_mbid_round_trip(cache): + assert await cache.get_file_mbid("file1") is None + + await cache.set_file_mbid("file1", "release-mbid-1") + + assert await cache.get_file_mbid("file1") == "release-mbid-1" + + +@pytest.mark.asyncio +async def test_mbid_meta_round_trip(cache): + assert await cache.get_mbid_meta("release-mbid-1") is None + + await cache.set_mbid_meta("release-mbid-1", {"artist": "Some Artist", "album": "An Album"}) + + meta = await cache.get_mbid_meta("release-mbid-1") + assert meta == {"artist": "Some Artist", "album": "An Album"} + + +@pytest.mark.asyncio +async def test_mbid_meta_expires_after_ttl(cache): + await cache._db.execute( + "INSERT INTO mbid_meta (mbid, json, fetched_at) VALUES (?, ?, ?)", + ("old-mbid", '{"album": "Old"}', time.time() - MUSICBRAINZ_META_TTL_SECS - 1), + ) + await cache._db.commit() + + assert await cache.get_mbid_meta("old-mbid") is None + + +@pytest.mark.asyncio +async def test_prune_file_removes_mbid_mapping_but_not_shared_meta(cache): + # Two tracks of the same release share one mbid_meta row. + await cache.set_file_mbid("track1", "release-mbid-1") + await cache.set_file_mbid("track2", "release-mbid-1") + await cache.set_mbid_meta("release-mbid-1", {"album": "An Album"}) + await cache.put_thumb("cover-hash", "track1", b"cover-bytes") + + await cache.prune_file("track1") + + assert await cache.get_file_mbid("track1") is None + assert await cache.get_thumb("cover-hash") is None + # track2's own mapping and the shared release metadata both survive + assert await cache.get_file_mbid("track2") == "release-mbid-1" + assert await cache.get_mbid_meta("release-mbid-1") == {"album": "An Album"} diff --git a/packages/meshbay-node/tests/test_musicbrainz.py b/packages/meshbay-node/tests/test_musicbrainz.py new file mode 100644 index 0000000..482395d --- /dev/null +++ b/packages/meshbay-node/tests/test_musicbrainz.py @@ -0,0 +1,163 @@ +"""Tests for musicbrainz.py against a mocked httpx transport — no live network in CI.""" + +import time + +import httpx +import pytest + +from meshbay_node.musicbrainz import MusicBrainzClient, _MIN_INTERVAL_SECS + + +class FakeRoster: + def __init__(self, contact: str | None = "operator@example.invalid"): + self._contact = contact + + async def musicbrainz_contact(self): + return self._contact + + +def _handler(response_map): + def handle(request: httpx.Request) -> httpx.Response: + path = request.url.path + for prefix, body in response_map.items(): + if path.endswith(prefix): + return httpx.Response(200, json=body) + return httpx.Response(404, json={}) + return handle + + +@pytest.mark.asyncio +async def test_search_release_returns_top_result_and_confidence(): + body = {"releases": [{"id": "abc-123", "title": "The Great Album", + "artist-credit": [{"name": "Some Artist"}]}]} + client = MusicBrainzClient( + roster=FakeRoster(), + transport=httpx.MockTransport(_handler({"release": body})), + ) + result, ratio = await client.search_release("Some Artist", "The Great Album") + + assert result is not None + assert result["id"] == "abc-123" + assert ratio > 0.9 + await client.close() + + +@pytest.mark.asyncio +async def test_no_results_returns_none_and_zero_confidence(): + client = MusicBrainzClient( + roster=FakeRoster(), + transport=httpx.MockTransport(_handler({"release": {"releases": []}})), + ) + result, ratio = await client.search_release("Nobody", "Nonexistent Obscure Album") + + assert result is None + assert ratio == 0.0 + await client.close() + + +@pytest.mark.asyncio +async def test_no_contact_configured_makes_no_request(monkeypatch): + monkeypatch.delenv("MESHBAY_MUSICBRAINZ_CONTACT_DEFAULT", raising=False) + calls = [] + + def handle(request: httpx.Request) -> httpx.Response: + calls.append(request) + return httpx.Response(200, json={"releases": []}) + + client = MusicBrainzClient( + roster=FakeRoster(contact=None), + transport=httpx.MockTransport(handle), + ) + result, ratio = await client.search_release("Anyone", "Anything") + + assert result is None + assert calls == [], "an unidentified client must never be sent — see musicbay.md §3.1" + await client.close() + + +@pytest.mark.asyncio +async def test_the_configured_contact_is_sent_as_user_agent(): + captured = {} + + def handle(request: httpx.Request) -> httpx.Response: + captured["ua"] = request.headers.get("user-agent") + return httpx.Response(200, json={"releases": []}) + + client = MusicBrainzClient( + roster=FakeRoster(contact="operator@example.invalid"), + transport=httpx.MockTransport(handle), + ) + await client.search_release("Anyone", "Anything") + + assert "operator@example.invalid" in captured["ua"] + await client.close() + + +@pytest.mark.asyncio +async def test_http_error_returns_none_gracefully(): + def handle(request: httpx.Request) -> httpx.Response: + return httpx.Response(500, json={"error": "server error"}) + + client = MusicBrainzClient( + roster=FakeRoster(), + transport=httpx.MockTransport(handle), + ) + result, ratio = await client.search_release("Anyone", "Anything") + + assert result is None + assert ratio == 0.0 + await client.close() + + +@pytest.mark.asyncio +async def test_cover_art_missing_returns_none_not_an_error(): + def handle(request: httpx.Request) -> httpx.Response: + return httpx.Response(404) + + client = MusicBrainzClient( + roster=FakeRoster(), + transport=httpx.MockTransport(handle), + ) + content = await client.fetch_cover_art("abc-123") + + assert content is None + await client.close() + + +@pytest.mark.asyncio +async def test_cover_art_found_returns_bytes(): + def handle(request: httpx.Request) -> httpx.Response: + return httpx.Response(200, content=b"\xff\xd8fake-jpeg-bytes") + + client = MusicBrainzClient( + roster=FakeRoster(), + transport=httpx.MockTransport(handle), + ) + content = await client.fetch_cover_art("abc-123") + + assert content == b"\xff\xd8fake-jpeg-bytes" + await client.close() + + +@pytest.mark.asyncio +async def test_calls_are_paced_at_least_min_interval_apart(): + """ + docs/musicbay.md §3.2: the ~1 req/s courtesy limit is this node's own + job, not something the server hands out — verified by timing two calls + back to back rather than mocking the clock, so a change to the pacing + implementation that still meets the contract doesn't break this test. + """ + def handle(request: httpx.Request) -> httpx.Response: + return httpx.Response(200, json={"releases": []}) + + client = MusicBrainzClient( + roster=FakeRoster(), + transport=httpx.MockTransport(handle), + ) + start = time.monotonic() + await client.search_release("A", "One") + await client.search_release("B", "Two") + elapsed = time.monotonic() - start + + assert elapsed >= _MIN_INTERVAL_SECS * 0.9 + await client.close() diff --git a/packages/meshbay-node/tests/test_musicbrainz_config_policy.py b/packages/meshbay-node/tests/test_musicbrainz_config_policy.py new file mode 100644 index 0000000..3590f01 --- /dev/null +++ b/packages/meshbay-node/tests/test_musicbrainz_config_policy.py @@ -0,0 +1,179 @@ +""" +The operator's MusicBrainz User-Agent contact string — docs/musicbay.md +§3.2/§6. Same shape as test_tmdb_config_policy.py: a signed operator +instruction, node-wide (group_id="") rather than per-group, stored via +roster.py's group_settings table. + +Unlike TMDB's token, a contact string is not a secret — MusicBrainz's usage +policy expects it to be visible to the service it's sent to — but the +subject signed/audited still only ever says whether one was configured +(never the address itself), the same "yes/no" shape as tmdb_config's +subject, to keep a personal contact out of the audit log as free text. +""" + +from pathlib import Path + +import pytest + +from meshbay_common.adminop import OP_MUSICBRAINZ_CONFIG +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 = None + session._user_id = user_id + session._pk_user = "" + session.sent = [] + session._send = session.sent.append + session._audit = lambda *a, **k: None + return session + + +def _fake_challenge(issued: list): + return lambda op, subject, payload=None, group_id=None: issued.append( + (op, subject, payload, group_id)) + + +# ── Refused before a challenge is even issued ─────────────────────────────── + +async def test_non_string_contact_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_musicbrainz_config({"contact": 12345}) + + 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_musicbrainz_config({"contact": "https://example.invalid/contact"}) + + 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): + session = _session(tmp_path, "op", operator="op") + session._has_admin_authority = lambda: True + issued = [] + session._issue_admin_challenge = _fake_challenge(issued) + + session._do_musicbrainz_config({}) + + assert len(issued) == 1 + op, subject, payload, group_id = issued[0] + assert op == OP_MUSICBRAINZ_CONFIG + assert group_id == "", "node-wide, like tmdb_config — not tied to self._group_id" + + +async def test_the_contact_itself_never_appears_in_the_signed_subject(tmp_path): + """ + Not a secret the way a TMDB token is, but still kept out of the audited + subject line as free text — same "yes/no configured" shape. + """ + session = _session(tmp_path, "op", operator="op") + session._has_admin_authority = lambda: True + issued = [] + session._issue_admin_challenge = _fake_challenge(issued) + + contact = "operator@example.invalid" + session._do_musicbrainz_config({"contact": contact}) + + _, subject, payload, _ = issued[0] + assert contact not in subject + assert payload["contact"] == contact, "the real value still has to reach the exec step somehow" + + +async def test_subject_reflects_whether_a_contact_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_musicbrainz_config({"contact": "x"}) + + _, subject, _, _ = issued[0] + assert subject == "contact_configured=yes" + + +async def test_subject_says_no_contact_when_none_given(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_musicbrainz_config({}) + + _, subject, _, _ = issued[0] + assert subject == "contact_configured=no" + + +# ── 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.musicbrainz_contact() is None, \ + "absent must mean 'no contact configured' — no shipped default to fall back to" + await roster.set_musicbrainz_contact("operator@example.invalid", set_by="op") + assert await roster.musicbrainz_contact() == "operator@example.invalid" + finally: + await roster.close() + + reopened = Roster(db_path=tmp_path / "roster.db") + await reopened.open() + try: + assert await reopened.musicbrainz_contact() == "operator@example.invalid" + finally: + await reopened.close() + + +async def test_clearing_the_contact_reverts_to_unconfigured(tmp_path): + roster = Roster(db_path=tmp_path / "roster.db") + await roster.open() + try: + await roster.set_musicbrainz_contact("a-contact", set_by="op") + assert await roster.musicbrainz_contact() == "a-contact" + + await roster.set_musicbrainz_contact("", set_by="op") + assert await roster.musicbrainz_contact() is None, \ + "an explicit empty string clears the contact" + finally: + await roster.close() + + +async def test_omitting_the_contact_leaves_it_unchanged(tmp_path): + roster = Roster(db_path=tmp_path / "roster.db") + await roster.open() + try: + await roster.set_musicbrainz_contact("a-contact", set_by="op") + await roster.set_musicbrainz_contact(None, set_by="op") + assert await roster.musicbrainz_contact() == "a-contact" + finally: + await roster.close() diff --git a/packages/meshbay-node/tests/test_musicbrainz_enabled_policy.py b/packages/meshbay-node/tests/test_musicbrainz_enabled_policy.py new file mode 100644 index 0000000..e86a3f3 --- /dev/null +++ b/packages/meshbay-node/tests/test_musicbrainz_enabled_policy.py @@ -0,0 +1,126 @@ +""" +Whether MusicBrainz lookups run *at all* for a group — docs/musicbay.md +§3.2/§6. Per-group from the start (unlike tmdb_enabled, which started +node-wide and moved per-group later once the lesson was already learned). +Same shape as test_tmdb_enabled_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 contact string stays node-wide — see test_musicbrainz_config_policy.py. +""" + +from pathlib import Path + +import pytest + +from meshbay_common.adminop import OP_MUSICBRAINZ_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_musicbrainz_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_musicbrainz_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_musicbrainz_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_musicbrainz_enabled({"enabled": True}) + + assert issued == [(OP_MUSICBRAINZ_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_musicbrainz_enabled({"enabled": False}) + + assert issued == [(OP_MUSICBRAINZ_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.musicbrainz_enabled("g1") is True, "absent must mean on" + await roster.set_musicbrainz_enabled("g1", False, set_by="op") + assert await roster.musicbrainz_enabled("g1") is False + finally: + await roster.close() + + reopened = Roster(db_path=tmp_path / "roster.db") + await reopened.open() + try: + assert await reopened.musicbrainz_enabled("g1") is False + assert await reopened.musicbrainz_enabled("g2") is True, \ + "one group's setting must not answer for another" + finally: + await reopened.close() |