"""Tests for queries module.""" import pytest from parse_email import Track, Album from queries import ( esc, esc_like, esc_prompt, validate_params, build_track_query, build_album_query, build_ai_track_query, build_ai_album_query, build_wide_track_query, build_wide_album_query, build_grps_track_query, build_grps_album_query, build_grps_wide_track_query, build_grps_wide_album_query, _dim_artist_condition, _grps_artist_condition, _strip_featuring, ) class TestEscaping: def test_esc_single_quotes(self): assert esc("O'Brien") == "O''Brien" def test_esc_like_wildcards(self): assert "\\%" in esc_like("100%") assert "\\_" in esc_like("under_score") def test_esc_prompt_strips_dangerous_chars(self): assert "<" not in esc_prompt('') assert ">" not in esc_prompt('') assert '"' not in esc_prompt('say "hello"') def test_esc_prompt_strips_newlines(self): assert "\n" not in esc_prompt("line1\nline2") assert "\r" not in esc_prompt("line1\r\nline2") def test_esc_prompt_strips_sql_comments(self): assert "--" not in esc_prompt("great song -- ignore above") assert "/*" not in esc_prompt("title /* comment */") assert "*/" not in esc_prompt("title /* comment */") class TestValidateParams: def test_valid_params(self): validate_params("2026-04-03", 7) # should not raise def test_invalid_date_format(self): with pytest.raises(ValueError, match="Invalid release date"): validate_params("04-03-2026", 7) def test_invalid_date_window_too_low(self): with pytest.raises(ValueError, match="Invalid date window"): validate_params("2026-04-03", 0) def test_invalid_date_window_too_high(self): with pytest.raises(ValueError, match="Invalid date window"): validate_params("2026-04-03", 31) def test_invalid_calendar_date(self): with pytest.raises(ValueError, match="Invalid calendar date"): validate_params("2026-02-31", 7) def test_invalid_calendar_date_apr_31(self): with pytest.raises(ValueError, match="Invalid calendar date"): validate_params("2026-04-31", 7) class TestDimArtistCondition: def test_exact_match(self): cond = _dim_artist_condition("a.ARTISTNAME", "BUNT.") assert "UPPER(a.ARTISTNAME) = 'BUNT.'" in cond def test_collaborator_patterns(self): cond = _dim_artist_condition("a.ARTISTNAME", "BUNT.") assert "LIKE 'BUNT. &%'" in cond assert "LIKE 'BUNT. X %'" in cond def test_reverse_collaborator(self): cond = _dim_artist_condition("a.ARTISTNAME", "BUNT. & Malou") # Should include reverse match: email artist starts with DB artist assert "LIKE UPPER(a.ARTISTNAME) || ' &%'" in cond class TestStripFeaturing: def test_feat_dot(self): assert _strip_featuring("Osé feat. Victony") == "Osé" def test_ft_dot(self): assert _strip_featuring("Drake ft. Future") == "Drake" def test_featuring(self): assert _strip_featuring("Rihanna featuring Jay-Z") == "Rihanna" def test_feat_no_dot(self): assert _strip_featuring("Osé feat Victony") == "Osé" def test_no_featuring(self): assert _strip_featuring("Central Cee") is None def test_case_insensitive(self): assert _strip_featuring("BIA Feat. Lil Wayne") == "BIA" class TestFeatArtistCondition: def test_dim_includes_primary_artist(self): cond = _dim_artist_condition("a.ARTISTNAME", "Osé feat. Victony") assert "UPPER(a.ARTISTNAME) = 'OSÉ FEAT. VICTONY'" in cond assert "UPPER(a.ARTISTNAME) = 'OSÉ'" in cond def test_dim_no_featuring_no_extra_conditions(self): cond = _dim_artist_condition("a.ARTISTNAME", "Central Cee") assert "CENTRAL CEE" in cond # Should not have duplicate exact match assert cond.count("UPPER(a.ARTISTNAME) = 'CENTRAL CEE'") == 1 def test_grps_includes_primary_artist(self): cond = _grps_artist_condition("p.PARTICIP_FULL_NAME", "Drake ft. Future") assert "UPPER(p.PARTICIP_FULL_NAME) = 'DRAKE FT. FUTURE'" in cond assert "UPPER(p.PARTICIP_FULL_NAME) = 'DRAKE'" in cond def test_dim_track_query_featuring_exact(self): tracks = [Track(artist="Osé feat. Victony", title="Bother Me")] sql = build_track_query(tracks, "2026-04-10", 7, "exact") assert "'OSÉ'" in sql assert "BOTHER ME" in sql def test_dim_track_query_featuring_fuzzy(self): tracks = [Track(artist="Osé feat. Victony", title="Bother Me")] sql = build_track_query(tracks, "2026-04-10", 7, "fuzzy") assert "JAROWINKLER_SIMILARITY(UPPER(a.ARTISTNAME), 'OSÉ')" in sql def test_grps_track_query_featuring_exact(self): tracks = [Track(artist="Osé feat. Victony", title="Bother Me")] sql = build_grps_track_query(tracks, "2026-04-10", 7, "exact") assert "'OSÉ'" in sql class TestTierValidation: def test_invalid_tier_track(self): tracks = [Track(artist="BIA", title="LIGHTS OUT")] with pytest.raises(ValueError, match="Invalid tier"): build_track_query(tracks, "2026-04-03", 7, "invalid") def test_invalid_tier_album(self): albums = [Album(artist="Central Cee", title="Wild West")] with pytest.raises(ValueError, match="Invalid tier"): build_album_query(albums, "2026-04-03", 7, "invalid") def test_invalid_tier_grps_track(self): tracks = [Track(artist="BIA", title="LIGHTS OUT")] with pytest.raises(ValueError, match="Invalid tier"): build_grps_track_query(tracks, "2026-04-03", 7, "bad") def test_invalid_tier_grps_album(self): albums = [Album(artist="Central Cee", title="Wild West")] with pytest.raises(ValueError, match="Invalid tier"): build_grps_album_query(albums, "2026-04-03", 7, "bad") class TestBuildTrackQuery: def test_returns_empty_for_no_tracks(self): assert build_track_query([], "2026-04-03", 7, "exact") == "" def test_exact_tier_contains_artist_condition(self): tracks = [Track(artist="BIA", title="LIGHTS OUT")] sql = build_track_query(tracks, "2026-04-03", 7, "exact") assert "UPPER(a.ARTISTNAME)" in sql assert "TRACKNAME" in sql assert "LIGHTS OUT" in sql def test_partial_tier_uses_like(self): tracks = [Track(artist="BIA", title="LIGHTS")] sql = build_track_query(tracks, "2026-04-03", 7, "partial") assert "LIKE" in sql def test_fuzzy_tier_uses_jarowinkler(self): tracks = [Track(artist="BIA", title="LIGHTS OUT")] sql = build_track_query(tracks, "2026-04-03", 7, "fuzzy") assert "JAROWINKLER_SIMILARITY" in sql def test_date_window_in_query(self): tracks = [Track(artist="BIA", title="LIGHTS OUT")] sql = build_track_query(tracks, "2026-04-03", 14, "exact") assert "2026-04-03" in sql assert "14" in sql def test_sql_injection_escaped(self): tracks = [Track(artist="O'Brien", title="It's a Test")] sql = build_track_query(tracks, "2026-04-03", 7, "exact") assert "O''BRIEN" in sql assert "IT''S A TEST" in sql def test_multiple_tracks_creates_or_conditions(self): tracks = [ Track(artist="BIA", title="LIGHTS OUT"), Track(artist="Central Cee", title="WAGWAN"), ] sql = build_track_query(tracks, "2026-04-03", 7, "exact") assert "OR" in sql def test_dim_bidirectional_artist_matching(self): tracks = [Track(artist="BUNT. & Malou", title="i need u")] sql = build_track_query(tracks, "2026-04-03", 7, "exact") # Should match DB artist "BUNT." when email has "BUNT. & Malou" assert "LIKE UPPER(a.ARTISTNAME)" in sql class TestBuildAlbumQuery: def test_returns_empty_for_no_albums(self): assert build_album_query([], "2026-04-03", 7, "exact") == "" def test_exact_tier(self): albums = [Album(artist="Central Cee", title="Wild West")] sql = build_album_query(albums, "2026-04-03", 7, "exact") assert "RELEASENAME" in sql assert "WILD WEST" in sql class TestBuildWideQueries: def test_wide_track_no_date_filter(self): tracks = [Track(artist="BIA", title="LIGHTS OUT")] sql = build_wide_track_query(tracks) assert "BETWEEN" not in sql assert "LIMIT 100" in sql def test_wide_album_no_date_filter(self): albums = [Album(artist="Central Cee", title="Wild West")] sql = build_wide_album_query(albums) assert "BETWEEN" not in sql def test_wide_empty_returns_empty(self): assert build_wide_track_query([]) == "" assert build_wide_album_query([]) == "" class TestBuildAiQueries: def test_ai_track_uses_cortex(self): track = Track(artist="BIA", title="LIGHTS OUT") sql = build_ai_track_query(track, "2026-04-03", 7) assert "SNOWFLAKE.CORTEX.COMPLETE" in sql assert "llama3.1-8b" in sql assert "AI_MATCH" in sql def test_ai_album_uses_cortex(self): album = Album(artist="Central Cee", title="Wild West") sql = build_ai_album_query(album, "2026-04-03", 7) assert "SNOWFLAKE.CORTEX.COMPLETE" in sql class TestGrpsQueries: def test_grps_track_uses_correct_tables(self): tracks = [Track(artist="BIA", title="LIGHTS OUT")] sql = build_grps_track_query(tracks, "2026-04-03", 7, "exact") assert "TRAS007" in sql assert "TRAS398" in sql assert "PARTICIP_FULL_NAME" in sql def test_grps_album_uses_correct_tables(self): albums = [Album(artist="Central Cee", title="Wild West")] sql = build_grps_album_query(albums, "2026-04-03", 7, "exact") assert "TRAS002" in sql assert "TRAS001" in sql assert "PROD_TITLE" in sql def test_grps_artist_condition_handles_collaborators(self): cond = _grps_artist_condition("p.PARTICIP_FULL_NAME", "BIA") assert "UPPER(p.PARTICIP_FULL_NAME) = 'BIA'" in cond assert "'BIA &%'" in cond assert "'BIA X %'" in cond def test_grps_wide_track(self): tracks = [Track(artist="BIA", title="LIGHTS OUT")] sql = build_grps_wide_track_query(tracks) assert "BETWEEN" not in sql assert "TRAS007" in sql def test_grps_wide_album(self): albums = [Album(artist="Central Cee", title="Wild West")] sql = build_grps_wide_album_query(albums) assert "BETWEEN" not in sql assert "TRAS002" in sql