Adds auth module

This commit is contained in:
Urtzi Alfaro
2025-07-17 21:25:27 +02:00
parent 654d1c2fe8
commit efca9a125a
19 changed files with 1169 additions and 406 deletions

View File

@@ -0,0 +1 @@
"""Authentication service tests"""

View File

@@ -0,0 +1,69 @@
# ================================================================
# services/auth/tests/conftest.py
# ================================================================
"""Test configuration for auth service"""
import pytest
import asyncio
from typing import AsyncGenerator
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
from fastapi.testclient import TestClient
from app.main import app
from app.core.database import get_db
from shared.database.base import Base
# Test database URL
TEST_DATABASE_URL = "postgresql+asyncpg://test_user:test_pass@localhost:5433/test_auth_db"
# Create test engine
test_engine = create_async_engine(TEST_DATABASE_URL, echo=False)
# Create test session
TestSessionLocal = sessionmaker(
test_engine, class_=AsyncSession, expire_on_commit=False
)
@pytest.fixture(scope="session")
def event_loop():
"""Create an instance of the default event loop for the test session."""
loop = asyncio.get_event_loop_policy().new_event_loop()
yield loop
loop.close()
@pytest.fixture
async def db() -> AsyncGenerator[AsyncSession, None]:
"""Database fixture"""
async with test_engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
async with TestSessionLocal() as session:
yield session
async with test_engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
@pytest.fixture
def client(db: AsyncSession):
"""Test client fixture"""
def override_get_db():
yield db
app.dependency_overrides[get_db] = override_get_db
with TestClient(app) as test_client:
yield test_client
app.dependency_overrides.clear()
@pytest.fixture
def test_user_data():
"""Test user data fixture"""
return {
"email": "test@bakery.es",
"password": "TestPass123",
"full_name": "Test User",
"phone": "+34123456789",
"language": "es"
}

View File

@@ -0,0 +1,79 @@
# ================================================================
# services/auth/tests/test_auth.py
# ================================================================
"""Authentication tests"""
import pytest
from fastapi.testclient import TestClient
from sqlalchemy.ext.asyncio import AsyncSession
from app.services.auth_service import AuthService
from app.schemas.auth import UserRegistration, UserLogin
@pytest.mark.asyncio
async def test_register_user(db: AsyncSession):
"""Test user registration"""
user_data = UserRegistration(
email="test@bakery.es",
password="TestPass123",
full_name="Test User",
language="es"
)
result = await AuthService.register_user(user_data, db)
assert result.email == "test@bakery.es"
assert result.full_name == "Test User"
assert result.is_active is True
assert result.is_verified is False
@pytest.mark.asyncio
async def test_login_user(db: AsyncSession):
"""Test user login"""
# First register a user
user_data = UserRegistration(
email="test@bakery.es",
password="TestPass123",
full_name="Test User",
language="es"
)
await AuthService.register_user(user_data, db)
# Then login
login_data = UserLogin(
email="test@bakery.es",
password="TestPass123"
)
result = await AuthService.login_user(login_data, db, "127.0.0.1", "test-agent")
assert result.access_token is not None
assert result.refresh_token is not None
assert result.token_type == "bearer"
def test_register_endpoint(client: TestClient, test_user_data):
"""Test registration endpoint"""
response = client.post("/auth/register", json=test_user_data)
assert response.status_code == 200
data = response.json()
assert data["email"] == test_user_data["email"]
assert "id" in data
def test_login_endpoint(client: TestClient, test_user_data):
"""Test login endpoint"""
# First register
client.post("/auth/register", json=test_user_data)
# Then login
login_data = {
"email": test_user_data["email"],
"password": test_user_data["password"]
}
response = client.post("/auth/login", json=login_data)
assert response.status_code == 200
data = response.json()
assert "access_token" in data
assert "refresh_token" in data
assert data["token_type"] == "bearer"

View File

@@ -0,0 +1,74 @@
# ================================================================
# services/auth/tests/test_users.py
# ================================================================
"""User management tests"""
import pytest
from fastapi.testclient import TestClient
from sqlalchemy.ext.asyncio import AsyncSession
from app.services.user_service import UserService
from app.services.auth_service import AuthService
from app.schemas.auth import UserRegistration
@pytest.mark.asyncio
async def test_get_user_by_email(db: AsyncSession):
"""Test getting user by email"""
# Create a user first
user_data = UserRegistration(
email="test@bakery.es",
password="TestPass123",
full_name="Test User",
language="es"
)
created_user = await AuthService.register_user(user_data, db)
# Get user by email
user = await UserService.get_user_by_email("test@bakery.es", db)
assert user is not None
assert user.email == "test@bakery.es"
assert str(user.id) == created_user.id
@pytest.mark.asyncio
async def test_update_user(db: AsyncSession):
"""Test updating user"""
# Create a user first
user_data = UserRegistration(
email="test@bakery.es",
password="TestPass123",
full_name="Test User",
language="es"
)
created_user = await AuthService.register_user(user_data, db)
# Update user
update_data = {
"full_name": "Updated User",
"phone": "+34987654321"
}
updated_user = await UserService.update_user(created_user.id, update_data, db)
assert updated_user.full_name == "Updated User"
assert updated_user.phone == "+34987654321"
def test_get_current_user_endpoint(client: TestClient, test_user_data):
"""Test get current user endpoint"""
# Register and login first
client.post("/auth/register", json=test_user_data)
login_response = client.post("/auth/login", json={
"email": test_user_data["email"],
"password": test_user_data["password"]
})
token = login_response.json()["access_token"]
# Get current user
headers = {"Authorization": f"Bearer {token}"}
response = client.get("/users/me", headers=headers)
assert response.status_code == 200
data = response.json()
assert data["email"] == test_user_data["email"]
assert data["full_name"] == test_user_data["full_name"]