-
-
Notifications
You must be signed in to change notification settings - Fork 210
/
test_msid_mbid_mapping.py
144 lines (131 loc) · 7.17 KB
/
test_msid_mbid_mapping.py
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
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
from sqlalchemy import text
from listenbrainz import messybrainz
from listenbrainz.db.msid_mbid_mapping import load_recordings_from_mapping, fetch_track_metadata_for_items, MsidMbidModel
from listenbrainz.db.testing import TimescaleTestCase
from listenbrainz.db import timescale as ts
class MappingTestCase(TimescaleTestCase):
def insert_recording_in_mapping(self, recording):
with ts.engine.connect() as connection:
if recording["recording_mbid"]:
connection.execute(text("""
INSERT INTO mbid_mapping_metadata (artist_credit_id, recording_mbid, release_mbid, release_name,
artist_mbids, artist_credit_name, recording_name)
VALUES (:artist_credit_id, :recording_mbid ::UUID, :release_mbid ::UUID, :release,
:artist_mbids ::UUID[], :artist, :title)
"""), **recording)
match_type = "exact_match"
else:
match_type = "no_match"
connection.execute(
text("""
INSERT INTO mbid_mapping (recording_msid, recording_mbid, match_type)
VALUES (:recording_msid, :recording_mbid, :match_type)
"""),
recording_msid=recording["recording_msid"],
recording_mbid=recording["recording_mbid"],
match_type=match_type
)
def insert_recordings(self):
recordings = [
{
"artist_credit_id": 204,
"recording_mbid": "00000737-3a59-4499-b30a-31fe2464555d",
"release_mbid": "a2589025-8517-45ab-9d64-fe927ba087b1",
"release": "Batman Returns",
"artist_mbids": ["5b24fbab-c58f-4c37-a59d-ab232e2d98c4"],
"artist": "Danny Elfman",
"title": "The Final Confrontation, Part 1"
},
{
"artist_credit_id": 133549,
"recording_mbid": "c5bfd98d-ccde-4cf3-8abb-63fad1b6065a",
"release_mbid": "5da4af04-d796-4d07-801d-a878e83dea48",
"release": "Random Is Resistance",
"artist_mbids": ["797bcf41-0e02-431d-ab99-020e1cb3d0fd"],
"artist": "Rotersand",
"title": "A Number and a Name"
},
{
"artist": "James S.A. Corey",
"title": "The Churn"
},
{
"artist_credit_id": 347,
"recording_mbid": "67bcde07-bfb1-4b30-88ba-6b995ec04123",
"release_mbid": "27280632-fa33-3801-a5b1-081ed0b65bb3",
"release": "Year Zero",
"artist_mbids": ["b7ffd2af-418f-4be2-bdd1-22f8b48613da"],
"artist": "Nine Inch Nails",
"title": "The Warning"
}
]
submitted = messybrainz.insert_all_in_transaction(recordings)
# data sent to msb cannot contain nulls but we want it when inserting in mapping
recordings[2].update(**{
"artist_credit_id": None,
"recording_mbid": None,
"release_mbid": None,
"release": None,
"artist_mbids": None,
})
for recording, submission in zip(recordings, submitted):
recording["recording_msid"] = submission["ids"]["recording_msid"]
self.insert_recording_in_mapping(recording)
# artist_credit_id is not retrieved, remove from dict after submitting
del recording["artist_credit_id"]
return recordings
def test_load_recordings_from_mapping(self):
recordings = self.insert_recordings()
expected_mbid_map = {
recordings[0]["recording_mbid"]: recordings[0],
recordings[1]["recording_mbid"]: recordings[1]
}
expected_msid_map = {
recordings[3]["recording_msid"]: recordings[3]
}
mbid_map, msid_map = load_recordings_from_mapping(
mbids=[recordings[0]["recording_mbid"], recordings[1]["recording_mbid"]],
msids=[recordings[2]["recording_msid"], recordings[3]["recording_msid"]]
)
self.assertEqual(expected_msid_map, msid_map)
self.assertEqual(expected_mbid_map, mbid_map)
def test_fetch_track_metadata_for_items(self):
recordings = self.insert_recordings()
models = [
# these recordings test we find mapping metadata from mbids
MsidMbidModel(recording_msid=recordings[0]["recording_msid"], recording_mbid=recordings[0]["recording_mbid"]),
MsidMbidModel(recording_msid=recordings[1]["recording_msid"], recording_mbid=recordings[1]["recording_mbid"]),
# this recording tests loading data from MsB when no mapping is available
MsidMbidModel(recording_msid=recordings[2]["recording_msid"], recording_mbid=None),
# this recording tests loading mapping metadata from msid but no mbid
# (we actually have a mapped mbid in the corresponding recording but omit it here for testing this case)
MsidMbidModel(recording_msid=recordings[3]["recording_msid"], recording_mbid=None)
]
models = fetch_track_metadata_for_items(models)
for model, recording in zip(models, recordings):
metadata = model.track_metadata
self.assertEqual(metadata["track_name"], recording["title"])
self.assertEqual(metadata["artist_name"], recording["artist"])
if recording["recording_mbid"] is None: # 3rd recording which only present in MsB
continue
self.assertEqual(metadata["release_name"], recording["release"])
self.assertEqual(metadata["additional_info"]["recording_mbid"], recording["recording_mbid"])
self.assertEqual(metadata["additional_info"]["recording_msid"], recording["recording_msid"])
self.assertEqual(metadata["additional_info"]["release_mbid"], recording["release_mbid"])
self.assertEqual(metadata["additional_info"]["artist_mbids"], recording["artist_mbids"])
def test_fetch_track_metadata_for_items_with_same_mbid(self):
recording = self.insert_recordings()[0]
models = [
MsidMbidModel(recording_msid=recording["recording_msid"], recording_mbid=recording["recording_mbid"]),
MsidMbidModel(recording_msid=recording["recording_msid"], recording_mbid=recording["recording_mbid"]),
]
models = fetch_track_metadata_for_items(models)
for model in models:
metadata = model.track_metadata
self.assertEqual(metadata["track_name"], recording["title"])
self.assertEqual(metadata["artist_name"], recording["artist"])
self.assertEqual(metadata["release_name"], recording["release"])
self.assertEqual(metadata["additional_info"]["recording_mbid"], recording["recording_mbid"])
self.assertEqual(metadata["additional_info"]["recording_msid"], recording["recording_msid"])
self.assertEqual(metadata["additional_info"]["release_mbid"], recording["release_mbid"])
self.assertEqual(metadata["additional_info"]["artist_mbids"], recording["artist_mbids"])