Spaces:
Running
Running
| import unittest | |
| from unittest.mock import AsyncMock, patch | |
| from fastapi.testclient import TestClient | |
| from backend.main import app | |
| def _profile(role: str): | |
| return { | |
| "id": "profile-1", | |
| "email": "user@example.com", | |
| "role": role, | |
| "is_active": True, | |
| "monthly_review_limit": None, | |
| "monthly_cost_limit_usd": None, | |
| } | |
| class AdminAuthTests(unittest.TestCase): | |
| def setUp(self): | |
| self.client = TestClient(app) | |
| self.auth_user = {"id": "auth-1", "email": "user@example.com", "user_metadata": {}} | |
| def test_admin_route_requires_authentication(self): | |
| # No Authorization header → 401. | |
| response = self.client.get("/api/admin/overview") | |
| self.assertEqual(response.status_code, 401) | |
| def test_admin_route_forbidden_for_regular_user(self): | |
| with patch("backend.auth._verify_token", new=AsyncMock(return_value=self.auth_user)), \ | |
| patch("backend.auth.job_store.ensure_profile_for_auth_user", | |
| new=AsyncMock(return_value=_profile("user"))): | |
| response = self.client.get( | |
| "/api/admin/overview", headers={"Authorization": "Bearer faketoken"} | |
| ) | |
| self.assertEqual(response.status_code, 403) | |
| self.assertIn("Admin access required", response.json()["detail"]) | |
| def test_admin_route_ok_for_admin(self): | |
| with patch("backend.auth._verify_token", new=AsyncMock(return_value=self.auth_user)), \ | |
| patch("backend.auth.job_store.ensure_profile_for_auth_user", | |
| new=AsyncMock(return_value=_profile("admin"))), \ | |
| patch("backend.main.job_store.get_admin_overview", | |
| new=AsyncMock(return_value={"total_submissions": 5, "total_cost_usd": 1.23})), \ | |
| patch("backend.main.job_store.list_cost_by_day", new=AsyncMock(return_value=[])), \ | |
| patch("backend.main.job_store.list_cost_by_provider", new=AsyncMock(return_value=[])): | |
| response = self.client.get( | |
| "/api/admin/overview", headers={"Authorization": "Bearer faketoken"} | |
| ) | |
| self.assertEqual(response.status_code, 200) | |
| body = response.json() | |
| self.assertEqual(body["overview"]["total_submissions"], 5) | |
| def test_admin_cannot_demote_self(self): | |
| with patch("backend.auth._verify_token", new=AsyncMock(return_value=self.auth_user)), \ | |
| patch("backend.auth.job_store.ensure_profile_for_auth_user", | |
| new=AsyncMock(return_value=_profile("admin"))): | |
| response = self.client.patch( | |
| "/api/admin/users/profile-1", | |
| headers={"Authorization": "Bearer faketoken"}, | |
| json={"role": "user"}, | |
| ) | |
| self.assertEqual(response.status_code, 400) | |
| self.assertIn("your own admin role", response.json()["detail"]) | |
| def test_submit_still_anonymous_without_token(self): | |
| # Optional auth: with no Authorization header the submit path never calls | |
| # the verifier and proceeds anonymously. | |
| fake_job = {"access_key": "anon-key", "status": "pending"} | |
| with patch("backend.main.job_store.create_job", new=AsyncMock(return_value=fake_job)), \ | |
| patch("backend.main.send_access_key_email"), \ | |
| patch("backend.main._run_pipeline", new=AsyncMock()): | |
| import io | |
| response = self.client.post( | |
| "/api/submit", | |
| files={"file": ("paper.pdf", io.BytesIO(b"%PDF-1.4 x"), "application/pdf")}, | |
| data={"email": "anon@example.com"}, | |
| ) | |
| self.assertEqual(response.status_code, 200) | |
| self.assertEqual(response.json()["access_key"], "anon-key") | |
| if __name__ == "__main__": | |
| unittest.main() | |