zzstoatzz.io
audio streaming app
plyr.fm
1"""tests for oEmbed endpoint."""
2
3import pytest
4from fastapi import FastAPI
5from httpx import ASGITransport, AsyncClient
6from sqlalchemy.ext.asyncio import AsyncSession
7
8from backend.main import app
9from backend.models import Album, Artist, Playlist, Track
10
11
12@pytest.fixture
13async def test_track(db_session: AsyncSession) -> Track:
14 """create a test track for oEmbed testing."""
15 artist = Artist(
16 did="did:plc:oembed123",
17 handle="test.artist.social",
18 display_name="Test Artist",
19 )
20 db_session.add(artist)
21 await db_session.flush()
22
23 track = Track(
24 title="Test Track",
25 artist_did=artist.did,
26 file_id="oembed_test_123",
27 file_type="mp3",
28 r2_url="https://cdn.example.com/audio/test.mp3",
29 image_url="https://cdn.example.com/images/cover.png",
30 )
31 db_session.add(track)
32 await db_session.commit()
33 await db_session.refresh(track)
34
35 return track
36
37
38@pytest.fixture
39async def test_playlist(db_session: AsyncSession) -> Playlist:
40 """create a test playlist for oEmbed testing."""
41 artist = Artist(
42 did="did:plc:oembed_pl",
43 handle="playlist.owner.social",
44 display_name="Playlist Owner",
45 )
46 db_session.add(artist)
47 await db_session.flush()
48
49 playlist = Playlist(
50 name="Test Playlist",
51 owner_did=artist.did,
52 image_url="https://cdn.example.com/images/playlist.png",
53 atproto_record_uri="at://did:plc:oembed_pl/fm.plyr.playlist/test",
54 atproto_record_cid="bafytest",
55 )
56 db_session.add(playlist)
57 await db_session.commit()
58 await db_session.refresh(playlist)
59
60 return playlist
61
62
63@pytest.fixture
64async def test_album(db_session: AsyncSession) -> Album:
65 """create a test album for oEmbed testing."""
66 artist = Artist(
67 did="did:plc:oembed_al",
68 handle="album.artist.social",
69 display_name="Album Artist",
70 )
71 db_session.add(artist)
72 await db_session.flush()
73
74 album = Album(
75 title="Test Album",
76 slug="test-album",
77 artist_did=artist.did,
78 image_url="https://cdn.example.com/images/album.png",
79 )
80 db_session.add(album)
81 await db_session.commit()
82 await db_session.refresh(album)
83
84 return album
85
86
87@pytest.fixture
88def test_app(db_session: AsyncSession) -> FastAPI:
89 """get test app with db session dependency to ensure correct database URL."""
90 _ = db_session # ensures database fixtures run first
91 return app
92
93
94async def test_oembed_returns_valid_response(
95 test_app: FastAPI, test_track: Track
96) -> None:
97 """test that oEmbed returns proper response for valid track URL."""
98 async with AsyncClient(
99 transport=ASGITransport(app=test_app), base_url="http://test"
100 ) as client:
101 response = await client.get(
102 "/oembed",
103 params={"url": f"https://plyr.fm/track/{test_track.id}"},
104 )
105
106 assert response.status_code == 200
107 data = response.json()
108
109 assert data["version"] == "1.0"
110 assert data["type"] == "rich"
111 assert data["provider_name"] == "plyr.fm"
112 assert "Test Track" in data["title"]
113 assert "Test Artist" in data["title"]
114 assert data["author_name"] == "Test Artist"
115 assert f"/embed/track/{test_track.id}" in data["html"]
116 assert "iframe" in data["html"]
117 assert data["height"] == 165
118 # should have thumbnail since track has image
119 assert data["thumbnail_url"] == test_track.image_url
120
121
122async def test_oembed_handles_encoded_url(test_app: FastAPI, test_track: Track) -> None:
123 """test that oEmbed handles URL-encoded URLs."""
124 import urllib.parse
125
126 encoded_url = urllib.parse.quote(f"https://plyr.fm/track/{test_track.id}", safe="")
127
128 async with AsyncClient(
129 transport=ASGITransport(app=test_app), base_url="http://test"
130 ) as client:
131 response = await client.get("/oembed", params={"url": encoded_url})
132
133 assert response.status_code == 200
134 data = response.json()
135 assert f"/embed/track/{test_track.id}" in data["html"]
136
137
138async def test_oembed_returns_404_for_invalid_url(test_app: FastAPI) -> None:
139 """test that oEmbed returns 404 for unrecognized URLs."""
140 async with AsyncClient(
141 transport=ASGITransport(app=test_app), base_url="http://test"
142 ) as client:
143 response = await client.get(
144 "/oembed", params={"url": "https://plyr.fm/not-a-thing"}
145 )
146
147 assert response.status_code == 404
148 assert "unsupported URL format" in response.json()["detail"]
149
150
151async def test_oembed_returns_404_for_nonexistent_track(test_app: FastAPI) -> None:
152 """test that oEmbed returns 404 for track that doesn't exist."""
153 async with AsyncClient(
154 transport=ASGITransport(app=test_app), base_url="http://test"
155 ) as client:
156 response = await client.get(
157 "/oembed", params={"url": "https://plyr.fm/track/99999"}
158 )
159
160 assert response.status_code == 404
161 assert "track not found" in response.json()["detail"]
162
163
164async def test_oembed_rejects_non_json_format(
165 test_app: FastAPI, test_track: Track
166) -> None:
167 """test that oEmbed returns 501 for non-JSON format."""
168 async with AsyncClient(
169 transport=ASGITransport(app=test_app), base_url="http://test"
170 ) as client:
171 response = await client.get(
172 "/oembed",
173 params={
174 "url": f"https://plyr.fm/track/{test_track.id}",
175 "format": "xml",
176 },
177 )
178
179 assert response.status_code == 501
180 assert "only json format is supported" in response.json()["detail"]
181
182
183async def test_oembed_respects_maxwidth(test_app: FastAPI, test_track: Track) -> None:
184 """test that oEmbed respects maxwidth parameter."""
185 async with AsyncClient(
186 transport=ASGITransport(app=test_app), base_url="http://test"
187 ) as client:
188 response = await client.get(
189 "/oembed",
190 params={
191 "url": f"https://plyr.fm/track/{test_track.id}",
192 "maxwidth": 300,
193 },
194 )
195
196 assert response.status_code == 200
197 data = response.json()
198 assert data["width"] == 300
199
200
201async def test_oembed_playlist_returns_valid_response(
202 test_app: FastAPI, test_playlist: Playlist
203) -> None:
204 """test that oEmbed returns proper response for valid playlist URL."""
205 async with AsyncClient(
206 transport=ASGITransport(app=test_app), base_url="http://test"
207 ) as client:
208 response = await client.get(
209 "/oembed",
210 params={"url": f"https://plyr.fm/playlist/{test_playlist.id}"},
211 )
212
213 assert response.status_code == 200
214 data = response.json()
215
216 assert data["version"] == "1.0"
217 assert data["type"] == "rich"
218 assert "Test Playlist" in data["title"]
219 assert "Playlist Owner" in data["title"]
220 assert data["author_name"] == "Playlist Owner"
221 assert f"/embed/playlist/{test_playlist.id}" in data["html"]
222 assert "iframe" in data["html"]
223 assert data["height"] == 380
224 assert data["thumbnail_url"] == test_playlist.image_url
225
226
227async def test_oembed_album_returns_valid_response(
228 test_app: FastAPI, test_album: Album
229) -> None:
230 """test that oEmbed returns proper response for valid album URL."""
231 async with AsyncClient(
232 transport=ASGITransport(app=test_app), base_url="http://test"
233 ) as client:
234 response = await client.get(
235 "/oembed",
236 params={"url": "https://plyr.fm/u/album.artist.social/album/test-album"},
237 )
238
239 assert response.status_code == 200
240 data = response.json()
241
242 assert data["version"] == "1.0"
243 assert data["type"] == "rich"
244 assert "Test Album" in data["title"]
245 assert "Album Artist" in data["title"]
246 assert data["author_name"] == "Album Artist"
247 assert "/embed/album/album.artist.social/test-album" in data["html"]
248 assert "iframe" in data["html"]
249 assert data["height"] == 380
250 assert data["thumbnail_url"] == test_album.image_url
251
252
253async def test_oembed_playlist_not_found(test_app: FastAPI) -> None:
254 """test that oEmbed returns 404 for nonexistent playlist."""
255 async with AsyncClient(
256 transport=ASGITransport(app=test_app), base_url="http://test"
257 ) as client:
258 response = await client.get(
259 "/oembed",
260 params={
261 "url": "https://plyr.fm/playlist/00000000-0000-0000-0000-000000000000"
262 },
263 )
264
265 assert response.status_code == 404
266 assert "playlist not found" in response.json()["detail"]
267
268
269async def test_oembed_album_not_found(test_app: FastAPI) -> None:
270 """test that oEmbed returns 404 for nonexistent album."""
271 async with AsyncClient(
272 transport=ASGITransport(app=test_app), base_url="http://test"
273 ) as client:
274 response = await client.get(
275 "/oembed",
276 params={"url": "https://plyr.fm/u/nobody/album/no-album"},
277 )
278
279 assert response.status_code == 404
280 assert "album not found" in response.json()["detail"]
281
282
283async def test_oembed_collection_respects_maxheight(
284 test_app: FastAPI, test_playlist: Playlist
285) -> None:
286 """test that playlist/album oEmbed caps height at 600px."""
287 async with AsyncClient(
288 transport=ASGITransport(app=test_app), base_url="http://test"
289 ) as client:
290 response = await client.get(
291 "/oembed",
292 params={
293 "url": f"https://plyr.fm/playlist/{test_playlist.id}",
294 "maxheight": 9999,
295 },
296 )
297
298 assert response.status_code == 200
299 data = response.json()
300 assert data["height"] == 600