diff --git a/fluxtuner/core/local_station_search.py b/fluxtuner/core/local_station_search.py new file mode 100644 index 0000000..47b170d --- /dev/null +++ b/fluxtuner/core/local_station_search.py @@ -0,0 +1,164 @@ +"""Read-only, profile-scoped local station discovery.""" + +from __future__ import annotations + +import sqlite3 +from typing import Any + +from fluxtuner.core.favorites import favorite_display_name, favorite_from_row +from fluxtuner.core.stations import station_bitrate, station_from_row, station_tags + +# Only shared station metadata belongs in a discovery result by default. +# metadata_json may contain legacy or source-specific private fields. +_PUBLIC_STATION_FIELDS = ( + "stationuuid", + "name", + "url", + "url_resolved", + "homepage", + "favicon", + "country", + "countrycode", + "language", + "tags", + "codec", + "bitrate", +) + + +def _local_station_from_row( + row: sqlite3.Row, +) -> dict[str, Any]: + """Expose shared station fields and only this profile's favorite data.""" + stored = station_from_row(row) + station = {key: stored[key] for key in _PUBLIC_STATION_FIELDS} + + # Keep only the MIME hints recognized by the playback compatibility + # contract. Do not expose other arbitrary metadata_json fields. + for key in ("content_type", "content-type", "mime_type", "mimetype"): + value = stored.get(key) + if isinstance(value, str) and value.strip(): + station[key] = value + + if row["is_favorite"]: + favorite = favorite_from_row(row) + station["custom_name"] = favorite["custom_name"] + station["favorite_tags"] = favorite["favorite_tags"] + + return station + + +def _rank(station: dict[str, Any], query: str) -> int: + """Return a deterministic relevance tier; lower is better.""" + if not query: + return 0 + + needle = query.casefold() + display_name = favorite_display_name(station).casefold() + + if display_name == needle: + return 0 + if display_name.startswith(needle): + return 1 + if needle in display_name: + return 2 + + tags = station_tags(station) + tags.extend(station.get("favorite_tags") or []) + + if any(needle in str(tag).casefold() for tag in tags): + return 3 + + return 4 + + +def _matches_country(station: dict[str, Any], country: str | None) -> bool: + if not country: + return True + + needle = country.casefold() + country_name = str(station.get("country") or "").casefold() + country_code = str(station.get("countrycode") or "").casefold() + + return needle in country_name or needle == country_code + + +def search_local_stations( + conn: sqlite3.Connection, + *, + profile_id: int, + query: str = "", + country: str | None = None, + min_bitrate: int | None = None, + limit: int = 50, +) -> list[dict[str, Any]]: + """Search stations saved in favorites or playlists of one profile. + + Uses the caller's existing SQLite connection without writing, + initializing storage or changing transaction boundaries. + """ + if type(profile_id) is not int or profile_id <= 0: + raise ValueError("An existing profile is required.") + + profile = conn.execute( + "SELECT 1 FROM profiles WHERE id = ?", + (profile_id,), + ).fetchone() + + if profile is None: + raise ValueError("The selected profile no longer exists.") + + clean_query = query.strip().casefold() + clean_country = country.strip().casefold() if country else None + normalized_bitrate = max(0, int(min_bitrate)) if min_bitrate is not None else None + normalized_limit = max(1, int(limit)) + + rows = conn.execute( + """ + SELECT + s.*, + f.custom_name, + f.favorite_tags_json, + CASE WHEN f.station_id IS NULL THEN 0 ELSE 1 END AS is_favorite + FROM stations AS s + LEFT JOIN favorites AS f + ON f.station_id = s.id + AND f.profile_id = ? + WHERE f.station_id IS NOT NULL + OR EXISTS ( + SELECT 1 + FROM playlist_stations AS ps + JOIN playlists AS p ON p.id = ps.playlist_id + WHERE ps.station_id = s.id + AND p.profile_id = ? + ) + ORDER BY s.station_key COLLATE BINARY ASC + """, + (profile_id, profile_id), + ).fetchall() + + matches: list[dict[str, Any]] = [] + + for row in rows: + station = _local_station_from_row(row) + + if clean_query and _rank(station, clean_query) == 4: + continue + + if not _matches_country(station, clean_country): + continue + + if normalized_bitrate is not None and station_bitrate(station) < normalized_bitrate: + continue + + matches.append(station) + + matches.sort( + key=lambda station: ( + _rank(station, clean_query), + favorite_display_name(station).casefold(), + str(station.get("url_resolved") or station.get("url") or ""), + ) + ) + + return matches[:normalized_limit] diff --git a/tests/test_local_station_search.py b/tests/test_local_station_search.py new file mode 100644 index 0000000..49d54ca --- /dev/null +++ b/tests/test_local_station_search.py @@ -0,0 +1,477 @@ +"""Contract tests for profile-scoped local station discovery.""" + +import pytest + +from fluxtuner.core import db, profiles +from fluxtuner.core.custom_stations import add_custom_station +from fluxtuner.core.local_station_search import search_local_stations +from fluxtuner.core.playlists import add_station_to_playlist_record +from fluxtuner.core.stations import station_key + + +@pytest.fixture +def library(tmp_path): + path = tmp_path / "library.sqlite3" + db.init_db(path) + with db.connect(path) as conn: + first = profiles.get_or_create_profile(conn, "first") + second = profiles.get_or_create_profile(conn, "second") + conn.commit() + yield conn, first, second + + +def test_union_of_favorites_and_playlist_stations(library): + conn, first, second = library + + add_custom_station( + conn, + profile_id=first, + name="Personal Radio", + url="https://example.org/personal", + ) + + playlist_station = { + "name": "Playlist Only", + "url": "https://example.org/playlist", + "tags": "ambient", + } + assert add_station_to_playlist_record(conn, "Collection", playlist_station, profile_id=first) + + # Adding the same stream to a playlist must not duplicate it. + assert add_station_to_playlist_record( + conn, + "Collection", + { + "name": "Personal Radio", + "url": "https://example.org/personal", + }, + profile_id=first, + ) + + results = search_local_stations(conn, profile_id=first) + + assert {station_key(item) for item in results} == { + "https://example.org/personal", + "https://example.org/playlist", + } + assert len(results) == 2 + + +def test_search_never_exposes_another_profile(library): + conn, first, second = library + + add_custom_station( + conn, + profile_id=first, + name="Private First", + url="https://example.org/first", + ) + add_custom_station( + conn, + profile_id=second, + name="Private Second", + url="https://example.org/second", + ) + + first_results = search_local_stations(conn, profile_id=first) + second_results = search_local_stations(conn, profile_id=second) + + assert [station_key(item) for item in first_results] == ["https://example.org/first"] + assert [station_key(item) for item in second_results] == ["https://example.org/second"] + + assert search_local_stations(conn, profile_id=first, query="Private Second") == [] + + +def test_custom_names_are_searchable_without_changing_shared_identity(library): + conn, first, second = library + + url = "https://example.org/shared" + + add_custom_station(conn, profile_id=first, name="My Jazz Station", url=url) + add_custom_station(conn, profile_id=second, name="Work Radio", url=url) + + own = search_local_stations(conn, profile_id=first, query="My Jazz Station") + other = search_local_stations(conn, profile_id=second, query="My Jazz Station") + + assert len(own) == 1 + assert station_key(own[0]) == url + assert own[0]["custom_name"] == "My Jazz Station" + assert other == [] + + +@pytest.mark.parametrize("invalid_profile", [None, 0, -1, True, 999999]) +def test_invalid_profile_never_falls_back_to_default(library, invalid_profile): + conn, first, second = library + + with pytest.raises(ValueError): + search_local_stations(conn, profile_id=invalid_profile) + + +def test_discovery_does_not_write_to_database(library): + conn, first, second = library + + add_custom_station( + conn, + profile_id=first, + name="Read Only", + url="https://example.org/readonly", + ) + conn.commit() + + before = conn.total_changes + results = search_local_stations(conn, profile_id=first) + + assert len(results) == 1 + assert conn.total_changes == before + + +# N9_SECURITY_CONTRACTS + + +def test_favorite_tags_are_scoped_to_the_owner(library): + from fluxtuner.core.favorites import update_favorite_record + + conn, first, second = library + url = "https://example.org/tagged" + + add_custom_station(conn, profile_id=first, name="Jazz", url=url) + add_custom_station(conn, profile_id=second, name="Jazz", url=url) + + assert update_favorite_record(conn, url, favorite_tags=["private-collection"], profile_id=first) + + assert len(search_local_stations(conn, profile_id=first, query="private-collection")) == 1 + + assert search_local_stations(conn, profile_id=second, query="private-collection") == [] + + +def test_country_and_bitrate_filters_preserve_unknown_metadata(library): + conn, first, _second = library + + add_station_to_playlist_record( + conn, + "Stations", + { + "name": "Spanish Jazz", + "url": "https://example.org/es", + "country": "Spain", + "countrycode": "ES", + "bitrate": 192, + }, + profile_id=first, + ) + + add_station_to_playlist_record( + conn, + "Stations", + { + "name": "Unknown Metadata", + "url": "https://example.org/unknown", + }, + profile_id=first, + ) + + results = search_local_stations(conn, profile_id=first, country="ES", min_bitrate=128) + + assert [station_key(item) for item in results] == ["https://example.org/es"] + + assert len(search_local_stations(conn, profile_id=first)) == 2 + + +def test_ranking_and_limits_are_deterministic(library): + conn, first, _second = library + + for name, suffix in [ + ("Jazz Collection", "collection"), + ("Jazz", "exact"), + ("Evening Jazz", "evening"), + ]: + add_custom_station( + conn, + profile_id=first, + name=name, + url=f"https://example.org/{suffix}", + ) + + first_run = search_local_stations(conn, profile_id=first, query="Jazz", limit=2) + second_run = search_local_stations(conn, profile_id=first, query="Jazz", limit=2) + + assert [station_key(item) for item in first_run] == [ + "https://example.org/exact", + "https://example.org/collection", + ] + assert first_run == second_run + + +def test_untrusted_station_metadata_cannot_supply_private_favorite_fields(library): + """Station metadata must not impersonate per-profile favorite metadata.""" + from fluxtuner.core.stations import upsert_station + + conn, first, second = library + url = "https://example.org/metadata" + + upsert_station( + conn, + { + "name": "Public Station", + "url": url, + "custom_name": "Secret Alias", + "favorite_tags": ["secret-tag"], + }, + ) + + add_station_to_playlist_record( + conn, + "Second Profile Playlist", + {"name": "Public Station", "url": url}, + profile_id=second, + ) + + results = search_local_stations(conn, profile_id=second) + assert len(results) == 1 + + assert not results[0].get("custom_name") + assert not results[0].get("favorite_tags") + assert search_local_stations(conn, profile_id=second, query="Secret Alias") == [] + assert search_local_stations(conn, profile_id=second, query="secret-tag") == [] + + +def test_search_is_scoped_to_profile_even_across_users(tmp_path): + path = tmp_path / "users.sqlite3" + db.init_db(path) + + with db.connect(path) as conn: + first_user = db.get_or_create_user(conn, "discovery-user-a") + second_user = db.get_or_create_user(conn, "discovery-user-b") + + first_profile = profiles.get_or_create_profile(conn, "library", user_id=first_user) + second_profile = profiles.get_or_create_profile(conn, "library", user_id=second_user) + + add_custom_station( + conn, + profile_id=first_profile, + name="User A Station", + url="https://example.org/user-a", + ) + add_custom_station( + conn, + profile_id=second_profile, + name="User B Station", + url="https://example.org/user-b", + ) + + assert [ + station_key(item) for item in search_local_stations(conn, profile_id=first_profile) + ] == ["https://example.org/user-a"] + + assert [ + station_key(item) for item in search_local_stations(conn, profile_id=second_profile) + ] == ["https://example.org/user-b"] + + +# N10_LEGACY_METADATA_CONTRACT + + +@pytest.mark.parametrize("membership", ["favorite", "playlist"]) +def test_legacy_metadata_cannot_leak_private_fields(library, membership): + """Search must not expose per-profile fields from shared metadata_json.""" + import json + + conn, first, second = library + url = f"https://example.org/legacy-{membership}" + + if membership == "favorite": + add_custom_station( + conn, + profile_id=second, + name="Public Alias", + url=url, + ) + else: + add_station_to_playlist_record( + conn, + "Legacy Collection", + {"name": "Public Station", "url": url}, + profile_id=second, + ) + + # Deliberately inject legacy metadata AFTER persisting membership. + # No upsert follows, so the suspicious data cannot be overwritten. + conn.execute( + """ + UPDATE stations + SET metadata_json = ? + WHERE station_key = ? + """, + ( + json.dumps( + { + "custom_name": "Other Profile Secret", + "favorite_tags": ["confidential-tag"], + "source_profile_id": first, + } + ), + url, + ), + ) + + results = search_local_stations(conn, profile_id=second) + + assert len(results) == 1 + assert station_key(results[0]) == url + assert results[0].get("custom_name") in ( + None, + "Public Alias" if membership == "favorite" else None, + ) + assert not results[0].get("favorite_tags") + assert "source_profile_id" not in results[0] + + assert search_local_stations(conn, profile_id=second, query="Other Profile Secret") == [] + + assert search_local_stations(conn, profile_id=second, query="confidential-tag") == [] + + +# N14_CODEC_CONTRACT + + +@pytest.mark.parametrize( + ("mime_key", "mime_value"), + [ + ("content_type", "audio/mpeg"), + ("content-type", "audio/mpeg"), + ("mime_type", "audio/mpeg"), + ("mimetype", "audio/mpeg"), + ], +) +def test_local_discovery_preserves_public_mime_hints(library, mime_key, mime_value): + """Known MIME hints should remain useful to specialized players.""" + import json + + from fluxtuner.core.compatibility import station_is_supported + from fluxtuner.players.capabilities import PlayerCapabilities + + conn, first, _second = library + url = f"https://example.org/unknown-stream-{mime_key}" + + add_custom_station( + conn, + profile_id=first, + name="Codec Discovery", + url=url, + ) + + # Insert metadata after membership, avoiding a later upsert. + conn.execute( + """ + UPDATE stations + SET metadata_json = ? + WHERE station_key = ? + """, + ( + json.dumps( + { + mime_key: mime_value, + "private_note": "Never expose this", + } + ), + url, + ), + ) + + result = search_local_stations(conn, profile_id=first) + + assert len(result) == 1 + assert result[0].get(mime_key) == mime_value + assert "private_note" not in result[0] + + mp3_player = PlayerCapabilities(supported_codecs=frozenset({"mp3"})) + assert station_is_supported(result[0], mp3_player) + + +def test_local_discovery_unknown_codec_remains_unknown(library): + from fluxtuner.core.compatibility import station_is_supported + from fluxtuner.players.capabilities import PlayerCapabilities + + conn, first, _second = library + add_custom_station( + conn, + profile_id=first, + name="Unknown Format", + url="https://example.org/live", + ) + + result = search_local_stations(conn, profile_id=first) + assert len(result) == 1 + + specialized = PlayerCapabilities(supported_codecs=frozenset({"mp3"})) + general = PlayerCapabilities(general_purpose=True) + + assert not station_is_supported(result[0], specialized) + assert station_is_supported(result[0], general) + + +# N16_FINAL_CONTRACTS + + +def test_discovery_works_with_read_only_sqlite_connection(tmp_path): + import sqlite3 + + path = tmp_path / "readonly-library.sqlite3" + db.init_db(path) + + with db.connect(path) as conn: + profile_id = profiles.get_or_create_profile(conn, "readonly") + add_custom_station( + conn, + profile_id=profile_id, + name="Read Only Radio", + url="https://example.org/readonly-radio", + ) + conn.commit() + + readonly_uri = f"{path.as_uri()}?mode=ro" + with sqlite3.connect(readonly_uri, uri=True) as conn: + conn.row_factory = sqlite3.Row + + results = search_local_stations(conn, profile_id=profile_id) + + assert [station_key(item) for item in results] == ["https://example.org/readonly-radio"] + assert not conn.in_transaction + + +def test_empty_query_is_deterministic_and_respects_limit(library): + conn, first, _second = library + + for name, suffix in ( + ("Zulu Radio", "zulu"), + ("Alpha Radio", "alpha"), + ("Bravo Radio", "bravo"), + ): + add_custom_station( + conn, + profile_id=first, + name=name, + url=f"https://example.org/{suffix}", + ) + + results = search_local_stations(conn, profile_id=first, query=" ", limit=2) + + assert [station_key(item) for item in results] == [ + "https://example.org/alpha", + "https://example.org/bravo", + ] + assert results == search_local_stations(conn, profile_id=first, query="", limit=2) + + +def test_unknown_bitrate_does_not_satisfy_positive_minimum(library): + conn, first, _second = library + + add_custom_station( + conn, + profile_id=first, + name="Unknown Bitrate", + url="https://example.org/unknown-bitrate", + ) + + assert len(search_local_stations(conn, profile_id=first)) == 1 + + assert search_local_stations(conn, profile_id=first, min_bitrate=128) == []