| import pytest |
| import jwt |
| import os |
| from fastapi import HTTPException |
| from src.utils.auth import create_dev_token, get_current_user, get_active_user |
| from fastapi.security import HTTPAuthorizationCredentials |
|
|
| def test_jwt_generation_and_decoding(): |
| patient_id = "test-patient-uuid" |
| token = create_dev_token(patient_id, expires_minutes=10) |
| assert token is not None |
| assert isinstance(token, str) |
|
|
| jwt_secret = os.getenv("SUPABASE_JWT_SECRET") |
| payload = jwt.decode(token, jwt_secret, algorithms=["HS256"], audience="authenticated") |
| assert payload["sub"] == patient_id |
| assert payload["aud"] == "authenticated" |
|
|
| def test_get_current_user_valid(): |
| patient_id = "patient-123" |
| token = create_dev_token(patient_id) |
| credentials = HTTPAuthorizationCredentials(scheme="Bearer", credentials=token) |
| |
| user_id = get_current_user(credentials) |
| assert user_id == patient_id |
|
|
| def test_get_current_user_invalid(): |
| |
| invalid_token = jwt.encode({"sub": "patient-123", "aud": "authenticated"}, "wrong-secret", algorithm="HS256") |
| credentials = HTTPAuthorizationCredentials(scheme="Bearer", credentials=invalid_token) |
| |
| with pytest.raises(HTTPException) as exc_info: |
| get_current_user(credentials) |
| assert exc_info.value.status_code == 401 |
| assert "Invalid token" in exc_info.value.detail |
|
|
| def test_get_active_user_fallback(clean_db): |
| from unittest.mock import patch |
| |
| with patch("src.utils.auth.DEV_ALLOW_DEFAULT_PATIENT", True), patch("src.utils.auth.DEFAULT_PATIENT_ID", "fallback-uuid"): |
| |
| user_id = get_active_user(None) |
| assert user_id == "fallback-uuid" |
|
|
| |
| with patch("src.utils.auth.DEV_ALLOW_DEFAULT_PATIENT", False): |
| with pytest.raises(HTTPException) as exc_info: |
| get_active_user(None) |
| assert exc_info.value.status_code == 401 |
|
|
| def test_john_doe_registration_and_login_flow(client_main, clean_db): |
| username = "JohnDoe" |
| |
| |
| login_resp = client_main.post("/auth/login", json={"username": username}) |
| assert login_resp.status_code == 404 |
| assert "Username not found" in login_resp.json()["detail"] |
|
|
| |
| reg_payload = { |
| "username": username, |
| "first_name": "John", |
| "last_name": "Doe" |
| } |
| reg_resp = client_main.post("/auth/register", json=reg_payload) |
| assert reg_resp.status_code == 200 |
| reg_data = reg_resp.json() |
| assert reg_data["status"] == "ok" |
| assert "patient_id" in reg_data |
| assert "token" in reg_data |
| |
| patient_id = reg_data["patient_id"] |
| token = reg_data["token"] |
|
|
| |
| db_patient = clean_db.db["patients"][patient_id] |
| assert db_patient["resource"]["username"] == username |
| assert db_patient["resource"]["name"][0]["text"] == "John Doe" |
|
|
| |
| dup_resp = client_main.post("/auth/register", json=reg_payload) |
| assert dup_resp.status_code == 400 |
| assert "Username already exists" in dup_resp.json()["detail"] |
|
|
| |
| login_resp2 = client_main.post("/auth/login", json={"username": username}) |
| assert login_resp2.status_code == 200 |
| login_data = login_resp2.json() |
| assert login_data["status"] == "ok" |
| assert login_data["patient_id"] == patient_id |
| assert "token" in login_data |
|
|
| |
| credentials = HTTPAuthorizationCredentials(scheme="Bearer", credentials=login_data["token"]) |
| user_id = get_current_user(credentials) |
| assert user_id == patient_id |
|
|
