zzstoatzz.io
audio streaming app
plyr.fm
1"""tests for activity feed endpoint."""
2
3from datetime import UTC, datetime, timedelta
4
5import pytest
6from fastapi.testclient import TestClient
7from sqlalchemy.ext.asyncio import AsyncSession
8
9from backend.models import Artist, Track, TrackComment, TrackLike
10
11
12@pytest.fixture
13async def artist(db_session: AsyncSession) -> Artist:
14 """create a test artist."""
15 artist = Artist(
16 did="did:plc:activity_artist",
17 handle="activity-artist.bsky.social",
18 display_name="Activity Artist",
19 )
20 db_session.add(artist)
21 await db_session.commit()
22 return artist
23
24
25@pytest.fixture
26async def other_artist(db_session: AsyncSession) -> Artist:
27 """create a second test artist."""
28 artist = Artist(
29 did="did:plc:activity_other",
30 handle="other-artist.bsky.social",
31 display_name="Other Artist",
32 )
33 db_session.add(artist)
34 await db_session.commit()
35 return artist
36
37
38@pytest.fixture
39async def track(db_session: AsyncSession, artist: Artist) -> Track:
40 """create a test track."""
41 track = Track(
42 title="Test Track",
43 artist_did=artist.did,
44 file_id="activity_track_1",
45 file_type="mp3",
46 image_url="https://example.com/image.jpg",
47 thumbnail_url="https://example.com/thumb.jpg",
48 )
49 db_session.add(track)
50 await db_session.commit()
51 return track
52
53
54async def test_empty_feed(client: TestClient, db_session: AsyncSession) -> None:
55 """empty database returns empty events with no cursor."""
56 response = client.get("/activity/")
57 assert response.status_code == 200
58 data = response.json()
59 assert data["events"] == []
60 assert data["next_cursor"] is None
61 assert data["has_more"] is False
62
63
64async def test_all_event_types(
65 client: TestClient,
66 db_session: AsyncSession,
67 artist: Artist,
68 other_artist: Artist,
69 track: Track,
70) -> None:
71 """all four event types appear with correct type field."""
72 like = TrackLike(
73 track_id=track.id,
74 user_did=other_artist.did,
75 )
76 comment = TrackComment(
77 track_id=track.id,
78 user_did=other_artist.did,
79 text="great track!",
80 timestamp_ms=5000,
81 )
82 db_session.add_all([like, comment])
83 await db_session.commit()
84
85 response = client.get("/activity/")
86 assert response.status_code == 200
87 data = response.json()
88
89 event_types = {e["type"] for e in data["events"]}
90 assert event_types == {"like", "track", "comment", "join"}
91
92
93async def test_chronological_order(
94 client: TestClient,
95 db_session: AsyncSession,
96 artist: Artist,
97 track: Track,
98) -> None:
99 """events are ordered by created_at DESC."""
100 response = client.get("/activity/")
101 assert response.status_code == 200
102 data = response.json()
103
104 timestamps = [e["created_at"] for e in data["events"]]
105 assert timestamps == sorted(timestamps, reverse=True)
106
107
108async def test_cursor_pagination(
109 client: TestClient,
110 db_session: AsyncSession,
111 artist: Artist,
112) -> None:
113 """cursor pagination returns two pages with no overlap."""
114 now = datetime.now(UTC)
115 tracks = []
116 for i in range(5):
117 t = Track(
118 title=f"Track {i}",
119 artist_did=artist.did,
120 file_id=f"pagination_{i}",
121 file_type="mp3",
122 created_at=now + timedelta(seconds=i + 1),
123 )
124 tracks.append(t)
125 db_session.add_all(tracks)
126 await db_session.commit()
127
128 # page 1 (limit=3: 5 tracks + 1 join = 6 total events, should have more)
129 resp1 = client.get("/activity/", params={"limit": 3})
130 assert resp1.status_code == 200
131 page1 = resp1.json()
132 assert page1["has_more"] is True
133 assert page1["next_cursor"] is not None
134
135 # page 2
136 resp2 = client.get(
137 "/activity/", params={"limit": 3, "cursor": page1["next_cursor"]}
138 )
139 assert resp2.status_code == 200
140 page2 = resp2.json()
141
142 # no overlap
143 page1_times = {e["created_at"] for e in page1["events"]}
144 page2_times = {e["created_at"] for e in page2["events"]}
145 assert page1_times.isdisjoint(page2_times)
146
147
148async def test_invalid_cursor(client: TestClient, db_session: AsyncSession) -> None:
149 """invalid cursor returns 400."""
150 response = client.get("/activity/", params={"cursor": "not-a-date"})
151 assert response.status_code == 400
152 assert "cursor" in response.json()["detail"].lower()
153
154
155async def test_limit_clamping(
156 client: TestClient,
157 db_session: AsyncSession,
158 artist: Artist,
159) -> None:
160 """limit is clamped: 0 → 1, 200 → 100."""
161 resp_low = client.get("/activity/", params={"limit": 0})
162 assert resp_low.status_code == 200
163
164 resp_high = client.get("/activity/", params={"limit": 200})
165 assert resp_high.status_code == 200
166
167
168async def test_like_includes_track_info(
169 client: TestClient,
170 db_session: AsyncSession,
171 artist: Artist,
172 other_artist: Artist,
173 track: Track,
174) -> None:
175 """like events include track info."""
176 like = TrackLike(
177 track_id=track.id,
178 user_did=other_artist.did,
179 )
180 db_session.add(like)
181 await db_session.commit()
182
183 response = client.get("/activity/")
184 assert response.status_code == 200
185 data = response.json()
186
187 like_events = [e for e in data["events"] if e["type"] == "like"]
188 assert len(like_events) >= 1
189 like_event = like_events[0]
190 assert like_event["track"] is not None
191 assert like_event["track"]["id"] == track.id
192 assert like_event["track"]["title"] == track.title
193 assert like_event["track"]["artist_handle"] == artist.handle
194
195
196async def test_comment_includes_text(
197 client: TestClient,
198 db_session: AsyncSession,
199 artist: Artist,
200 other_artist: Artist,
201 track: Track,
202) -> None:
203 """comment events include comment_text."""
204 comment = TrackComment(
205 track_id=track.id,
206 user_did=other_artist.did,
207 text="this slaps",
208 timestamp_ms=0,
209 )
210 db_session.add(comment)
211 await db_session.commit()
212
213 response = client.get("/activity/")
214 assert response.status_code == 200
215 data = response.json()
216
217 comment_events = [e for e in data["events"] if e["type"] == "comment"]
218 assert len(comment_events) >= 1
219 assert comment_events[0]["comment_text"] == "this slaps"
220 assert comment_events[0]["track"] is not None
221
222
223async def test_join_has_null_track(
224 client: TestClient,
225 db_session: AsyncSession,
226 artist: Artist,
227) -> None:
228 """join events have track: null."""
229 response = client.get("/activity/")
230 assert response.status_code == 200
231 data = response.json()
232
233 join_events = [e for e in data["events"] if e["type"] == "join"]
234 assert len(join_events) >= 1
235 assert join_events[0]["track"] is None
236 assert join_events[0]["comment_text"] is None