77 lines
3.0 KiB
Python
77 lines
3.0 KiB
Python
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from ea_chatbot.api.main import app
|
|
from ea_chatbot.api.utils import create_refresh_token, create_access_token
|
|
|
|
from unittest.mock import patch, MagicMock
|
|
from ea_chatbot.history.models import User
|
|
|
|
@pytest.fixture
|
|
def client():
|
|
return TestClient(app)
|
|
|
|
def test_refresh_token_success(client):
|
|
# 1. Setup: Create a valid refresh token for a user
|
|
user_id = "test-user-id"
|
|
refresh_token = create_refresh_token({"sub": user_id})
|
|
|
|
# 2. Set the cookie manually in the client
|
|
client.cookies.set("refresh_token", refresh_token)
|
|
|
|
# 3. Call the refresh endpoint with mock
|
|
with patch("ea_chatbot.api.routers.auth.history_manager") as mock_hm, \
|
|
patch("ea_chatbot.api.utils.datetime") as mock_datetime:
|
|
|
|
# Mock datetime to ensure the second token has a different timestamp
|
|
from datetime import datetime, timezone, timedelta
|
|
base_now = datetime.now(timezone.utc)
|
|
# First call (inside refresh) gets base_now + 1 second
|
|
mock_datetime.now.return_value = base_now + timedelta(seconds=1)
|
|
mock_datetime.side_effect = lambda *args, **kwargs: datetime(*args, **kwargs)
|
|
|
|
mock_hm.get_user_by_id.return_value = User(id=user_id, username="test@test.com")
|
|
|
|
response = client.post("/api/v1/auth/refresh")
|
|
|
|
# 4. Assert success
|
|
assert response.status_code == 200
|
|
assert "access_token" in response.cookies
|
|
assert "refresh_token" in response.cookies
|
|
|
|
# Verify tokens are rotated (different from original)
|
|
assert response.cookies["refresh_token"] != refresh_token
|
|
|
|
def test_refresh_token_missing_cookie(client):
|
|
response = client.post("/api/v1/auth/refresh")
|
|
assert response.status_code == 401
|
|
assert response.json()["detail"] == "Refresh token missing"
|
|
|
|
def test_refresh_token_invalid(client):
|
|
client.cookies.set("refresh_token", "invalid-token")
|
|
response = client.post("/api/v1/auth/refresh")
|
|
assert response.status_code == 401
|
|
assert response.json()["detail"] == "Invalid refresh token"
|
|
|
|
def test_refresh_token_wrong_type(client):
|
|
# Using an access token as a refresh token should fail
|
|
access_token = create_access_token({"sub": "user123"})
|
|
client.cookies.set("refresh_token", access_token)
|
|
|
|
response = client.post("/api/v1/auth/refresh")
|
|
assert response.status_code == 401
|
|
assert response.json()["detail"] == "Invalid token type"
|
|
|
|
def test_protected_endpoint_rejects_refresh_token(client):
|
|
"""Regression test: Ensure refresh tokens cannot be used to access protected endpoints."""
|
|
user_id = "test-user-id"
|
|
refresh_token = create_refresh_token({"sub": user_id})
|
|
|
|
# Attempt to access /auth/me with a refresh token in the cookie
|
|
client.cookies.set("access_token", refresh_token)
|
|
|
|
response = client.get("/api/v1/auth/me")
|
|
|
|
# Should be rejected with 401
|
|
assert response.status_code == 401
|
|
assert "Cannot use refresh token for this endpoint" in response.json()["detail"]
|