140 lines
4.1 KiB
Python
140 lines
4.1 KiB
Python
from types import SimpleNamespace
|
|
from unittest.mock import AsyncMock
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from app.api.v1.endpoints import auth as auth_endpoint
|
|
from app.auth.dependencies import get_current_active_user, get_user_repository
|
|
from app.core.security import create_access_token, create_refresh_token, get_password_hash
|
|
from tests.conftest import build_test_app, make_user
|
|
|
|
|
|
def _build_client(repo, current_user=None) -> TestClient:
|
|
app = build_test_app(auth_endpoint.router, "/api/v1/auth")
|
|
app.dependency_overrides[get_user_repository] = lambda: repo
|
|
if current_user is not None:
|
|
app.dependency_overrides[get_current_active_user] = lambda: current_user
|
|
return TestClient(app)
|
|
|
|
|
|
def test_register_success():
|
|
repo = SimpleNamespace(
|
|
user_exists=AsyncMock(side_effect=[False, False]),
|
|
create_user=AsyncMock(return_value=make_user()),
|
|
)
|
|
client = _build_client(repo)
|
|
|
|
response = client.post(
|
|
"/api/v1/auth/register",
|
|
json={
|
|
"username": "tester",
|
|
"email": "tester@example.com",
|
|
"password": "secret123",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 201
|
|
assert response.json()["username"] == "tester"
|
|
|
|
|
|
def test_register_rejects_duplicate_username():
|
|
repo = SimpleNamespace(
|
|
user_exists=AsyncMock(side_effect=[True]),
|
|
create_user=AsyncMock(),
|
|
)
|
|
client = _build_client(repo)
|
|
|
|
response = client.post(
|
|
"/api/v1/auth/register",
|
|
json={
|
|
"username": "tester",
|
|
"email": "tester@example.com",
|
|
"password": "secret123",
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 400
|
|
assert response.json()["detail"] == "Username already registered"
|
|
repo.create_user.assert_not_awaited()
|
|
|
|
|
|
def test_login_supports_email_lookup():
|
|
hashed_password = get_password_hash("secret123")
|
|
repo = SimpleNamespace(
|
|
get_user_by_username=AsyncMock(return_value=None),
|
|
get_user_by_email=AsyncMock(
|
|
return_value=make_user(
|
|
email="tester@example.com",
|
|
hashed_password=hashed_password,
|
|
)
|
|
),
|
|
)
|
|
client = _build_client(repo)
|
|
|
|
response = client.post(
|
|
"/api/v1/auth/login",
|
|
data={"username": "tester@example.com", "password": "secret123"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["token_type"] == "bearer"
|
|
repo.get_user_by_email.assert_awaited_once_with("tester@example.com")
|
|
|
|
|
|
def test_login_simple_uses_query_params():
|
|
hashed_password = get_password_hash("secret123")
|
|
repo = SimpleNamespace(
|
|
get_user_by_username=AsyncMock(
|
|
return_value=make_user(hashed_password=hashed_password)
|
|
),
|
|
get_user_by_email=AsyncMock(),
|
|
)
|
|
client = _build_client(repo)
|
|
|
|
response = client.post(
|
|
"/api/v1/auth/login/simple",
|
|
params={"username": "tester", "password": "secret123"},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["token_type"] == "bearer"
|
|
|
|
|
|
def test_me_returns_current_user_info():
|
|
client = _build_client(SimpleNamespace(), current_user=make_user(username="alice"))
|
|
|
|
response = client.get("/api/v1/auth/me")
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()["username"] == "alice"
|
|
|
|
|
|
def test_refresh_rejects_access_token():
|
|
repo = SimpleNamespace(get_user_by_username=AsyncMock())
|
|
client = _build_client(repo)
|
|
|
|
response = client.post(
|
|
"/api/v1/auth/refresh",
|
|
params={"refresh_token": create_access_token("tester")},
|
|
)
|
|
|
|
assert response.status_code == 401
|
|
|
|
|
|
def test_refresh_success_returns_new_access_token():
|
|
repo = SimpleNamespace(
|
|
get_user_by_username=AsyncMock(return_value=make_user()),
|
|
)
|
|
client = _build_client(repo)
|
|
refresh_token = create_refresh_token("tester")
|
|
|
|
response = client.post(
|
|
"/api/v1/auth/refresh",
|
|
params={"refresh_token": refresh_token},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
payload = response.json()
|
|
assert payload["refresh_token"] == refresh_token
|
|
assert payload["token_type"] == "bearer"
|