diff --git a/app/api/router.py b/app/api/router.py index e875fcd..6af65cf 100644 --- a/app/api/router.py +++ b/app/api/router.py @@ -2,8 +2,10 @@ from app.api.routes.health import router as health_router from app.api.routes.tickets import router as tickets_router +from app.api.routes.users import router as users_router from app.core.config import settings api_router = APIRouter(prefix=settings.api_v1_prefix) api_router.include_router(health_router, tags=["health"]) api_router.include_router(tickets_router, tags=["tickets"]) +api_router.include_router(users_router, tags=["users"]) diff --git a/app/api/routes/users.py b/app/api/routes/users.py new file mode 100644 index 0000000..0a67bbc --- /dev/null +++ b/app/api/routes/users.py @@ -0,0 +1,80 @@ +from fastapi import APIRouter, Depends, status +from sqlalchemy.orm import Session + +from app.core.errors import AppError +from app.crud.users import create_user, get_user, get_user_by_email, get_users, update_user +from app.db.session import get_db +from app.models.user import User +from app.schemas.user import UserCreate, UserResponse, UserUpdate + +router = APIRouter() + + +def _validate_unique_email(db: Session, email: str, current_user: User | None = None) -> None: + existing_user = get_user_by_email(db, email) + + if existing_user is not None and ( + current_user is None or existing_user.id != current_user.id + ): + raise AppError("Email already exists.", status.HTTP_409_CONFLICT) + + +@router.post( + "/users", + response_model=UserResponse, + status_code=status.HTTP_201_CREATED, + summary="Create user", +) +def create_user_endpoint( + user_in: UserCreate, + db: Session = Depends(get_db), +) -> UserResponse: + _validate_unique_email(db, user_in.email) + return create_user(db, user_in) + + +@router.get( + "/users", + response_model=list[UserResponse], + summary="List users", +) +def list_users_endpoint(db: Session = Depends(get_db)) -> list[UserResponse]: + return get_users(db) + + +@router.get( + "/users/{user_id}", + response_model=UserResponse, + summary="Get user", +) +def get_user_endpoint( + user_id: int, + db: Session = Depends(get_db), +) -> UserResponse: + user = get_user(db, user_id) + + if user is None: + raise AppError("User not found.", status.HTTP_404_NOT_FOUND) + + return user + + +@router.patch( + "/users/{user_id}", + response_model=UserResponse, + summary="Update user", +) +def update_user_endpoint( + user_id: int, + user_in: UserUpdate, + db: Session = Depends(get_db), +) -> UserResponse: + user = get_user(db, user_id) + + if user is None: + raise AppError("User not found.", status.HTTP_404_NOT_FOUND) + + if user_in.email is not None: + _validate_unique_email(db, user_in.email, current_user=user) + + return update_user(db, user, user_in) diff --git a/app/crud/users.py b/app/crud/users.py new file mode 100644 index 0000000..c01dc0d --- /dev/null +++ b/app/crud/users.py @@ -0,0 +1,37 @@ +from sqlalchemy import select +from sqlalchemy.orm import Session + +from app.models.user import User +from app.schemas.user import UserCreate, UserUpdate + + +def create_user(db: Session, user_in: UserCreate) -> User: + user = User(**user_in.model_dump()) + db.add(user) + db.commit() + db.refresh(user) + return user + + +def get_user(db: Session, user_id: int) -> User | None: + return db.get(User, user_id) + + +def get_user_by_email(db: Session, email: str) -> User | None: + return db.scalar(select(User).where(User.email == email)) + + +def get_users(db: Session) -> list[User]: + return list(db.scalars(select(User)).all()) + + +def update_user(db: Session, user: User, user_in: UserUpdate) -> User: + update_data = user_in.model_dump(exclude_unset=True) + + for field, value in update_data.items(): + setattr(user, field, value) + + db.add(user) + db.commit() + db.refresh(user) + return user diff --git a/app/schemas/user.py b/app/schemas/user.py new file mode 100644 index 0000000..fca6085 --- /dev/null +++ b/app/schemas/user.py @@ -0,0 +1,29 @@ +from datetime import datetime + +from pydantic import BaseModel, ConfigDict + + +class UserBase(BaseModel): + email: str + full_name: str + role: str = "requester" + is_active: bool = True + + +class UserCreate(UserBase): + hashed_password: str + + +class UserUpdate(BaseModel): + email: str | None = None + full_name: str | None = None + hashed_password: str | None = None + role: str | None = None + is_active: bool | None = None + + +class UserResponse(UserBase): + model_config = ConfigDict(from_attributes=True) + + id: int + created_at: datetime diff --git a/tests/test_users.py b/tests/test_users.py new file mode 100644 index 0000000..9b0346d --- /dev/null +++ b/tests/test_users.py @@ -0,0 +1,152 @@ +from collections.abc import Generator + +import pytest +from fastapi.testclient import TestClient +from sqlalchemy import create_engine +from sqlalchemy.orm import Session, sessionmaker +from sqlalchemy.pool import StaticPool + +from app.db.base import Base +from app.db.session import get_db +from app.main import app +import app.models as models # noqa: F401 + + +engine = create_engine( + "sqlite://", + connect_args={"check_same_thread": False}, + poolclass=StaticPool, +) + +TestingSessionLocal = sessionmaker( + autocommit=False, + autoflush=False, + bind=engine, +) + + +@pytest.fixture() +def db_session() -> Generator[Session, None, None]: + Base.metadata.create_all(bind=engine) + db = TestingSessionLocal() + + try: + yield db + finally: + db.close() + Base.metadata.drop_all(bind=engine) + + +@pytest.fixture() +def client(db_session: Session) -> Generator[TestClient, None, None]: + def override_get_db() -> Generator[Session, None, None]: + yield db_session + + app.dependency_overrides[get_db] = override_get_db + + with TestClient(app) as test_client: + yield test_client + + app.dependency_overrides.clear() + + +def create_user_payload(email: str = "user@example.com") -> dict[str, object]: + return { + "email": email, + "full_name": "Test User", + "hashed_password": "not-a-real-password-hash", + } + + +def create_user(client: TestClient, email: str = "user@example.com") -> dict[str, object]: + response = client.post("/api/v1/users", json=create_user_payload(email=email)) + assert response.status_code == 201 + return response.json() + + +def test_create_user_returns_created_user_without_hashed_password(client: TestClient) -> None: + response = client.post("/api/v1/users", json=create_user_payload()) + + assert response.status_code == 201 + + data = response.json() + assert data["id"] == 1 + assert data["email"] == "user@example.com" + assert data["full_name"] == "Test User" + assert data["role"] == "requester" + assert data["is_active"] is True + assert "created_at" in data + assert "hashed_password" not in data + + +def test_list_users_returns_created_users(client: TestClient) -> None: + create_user(client) + + response = client.get("/api/v1/users") + + assert response.status_code == 200 + + data = response.json() + assert len(data) == 1 + assert data[0]["email"] == "user@example.com" + assert "hashed_password" not in data[0] + + +def test_get_user_returns_existing_user(client: TestClient) -> None: + user = create_user(client) + + response = client.get(f"/api/v1/users/{user['id']}") + + assert response.status_code == 200 + assert response.json()["id"] == user["id"] + assert response.json()["email"] == "user@example.com" + + +def test_get_user_returns_not_found_for_missing_user(client: TestClient) -> None: + response = client.get("/api/v1/users/999") + + assert response.status_code == 404 + assert response.json()["error"]["message"] == "User not found." + + +def test_update_user_partially_updates_existing_user(client: TestClient) -> None: + user = create_user(client) + + response = client.patch( + f"/api/v1/users/{user['id']}", + json={ + "full_name": "Updated User", + "is_active": False, + }, + ) + + assert response.status_code == 200 + + data = response.json() + assert data["id"] == user["id"] + assert data["email"] == "user@example.com" + assert data["full_name"] == "Updated User" + assert data["is_active"] is False + assert "hashed_password" not in data + + +def test_create_user_returns_conflict_for_duplicate_email(client: TestClient) -> None: + create_user(client) + + response = client.post("/api/v1/users", json=create_user_payload()) + + assert response.status_code == 409 + assert response.json()["error"]["message"] == "Email already exists." + + +def test_update_user_returns_conflict_for_duplicate_email(client: TestClient) -> None: + create_user(client, email="first@example.com") + second_user = create_user(client, email="second@example.com") + + response = client.patch( + f"/api/v1/users/{second_user['id']}", + json={"email": "first@example.com"}, + ) + + assert response.status_code == 409 + assert response.json()["error"]["message"] == "Email already exists."