music on atproto
plyr.fm
zzstoatzz.io
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 Artist, 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
39def test_app(db_session: AsyncSession) -> FastAPI:
40 """get test app with db session dependency to ensure correct database URL."""
41 _ = db_session # ensures database fixtures run first
42 return app
43
44
45async def test_oembed_returns_valid_response(
46 test_app: FastAPI, test_track: Track
47) -> None:
48 """test that oEmbed returns proper response for valid track URL."""
49 async with AsyncClient(
50 transport=ASGITransport(app=test_app), base_url="http://test"
51 ) as client:
52 response = await client.get(
53 "/oembed",
54 params={"url": f"https://plyr.fm/track/{test_track.id}"},
55 )
56
57 assert response.status_code == 200
58 data = response.json()
59
60 assert data["version"] == "1.0"
61 assert data["type"] == "rich"
62 assert data["provider_name"] == "plyr.fm"
63 assert "Test Track" in data["title"]
64 assert "Test Artist" in data["title"]
65 assert data["author_name"] == "Test Artist"
66 assert f"/embed/track/{test_track.id}" in data["html"]
67 assert "iframe" in data["html"]
68 assert data["height"] == 165
69 # should have thumbnail since track has image
70 assert data["thumbnail_url"] == test_track.image_url
71
72
73async def test_oembed_handles_encoded_url(test_app: FastAPI, test_track: Track) -> None:
74 """test that oEmbed handles URL-encoded URLs."""
75 import urllib.parse
76
77 encoded_url = urllib.parse.quote(f"https://plyr.fm/track/{test_track.id}", safe="")
78
79 async with AsyncClient(
80 transport=ASGITransport(app=test_app), base_url="http://test"
81 ) as client:
82 response = await client.get("/oembed", params={"url": encoded_url})
83
84 assert response.status_code == 200
85 data = response.json()
86 assert f"/embed/track/{test_track.id}" in data["html"]
87
88
89async def test_oembed_returns_404_for_invalid_url(test_app: FastAPI) -> None:
90 """test that oEmbed returns 404 for non-track URLs."""
91 async with AsyncClient(
92 transport=ASGITransport(app=test_app), base_url="http://test"
93 ) as client:
94 response = await client.get(
95 "/oembed", params={"url": "https://plyr.fm/not-a-track"}
96 )
97
98 assert response.status_code == 404
99 assert "invalid track URL" in response.json()["detail"]
100
101
102async def test_oembed_returns_404_for_nonexistent_track(test_app: FastAPI) -> None:
103 """test that oEmbed returns 404 for track that doesn't exist."""
104 async with AsyncClient(
105 transport=ASGITransport(app=test_app), base_url="http://test"
106 ) as client:
107 response = await client.get(
108 "/oembed", params={"url": "https://plyr.fm/track/99999"}
109 )
110
111 assert response.status_code == 404
112 assert "track not found" in response.json()["detail"]
113
114
115async def test_oembed_rejects_non_json_format(
116 test_app: FastAPI, test_track: Track
117) -> None:
118 """test that oEmbed returns 501 for non-JSON format."""
119 async with AsyncClient(
120 transport=ASGITransport(app=test_app), base_url="http://test"
121 ) as client:
122 response = await client.get(
123 "/oembed",
124 params={
125 "url": f"https://plyr.fm/track/{test_track.id}",
126 "format": "xml",
127 },
128 )
129
130 assert response.status_code == 501
131 assert "only json format is supported" in response.json()["detail"]
132
133
134async def test_oembed_respects_maxwidth(test_app: FastAPI, test_track: Track) -> None:
135 """test that oEmbed respects maxwidth parameter."""
136 async with AsyncClient(
137 transport=ASGITransport(app=test_app), base_url="http://test"
138 ) as client:
139 response = await client.get(
140 "/oembed",
141 params={
142 "url": f"https://plyr.fm/track/{test_track.id}",
143 "maxwidth": 300,
144 },
145 )
146
147 assert response.status_code == 200
148 data = response.json()
149 assert data["width"] == 300