zzstoatzz.io
audio streaming app
plyr.fm
1"""tests for artist analytics endpoints."""
2
3from collections.abc import Generator
4
5import pytest
6from fastapi import FastAPI
7from httpx import ASGITransport, AsyncClient
8from sqlalchemy.ext.asyncio import AsyncSession
9
10from backend._internal import Session, require_auth
11from backend.main import app
12from backend.models import Artist, Track, TrackLike
13
14
15class MockSession(Session):
16 """mock session for auth bypass in tests."""
17
18 def __init__(self, did: str = "did:test:user123"):
19 self.did = did
20 self.handle = "testuser.bsky.social"
21 self.session_id = "test_session_id"
22 self.access_token = "test_token"
23 self.refresh_token = "test_refresh"
24 self.oauth_session = {
25 "did": did,
26 "handle": "testuser.bsky.social",
27 "pds_url": "https://test.pds",
28 "authserver_iss": "https://auth.test",
29 "scope": "atproto transition:generic",
30 "access_token": "test_token",
31 "refresh_token": "test_refresh",
32 "dpop_private_key_pem": "fake_key",
33 "dpop_authserver_nonce": "",
34 "dpop_pds_nonce": "",
35 }
36
37
38@pytest.fixture
39def test_app(db_session: AsyncSession) -> Generator[FastAPI, None, None]:
40 """create test app with mocked auth."""
41
42 async def mock_require_auth() -> Session:
43 return MockSession()
44
45 app.dependency_overrides[require_auth] = mock_require_auth
46
47 yield app
48
49 app.dependency_overrides.clear()
50
51
52@pytest.fixture
53async def artist_with_tracks(db_session: AsyncSession) -> Artist:
54 """create artist with tracks having different play counts and likes."""
55 artist = Artist(
56 did="did:plc:artist123",
57 handle="artist.bsky.social",
58 display_name="Test Artist",
59 )
60 db_session.add(artist)
61 await db_session.flush()
62
63 # create tracks with varying play counts
64 tracks = [
65 Track(
66 title="Most Played",
67 artist_did=artist.did,
68 file_id="file_1",
69 file_type="mp3",
70 play_count=100,
71 atproto_record_uri="at://did:plc:artist123/fm.plyr.track/1",
72 atproto_record_cid="cid_1",
73 ),
74 Track(
75 title="Most Liked",
76 artist_did=artist.did,
77 file_id="file_2",
78 file_type="mp3",
79 play_count=50,
80 atproto_record_uri="at://did:plc:artist123/fm.plyr.track/2",
81 atproto_record_cid="cid_2",
82 ),
83 Track(
84 title="Least Popular",
85 artist_did=artist.did,
86 file_id="file_3",
87 file_type="mp3",
88 play_count=10,
89 atproto_record_uri="at://did:plc:artist123/fm.plyr.track/3",
90 atproto_record_cid="cid_3",
91 ),
92 ]
93
94 for track in tracks:
95 db_session.add(track)
96
97 await db_session.commit()
98
99 # refresh to get IDs
100 for track in tracks:
101 await db_session.refresh(track)
102
103 # add likes: "Most Liked" gets 5 likes, "Most Played" gets 2 likes
104 likes = []
105 for i in range(5):
106 likes.append(
107 TrackLike(
108 track_id=tracks[1].id, # Most Liked
109 user_did=f"did:test:user{i}",
110 atproto_like_uri=f"at://did:test:user{i}/fm.plyr.like/1",
111 )
112 )
113
114 for i in range(2):
115 likes.append(
116 TrackLike(
117 track_id=tracks[0].id, # Most Played
118 user_did=f"did:test:user{i + 10}",
119 atproto_like_uri=f"at://did:test:user{i + 10}/fm.plyr.like/1",
120 )
121 )
122
123 for like in likes:
124 db_session.add(like)
125
126 await db_session.commit()
127
128 return artist
129
130
131async def test_get_artist_analytics_with_likes(
132 test_app: FastAPI, db_session: AsyncSession, artist_with_tracks: Artist
133):
134 """test analytics returns both top played and top liked tracks."""
135 async with AsyncClient(
136 transport=ASGITransport(app=test_app), base_url="http://test"
137 ) as client:
138 response = await client.get(f"/artists/{artist_with_tracks.did}/analytics")
139
140 assert response.status_code == 200
141 data = response.json()
142
143 # verify total metrics
144 assert data["total_plays"] == 160 # 100 + 50 + 10
145 assert data["total_items"] == 3
146 assert data["total_duration_seconds"] == 0 # no duration set on tracks
147
148 # verify top played track
149 assert data["top_item"]["title"] == "Most Played"
150 assert data["top_item"]["play_count"] == 100
151
152 # verify top liked track
153 assert data["top_liked"]["title"] == "Most Liked"
154 assert data["top_liked"]["play_count"] == 5 # like count
155
156
157async def test_get_artist_analytics_no_tracks(
158 test_app: FastAPI, db_session: AsyncSession
159):
160 """test analytics returns zeros for artist with no tracks."""
161 # create artist with no tracks
162 artist = Artist(
163 did="did:plc:newartist",
164 handle="newartist.bsky.social",
165 display_name="New Artist",
166 )
167 db_session.add(artist)
168 await db_session.commit()
169
170 async with AsyncClient(
171 transport=ASGITransport(app=test_app), base_url="http://test"
172 ) as client:
173 response = await client.get(f"/artists/{artist.did}/analytics")
174
175 assert response.status_code == 200
176 data = response.json()
177
178 assert data["total_plays"] == 0
179 assert data["total_items"] == 0
180 assert data["total_duration_seconds"] == 0
181 assert data["top_item"] is None
182 assert data["top_liked"] is None
183
184
185async def test_get_artist_analytics_no_likes(
186 test_app: FastAPI, db_session: AsyncSession
187):
188 """test analytics when artist has tracks but no likes."""
189 artist = Artist(
190 did="did:plc:artist456",
191 handle="artist456.bsky.social",
192 display_name="Test Artist 2",
193 )
194 db_session.add(artist)
195 await db_session.flush()
196
197 track = Track(
198 title="Unloved Track",
199 artist_did=artist.did,
200 file_id="file_1",
201 file_type="mp3",
202 play_count=50,
203 atproto_record_uri="at://did:plc:artist456/fm.plyr.track/1",
204 atproto_record_cid="cid_1",
205 )
206 db_session.add(track)
207 await db_session.commit()
208
209 async with AsyncClient(
210 transport=ASGITransport(app=test_app), base_url="http://test"
211 ) as client:
212 response = await client.get(f"/artists/{artist.did}/analytics")
213
214 assert response.status_code == 200
215 data = response.json()
216
217 assert data["total_plays"] == 50
218 assert data["total_items"] == 1
219 assert data["total_duration_seconds"] == 0 # no duration set
220 assert data["top_item"]["title"] == "Unloved Track"
221 assert data["top_liked"] is None # no likes
222
223
224async def test_get_artist_analytics_with_duration(
225 test_app: FastAPI, db_session: AsyncSession
226):
227 """test analytics returns total duration from tracks."""
228 artist = Artist(
229 did="did:plc:artist_duration",
230 handle="duration-artist.bsky.social",
231 display_name="Duration Artist",
232 )
233 db_session.add(artist)
234 await db_session.flush()
235
236 # create tracks with duration in extra field
237 tracks = [
238 Track(
239 title="Short Track",
240 artist_did=artist.did,
241 file_id="short_1",
242 file_type="mp3",
243 play_count=10,
244 extra={"duration": 180}, # 3 minutes
245 ),
246 Track(
247 title="Long Track",
248 artist_did=artist.did,
249 file_id="long_1",
250 file_type="mp3",
251 play_count=5,
252 extra={"duration": 3600}, # 1 hour
253 ),
254 Track(
255 title="No Duration Track",
256 artist_did=artist.did,
257 file_id="no_dur_1",
258 file_type="mp3",
259 play_count=3,
260 extra={}, # no duration
261 ),
262 ]
263 for track in tracks:
264 db_session.add(track)
265 await db_session.commit()
266
267 async with AsyncClient(
268 transport=ASGITransport(app=test_app), base_url="http://test"
269 ) as client:
270 response = await client.get(f"/artists/{artist.did}/analytics")
271
272 assert response.status_code == 200
273 data = response.json()
274
275 assert data["total_plays"] == 18 # 10 + 5 + 3
276 assert data["total_items"] == 3
277 # duration should sum only tracks that have it: 180 + 3600 = 3780
278 assert data["total_duration_seconds"] == 3780