zzstoatzz.io
audio streaming app
plyr.fm
1"""tests for developer token api endpoints."""
2
3from collections.abc import Generator
4from unittest.mock import AsyncMock, patch
5
6import pytest
7from fastapi import FastAPI
8from httpx import ASGITransport, AsyncClient
9from sqlalchemy.ext.asyncio import AsyncSession
10
11from backend._internal import Session, create_session, require_auth
12from backend.main import app
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
52async def test_start_developer_token_flow(test_app: FastAPI, db_session: AsyncSession):
53 """test starting the developer token OAuth flow."""
54 with patch(
55 "backend.api.auth.start_oauth_flow", new_callable=AsyncMock
56 ) as mock_oauth:
57 mock_oauth.return_value = (
58 "https://auth.example.com/authorize?...",
59 "test_state",
60 )
61
62 async with AsyncClient(
63 transport=ASGITransport(app=test_app), base_url="http://test"
64 ) as client:
65 response = await client.post(
66 "/auth/developer-token/start",
67 json={"name": "my-token", "expires_in_days": 30},
68 )
69
70 assert response.status_code == 200
71 data = response.json()
72 assert "auth_url" in data
73 assert data["auth_url"].startswith("https://auth.example.com")
74 mock_oauth.assert_called_once_with("testuser.bsky.social")
75
76
77async def test_start_developer_token_default_expiration(
78 test_app: FastAPI, db_session: AsyncSession
79):
80 """test starting dev token flow with default expiration."""
81 with patch(
82 "backend.api.auth.start_oauth_flow", new_callable=AsyncMock
83 ) as mock_oauth:
84 mock_oauth.return_value = ("https://auth.example.com/authorize", "test_state")
85
86 async with AsyncClient(
87 transport=ASGITransport(app=test_app), base_url="http://test"
88 ) as client:
89 response = await client.post(
90 "/auth/developer-token/start",
91 json={},
92 )
93
94 assert response.status_code == 200
95 # verify pending dev token was saved (would fail if expiration wasn't set)
96
97
98async def test_start_developer_token_exceeds_max(
99 test_app: FastAPI, db_session: AsyncSession
100):
101 """test that expiration cannot exceed max allowed."""
102 async with AsyncClient(
103 transport=ASGITransport(app=test_app), base_url="http://test"
104 ) as client:
105 response = await client.post(
106 "/auth/developer-token/start",
107 json={"expires_in_days": 999}, # exceeds default max of 365
108 )
109
110 assert response.status_code == 400
111 assert "cannot exceed" in response.json()["detail"]
112
113
114async def test_start_developer_token_requires_auth(db_session: AsyncSession):
115 """test that developer token start requires authentication."""
116 async with AsyncClient(
117 transport=ASGITransport(app=app), base_url="http://test"
118 ) as client:
119 response = await client.post(
120 "/auth/developer-token/start",
121 json={},
122 )
123
124 assert response.status_code == 401
125
126
127async def test_list_developer_tokens(test_app: FastAPI, db_session: AsyncSession):
128 """test listing developer tokens."""
129 mock_session = MockSession()
130
131 # create a dev token directly in the database
132 await create_session(
133 did=mock_session.did,
134 handle=mock_session.handle,
135 oauth_session=mock_session.oauth_session,
136 expires_in_days=30,
137 is_developer_token=True,
138 token_name="list-test-token",
139 )
140
141 async with AsyncClient(
142 transport=ASGITransport(app=test_app), base_url="http://test"
143 ) as client:
144 response = await client.get("/auth/developer-tokens")
145
146 assert response.status_code == 200
147 data = response.json()
148 assert "tokens" in data
149 assert len(data["tokens"]) >= 1
150
151 # find our token
152 token = next((t for t in data["tokens"] if t["name"] == "list-test-token"), None)
153 assert token is not None
154 assert "session_id" in token
155 assert "created_at" in token
156
157
158async def test_revoke_developer_token(test_app: FastAPI, db_session: AsyncSession):
159 """test revoking a developer token."""
160 mock_session = MockSession()
161
162 # create a dev token directly
163 await create_session(
164 did=mock_session.did,
165 handle=mock_session.handle,
166 oauth_session=mock_session.oauth_session,
167 expires_in_days=30,
168 is_developer_token=True,
169 token_name="revoke-test-token",
170 )
171
172 async with AsyncClient(
173 transport=ASGITransport(app=test_app), base_url="http://test"
174 ) as client:
175 # list tokens to get session_id prefix
176 list_response = await client.get("/auth/developer-tokens")
177 assert list_response.status_code == 200
178 tokens = list_response.json()["tokens"]
179 token = next((t for t in tokens if t["name"] == "revoke-test-token"), None)
180 assert token is not None
181
182 # revoke the token
183 revoke_response = await client.delete(
184 f"/auth/developer-tokens/{token['session_id']}"
185 )
186 assert revoke_response.status_code == 200
187 assert revoke_response.json()["message"] == "token revoked successfully"
188
189 # verify it's gone
190 final_list = await client.get("/auth/developer-tokens")
191 remaining = [
192 t for t in final_list.json()["tokens"] if t["name"] == "revoke-test-token"
193 ]
194 assert len(remaining) == 0
195
196
197async def test_revoke_nonexistent_token(test_app: FastAPI, db_session: AsyncSession):
198 """test revoking a token that doesn't exist."""
199 async with AsyncClient(
200 transport=ASGITransport(app=test_app), base_url="http://test"
201 ) as client:
202 response = await client.delete("/auth/developer-tokens/nonexist")
203
204 assert response.status_code == 404
205 assert response.json()["detail"] == "token not found"