test_audit_hk_songs_duplicates.py
3.03 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
#!/usr/bin/env python3
"""Tests for hk_songs duplicate audit SQL/report helpers."""
from audit_hk_songs_duplicates import (
build_duplicate_detail_sql,
build_duplicate_group_sql,
build_merge_plan,
build_soft_delete_sql,
summarize_duplicate_groups,
)
def test_group_sql_uses_name_lyricist_composer_and_deleted_filter():
sql = build_duplicate_group_sql("hk_songs_test", include_deleted=False, limit=20)
assert "`hk_songs_test`" in sql
assert "COALESCE(NULLIF(TRIM(name), ''), '') COLLATE utf8mb4_bin" in sql
assert "COALESCE(NULLIF(TRIM(lyricist), ''), '') COLLATE utf8mb4_bin" in sql
assert "COALESCE(NULLIF(TRIM(composer), ''), '') COLLATE utf8mb4_bin" in sql
assert "WHERE deleted = '0'" in sql
assert "HAVING COUNT(*) > 1" in sql
assert "LIMIT 20" in sql
def test_group_sql_can_use_database_default_case_insensitive_collation():
sql = build_duplicate_group_sql("hk_songs_test", case_sensitive=False)
assert "COLLATE utf8mb4_bin" not in sql
def test_detail_sql_joins_duplicate_groups_back_to_rows():
sql = build_duplicate_detail_sql("hk_songs_test", include_deleted=True, limit=None)
assert "JOIN (" in sql
assert "dup.name_key" in sql
assert "s.name_key" in sql
assert "WHERE deleted = '0'" not in sql
assert "LIMIT" not in sql
def test_summary_counts_groups_rows_and_extra_duplicates():
groups = [
{"duplicate_count": 2},
{"duplicate_count": 5},
]
summary = summarize_duplicate_groups(groups)
assert summary == {
"duplicate_groups": 2,
"duplicate_rows": 7,
"extra_duplicate_rows": 5,
}
def test_merge_plan_keeps_row_with_largest_record_count():
rows = [
{"id": 10, "name": "A", "lyricist": "L", "composer": "C", "source_song_id": "s10", "record_count": 1},
{"id": 11, "name": "A", "lyricist": "L", "composer": "C", "source_song_id": "s11", "record_count": 3},
{"id": 12, "name": "B", "lyricist": "", "composer": "", "source_song_id": "s12", "record_count": None},
{"id": 13, "name": "B", "lyricist": None, "composer": None, "source_song_id": "s13", "record_count": 2},
]
plan = build_merge_plan(rows)
assert [item["survivor_id"] for item in plan] == [11, 13]
assert [item["loser_id"] for item in plan] == [10, 12]
assert plan[0]["survivor_record_count"] == 3
assert plan[0]["loser_record_count"] == 1
def test_merge_plan_uses_lowest_id_as_tie_breaker():
rows = [
{"id": 22, "name": "A", "lyricist": "L", "composer": "C", "source_song_id": "s22", "record_count": 2},
{"id": 21, "name": "A", "lyricist": "L", "composer": "C", "source_song_id": "s21", "record_count": 2},
]
plan = build_merge_plan(rows)
assert plan[0]["survivor_id"] == 21
assert plan[0]["loser_id"] == 22
def test_soft_delete_sql_updates_only_planned_losers():
sql = build_soft_delete_sql("hk_songs_test", 3)
assert "UPDATE `hk_songs_test`" in sql
assert "SET deleted = '1'" in sql
assert "WHERE deleted = '0'" in sql
assert "id IN (%s,%s,%s)" in sql