| """Tests for auth middleware — get_current_user dependency.""" |
| import time |
| import hashlib |
| import pytest |
| from httpx import AsyncClient, ASGITransport |
| from app.main import create_app |
|
|
|
|
| def _make_token(email: str) -> str: |
| """Mirror the backend token generation for test fixtures.""" |
| secret = "aperture-mvp-secret-change-in-production" |
| hour = int(time.time() // 3600) |
| payload = f"{email}:{hour}:{secret}" |
| return hashlib.sha256(payload.encode()).hexdigest()[:32] |
|
|
|
|
| @pytest.fixture |
| async def client(tmp_path): |
| app = create_app(db_path=str(tmp_path / "test.db"), run_worker=False) |
| transport = ASGITransport(app=app) |
| async with AsyncClient(transport=transport, base_url="http://test") as c: |
| yield c |
|
|
|
|
| @pytest.mark.asyncio |
| async def test_auth_header_valid(client): |
| email = "user@example.com" |
| token = _make_token(email) |
| resp = await client.get( |
| "/api/jobs", |
| headers={"Authorization": f"Bearer {email}:{token}"}, |
| ) |
| assert resp.status_code == 200 |
|
|
|
|
| @pytest.mark.asyncio |
| async def test_auth_header_missing(client): |
| resp = await client.get("/api/jobs") |
| assert resp.status_code == 401 |
|
|
|
|
| @pytest.mark.asyncio |
| async def test_auth_header_bad_token(client): |
| resp = await client.get( |
| "/api/jobs", |
| headers={"Authorization": "Bearer user@example.com:badtoken"}, |
| ) |
| assert resp.status_code == 401 |
|
|
|
|
| @pytest.mark.asyncio |
| async def test_auth_header_malformed(client): |
| resp = await client.get( |
| "/api/jobs", |
| headers={"Authorization": "Bearer garbage"}, |
| ) |
| assert resp.status_code == 401 |
|
|
