"""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