first commit

backend/app/tests/api/routes/test_items.py

@@ -0,0 +1,164 @@
+import uuid
+
+from fastapi.testclient import TestClient
+from sqlmodel import Session
+
+from app.core.config import settings
+from app.tests.utils.item import create_random_item
+
+
+def test_create_item(
+    client: TestClient, superuser_token_headers: dict[str, str]
+) -> None:
+    data = {"title": "Foo", "description": "Fighters"}
+    response = client.post(
+        f"{settings.API_V1_STR}/items/",
+        headers=superuser_token_headers,
+        json=data,
+    )
+    assert response.status_code == 200
+    content = response.json()
+    assert content["title"] == data["title"]
+    assert content["description"] == data["description"]
+    assert "id" in content
+    assert "owner_id" in content
+
+
+def test_read_item(
+    client: TestClient, superuser_token_headers: dict[str, str], db: Session
+) -> None:
+    item = create_random_item(db)
+    response = client.get(
+        f"{settings.API_V1_STR}/items/{item.id}",
+        headers=superuser_token_headers,
+    )
+    assert response.status_code == 200
+    content = response.json()
+    assert content["title"] == item.title
+    assert content["description"] == item.description
+    assert content["id"] == str(item.id)
+    assert content["owner_id"] == str(item.owner_id)
+
+
+def test_read_item_not_found(
+    client: TestClient, superuser_token_headers: dict[str, str]
+) -> None:
+    response = client.get(
+        f"{settings.API_V1_STR}/items/{uuid.uuid4()}",
+        headers=superuser_token_headers,
+    )
+    assert response.status_code == 404
+    content = response.json()
+    assert content["detail"] == "Item not found"
+
+
+def test_read_item_not_enough_permissions(
+    client: TestClient, normal_user_token_headers: dict[str, str], db: Session
+) -> None:
+    item = create_random_item(db)
+    response = client.get(
+        f"{settings.API_V1_STR}/items/{item.id}",
+        headers=normal_user_token_headers,
+    )
+    assert response.status_code == 400
+    content = response.json()
+    assert content["detail"] == "Not enough permissions"
+
+
+def test_read_items(
+    client: TestClient, superuser_token_headers: dict[str, str], db: Session
+) -> None:
+    create_random_item(db)
+    create_random_item(db)
+    response = client.get(
+        f"{settings.API_V1_STR}/items/",
+        headers=superuser_token_headers,
+    )
+    assert response.status_code == 200
+    content = response.json()
+    assert len(content["data"]) >= 2
+
+
+def test_update_item(
+    client: TestClient, superuser_token_headers: dict[str, str], db: Session
+) -> None:
+    item = create_random_item(db)
+    data = {"title": "Updated title", "description": "Updated description"}
+    response = client.put(
+        f"{settings.API_V1_STR}/items/{item.id}",
+        headers=superuser_token_headers,
+        json=data,
+    )
+    assert response.status_code == 200
+    content = response.json()
+    assert content["title"] == data["title"]
+    assert content["description"] == data["description"]
+    assert content["id"] == str(item.id)
+    assert content["owner_id"] == str(item.owner_id)
+
+
+def test_update_item_not_found(
+    client: TestClient, superuser_token_headers: dict[str, str]
+) -> None:
+    data = {"title": "Updated title", "description": "Updated description"}
+    response = client.put(
+        f"{settings.API_V1_STR}/items/{uuid.uuid4()}",
+        headers=superuser_token_headers,
+        json=data,
+    )
+    assert response.status_code == 404
+    content = response.json()
+    assert content["detail"] == "Item not found"
+
+
+def test_update_item_not_enough_permissions(
+    client: TestClient, normal_user_token_headers: dict[str, str], db: Session
+) -> None:
+    item = create_random_item(db)
+    data = {"title": "Updated title", "description": "Updated description"}
+    response = client.put(
+        f"{settings.API_V1_STR}/items/{item.id}",
+        headers=normal_user_token_headers,
+        json=data,
+    )
+    assert response.status_code == 400
+    content = response.json()
+    assert content["detail"] == "Not enough permissions"
+
+
+def test_delete_item(
+    client: TestClient, superuser_token_headers: dict[str, str], db: Session
+) -> None:
+    item = create_random_item(db)
+    response = client.delete(
+        f"{settings.API_V1_STR}/items/{item.id}",
+        headers=superuser_token_headers,
+    )
+    assert response.status_code == 200
+    content = response.json()
+    assert content["message"] == "Item deleted successfully"
+
+
+def test_delete_item_not_found(
+    client: TestClient, superuser_token_headers: dict[str, str]
+) -> None:
+    response = client.delete(
+        f"{settings.API_V1_STR}/items/{uuid.uuid4()}",
+        headers=superuser_token_headers,
+    )
+    assert response.status_code == 404
+    content = response.json()
+    assert content["detail"] == "Item not found"
+
+
+def test_delete_item_not_enough_permissions(
+    client: TestClient, normal_user_token_headers: dict[str, str], db: Session
+) -> None:
+    item = create_random_item(db)
+    response = client.delete(
+        f"{settings.API_V1_STR}/items/{item.id}",
+        headers=normal_user_token_headers,
+    )
+    assert response.status_code == 400
+    content = response.json()
+    assert content["detail"] == "Not enough permissions"

backend/app/tests/api/routes/test_login.py

@@ -0,0 +1,104 @@
+from unittest.mock import patch
+
+from fastapi.testclient import TestClient
+from sqlmodel import Session, select
+
+from app.core.config import settings
+from app.core.security import verify_password
+from app.models import User
+from app.utils import generate_password_reset_token
+
+
+def test_get_access_token(client: TestClient) -> None:
+    login_data = {
+        "username": settings.FIRST_SUPERUSER,
+        "password": settings.FIRST_SUPERUSER_PASSWORD,
+    }
+    r = client.post(f"{settings.API_V1_STR}/login/access-token", data=login_data)
+    tokens = r.json()
+    assert r.status_code == 200
+    assert "access_token" in tokens
+    assert tokens["access_token"]
+
+
+def test_get_access_token_incorrect_password(client: TestClient) -> None:
+    login_data = {
+        "username": settings.FIRST_SUPERUSER,
+        "password": "incorrect",
+    }
+    r = client.post(f"{settings.API_V1_STR}/login/access-token", data=login_data)
+    assert r.status_code == 400
+
+
+def test_use_access_token(
+    client: TestClient, superuser_token_headers: dict[str, str]
+) -> None:
+    r = client.post(
+        f"{settings.API_V1_STR}/login/test-token",
+        headers=superuser_token_headers,
+    )
+    result = r.json()
+    assert r.status_code == 200
+    assert "email" in result
+
+
+def test_recovery_password(
+    client: TestClient, normal_user_token_headers: dict[str, str]
+) -> None:
+    with (
+        patch("app.core.config.settings.SMTP_HOST", "smtp.example.com"),
+        patch("app.core.config.settings.SMTP_USER", "admin@example.com"),
+    ):
+        email = "test@example.com"
+        r = client.post(
+            f"{settings.API_V1_STR}/password-recovery/{email}",
+            headers=normal_user_token_headers,
+        )
+        assert r.status_code == 200
+        assert r.json() == {"message": "Password recovery email sent"}
+
+
+def test_recovery_password_user_not_exits(
+    client: TestClient, normal_user_token_headers: dict[str, str]
+) -> None:
+    email = "jVgQr@example.com"
+    r = client.post(
+        f"{settings.API_V1_STR}/password-recovery/{email}",
+        headers=normal_user_token_headers,
+    )
+    assert r.status_code == 404
+
+
+def test_reset_password(
+    client: TestClient, superuser_token_headers: dict[str, str], db: Session
+) -> None:
+    token = generate_password_reset_token(email=settings.FIRST_SUPERUSER)
+    data = {"new_password": "changethis", "token": token}
+    r = client.post(
+        f"{settings.API_V1_STR}/reset-password/",
+        headers=superuser_token_headers,
+        json=data,
+    )
+    assert r.status_code == 200
+    assert r.json() == {"message": "Password updated successfully"}
+
+    user_query = select(User).where(User.email == settings.FIRST_SUPERUSER)
+    user = db.exec(user_query).first()
+    assert user
+    assert verify_password(data["new_password"], user.hashed_password)
+
+
+def test_reset_password_invalid_token(
+    client: TestClient, superuser_token_headers: dict[str, str]
+) -> None:
+    data = {"new_password": "changethis", "token": "invalid"}
+    r = client.post(
+        f"{settings.API_V1_STR}/reset-password/",
+        headers=superuser_token_headers,
+        json=data,
+    )
+    response = r.json()
+
+    assert "detail" in response
+    assert r.status_code == 400
+    assert response["detail"] == "Invalid token"

backend/app/tests/api/routes/test_users.py

@@ -0,0 +1,486 @@
+import uuid
+from unittest.mock import patch
+
+from fastapi.testclient import TestClient
+from sqlmodel import Session, select
+
+from app import crud
+from app.core.config import settings
+from app.core.security import verify_password
+from app.models import User, UserCreate
+from app.tests.utils.utils import random_email, random_lower_string
+
+
+def test_get_users_superuser_me(
+    client: TestClient, superuser_token_headers: dict[str, str]
+) -> None:
+    r = client.get(f"{settings.API_V1_STR}/users/me", headers=superuser_token_headers)
+    current_user = r.json()
+    assert current_user
+    assert current_user["is_active"] is True
+    assert current_user["is_superuser"]
+    assert current_user["email"] == settings.FIRST_SUPERUSER
+
+
+def test_get_users_normal_user_me(
+    client: TestClient, normal_user_token_headers: dict[str, str]
+) -> None:
+    r = client.get(f"{settings.API_V1_STR}/users/me", headers=normal_user_token_headers)
+    current_user = r.json()
+    assert current_user
+    assert current_user["is_active"] is True
+    assert current_user["is_superuser"] is False
+    assert current_user["email"] == settings.EMAIL_TEST_USER
+
+
+def test_create_user_new_email(
+    client: TestClient, superuser_token_headers: dict[str, str], db: Session
+) -> None:
+    with (
+        patch("app.utils.send_email", return_value=None),
+        patch("app.core.config.settings.SMTP_HOST", "smtp.example.com"),
+        patch("app.core.config.settings.SMTP_USER", "admin@example.com"),
+    ):
+        username = random_email()
+        password = random_lower_string()
+        data = {"email": username, "password": password}
+        r = client.post(
+            f"{settings.API_V1_STR}/users/",
+            headers=superuser_token_headers,
+            json=data,
+        )
+        assert 200 <= r.status_code < 300
+        created_user = r.json()
+        user = crud.get_user_by_email(session=db, email=username)
+        assert user
+        assert user.email == created_user["email"]
+
+
+def test_get_existing_user(
+    client: TestClient, superuser_token_headers: dict[str, str], db: Session
+) -> None:
+    username = random_email()
+    password = random_lower_string()
+    user_in = UserCreate(email=username, password=password)
+    user = crud.create_user(session=db, user_create=user_in)
+    user_id = user.id
+    r = client.get(
+        f"{settings.API_V1_STR}/users/{user_id}",
+        headers=superuser_token_headers,
+    )
+    assert 200 <= r.status_code < 300
+    api_user = r.json()
+    existing_user = crud.get_user_by_email(session=db, email=username)
+    assert existing_user
+    assert existing_user.email == api_user["email"]
+
+
+def test_get_existing_user_current_user(client: TestClient, db: Session) -> None:
+    username = random_email()
+    password = random_lower_string()
+    user_in = UserCreate(email=username, password=password)
+    user = crud.create_user(session=db, user_create=user_in)
+    user_id = user.id
+
+    login_data = {
+        "username": username,
+        "password": password,
+    }
+    r = client.post(f"{settings.API_V1_STR}/login/access-token", data=login_data)
+    tokens = r.json()
+    a_token = tokens["access_token"]
+    headers = {"Authorization": f"Bearer {a_token}"}
+
+    r = client.get(
+        f"{settings.API_V1_STR}/users/{user_id}",
+        headers=headers,
+    )
+    assert 200 <= r.status_code < 300
+    api_user = r.json()
+    existing_user = crud.get_user_by_email(session=db, email=username)
+    assert existing_user
+    assert existing_user.email == api_user["email"]
+
+
+def test_get_existing_user_permissions_error(
+    client: TestClient, normal_user_token_headers: dict[str, str]
+) -> None:
+    r = client.get(
+        f"{settings.API_V1_STR}/users/{uuid.uuid4()}",
+        headers=normal_user_token_headers,
+    )
+    assert r.status_code == 403
+    assert r.json() == {"detail": "The user doesn't have enough privileges"}
+
+
+def test_create_user_existing_username(
+    client: TestClient, superuser_token_headers: dict[str, str], db: Session
+) -> None:
+    username = random_email()
+    # username = email
+    password = random_lower_string()
+    user_in = UserCreate(email=username, password=password)
+    crud.create_user(session=db, user_create=user_in)
+    data = {"email": username, "password": password}
+    r = client.post(
+        f"{settings.API_V1_STR}/users/",
+        headers=superuser_token_headers,
+        json=data,
+    )
+    created_user = r.json()
+    assert r.status_code == 400
+    assert "_id" not in created_user
+
+
+def test_create_user_by_normal_user(
+    client: TestClient, normal_user_token_headers: dict[str, str]
+) -> None:
+    username = random_email()
+    password = random_lower_string()
+    data = {"email": username, "password": password}
+    r = client.post(
+        f"{settings.API_V1_STR}/users/",
+        headers=normal_user_token_headers,
+        json=data,
+    )
+    assert r.status_code == 403
+
+
+def test_retrieve_users(
+    client: TestClient, superuser_token_headers: dict[str, str], db: Session
+) -> None:
+    username = random_email()
+    password = random_lower_string()
+    user_in = UserCreate(email=username, password=password)
+    crud.create_user(session=db, user_create=user_in)
+
+    username2 = random_email()
+    password2 = random_lower_string()
+    user_in2 = UserCreate(email=username2, password=password2)
+    crud.create_user(session=db, user_create=user_in2)
+
+    r = client.get(f"{settings.API_V1_STR}/users/", headers=superuser_token_headers)
+    all_users = r.json()
+
+    assert len(all_users["data"]) > 1
+    assert "count" in all_users
+    for item in all_users["data"]:
+        assert "email" in item
+
+
+def test_update_user_me(
+    client: TestClient, normal_user_token_headers: dict[str, str], db: Session
+) -> None:
+    full_name = "Updated Name"
+    email = random_email()
+    data = {"full_name": full_name, "email": email}
+    r = client.patch(
+        f"{settings.API_V1_STR}/users/me",
+        headers=normal_user_token_headers,
+        json=data,
+    )
+    assert r.status_code == 200
+    updated_user = r.json()
+    assert updated_user["email"] == email
+    assert updated_user["full_name"] == full_name
+
+    user_query = select(User).where(User.email == email)
+    user_db = db.exec(user_query).first()
+    assert user_db
+    assert user_db.email == email
+    assert user_db.full_name == full_name
+
+
+def test_update_password_me(
+    client: TestClient, superuser_token_headers: dict[str, str], db: Session
+) -> None:
+    new_password = random_lower_string()
+    data = {
+        "current_password": settings.FIRST_SUPERUSER_PASSWORD,
+        "new_password": new_password,
+    }
+    r = client.patch(
+        f"{settings.API_V1_STR}/users/me/password",
+        headers=superuser_token_headers,
+        json=data,
+    )
+    assert r.status_code == 200
+    updated_user = r.json()
+    assert updated_user["message"] == "Password updated successfully"
+
+    user_query = select(User).where(User.email == settings.FIRST_SUPERUSER)
+    user_db = db.exec(user_query).first()
+    assert user_db
+    assert user_db.email == settings.FIRST_SUPERUSER
+    assert verify_password(new_password, user_db.hashed_password)
+
+    # Revert to the old password to keep consistency in test
+    old_data = {
+        "current_password": new_password,
+        "new_password": settings.FIRST_SUPERUSER_PASSWORD,
+    }
+    r = client.patch(
+        f"{settings.API_V1_STR}/users/me/password",
+        headers=superuser_token_headers,
+        json=old_data,
+    )
+    db.refresh(user_db)
+
+    assert r.status_code == 200
+    assert verify_password(settings.FIRST_SUPERUSER_PASSWORD, user_db.hashed_password)
+
+
+def test_update_password_me_incorrect_password(
+    client: TestClient, superuser_token_headers: dict[str, str]
+) -> None:
+    new_password = random_lower_string()
+    data = {"current_password": new_password, "new_password": new_password}
+    r = client.patch(
+        f"{settings.API_V1_STR}/users/me/password",
+        headers=superuser_token_headers,
+        json=data,
+    )
+    assert r.status_code == 400
+    updated_user = r.json()
+    assert updated_user["detail"] == "Incorrect password"
+
+
+def test_update_user_me_email_exists(
+    client: TestClient, normal_user_token_headers: dict[str, str], db: Session
+) -> None:
+    username = random_email()
+    password = random_lower_string()
+    user_in = UserCreate(email=username, password=password)
+    user = crud.create_user(session=db, user_create=user_in)
+
+    data = {"email": user.email}
+    r = client.patch(
+        f"{settings.API_V1_STR}/users/me",
+        headers=normal_user_token_headers,
+        json=data,
+    )
+    assert r.status_code == 409
+    assert r.json()["detail"] == "User with this email already exists"
+
+
+def test_update_password_me_same_password_error(
+    client: TestClient, superuser_token_headers: dict[str, str]
+) -> None:
+    data = {
+        "current_password": settings.FIRST_SUPERUSER_PASSWORD,
+        "new_password": settings.FIRST_SUPERUSER_PASSWORD,
+    }
+    r = client.patch(
+        f"{settings.API_V1_STR}/users/me/password",
+        headers=superuser_token_headers,
+        json=data,
+    )
+    assert r.status_code == 400
+    updated_user = r.json()
+    assert (
+        updated_user["detail"] == "New password cannot be the same as the current one"
+    )
+
+
+def test_register_user(client: TestClient, db: Session) -> None:
+    username = random_email()
+    password = random_lower_string()
+    full_name = random_lower_string()
+    data = {"email": username, "password": password, "full_name": full_name}
+    r = client.post(
+        f"{settings.API_V1_STR}/users/signup",
+        json=data,
+    )
+    assert r.status_code == 200
+    created_user = r.json()
+    assert created_user["email"] == username
+    assert created_user["full_name"] == full_name
+
+    user_query = select(User).where(User.email == username)
+    user_db = db.exec(user_query).first()
+    assert user_db
+    assert user_db.email == username
+    assert user_db.full_name == full_name
+    assert verify_password(password, user_db.hashed_password)
+
+
+def test_register_user_already_exists_error(client: TestClient) -> None:
+    password = random_lower_string()
+    full_name = random_lower_string()
+    data = {
+        "email": settings.FIRST_SUPERUSER,
+        "password": password,
+        "full_name": full_name,
+    }
+    r = client.post(
+        f"{settings.API_V1_STR}/users/signup",
+        json=data,
+    )
+    assert r.status_code == 400
+    assert r.json()["detail"] == "The user with this email already exists in the system"
+
+
+def test_update_user(
+    client: TestClient, superuser_token_headers: dict[str, str], db: Session
+) -> None:
+    username = random_email()
+    password = random_lower_string()
+    user_in = UserCreate(email=username, password=password)
+    user = crud.create_user(session=db, user_create=user_in)
+
+    data = {"full_name": "Updated_full_name"}
+    r = client.patch(
+        f"{settings.API_V1_STR}/users/{user.id}",
+        headers=superuser_token_headers,
+        json=data,
+    )
+    assert r.status_code == 200
+    updated_user = r.json()
+
+    assert updated_user["full_name"] == "Updated_full_name"
+
+    user_query = select(User).where(User.email == username)
+    user_db = db.exec(user_query).first()
+    db.refresh(user_db)
+    assert user_db
+    assert user_db.full_name == "Updated_full_name"
+
+
+def test_update_user_not_exists(
+    client: TestClient, superuser_token_headers: dict[str, str]
+) -> None:
+    data = {"full_name": "Updated_full_name"}
+    r = client.patch(
+        f"{settings.API_V1_STR}/users/{uuid.uuid4()}",
+        headers=superuser_token_headers,
+        json=data,
+    )
+    assert r.status_code == 404
+    assert r.json()["detail"] == "The user with this id does not exist in the system"
+
+
+def test_update_user_email_exists(
+    client: TestClient, superuser_token_headers: dict[str, str], db: Session
+) -> None:
+    username = random_email()
+    password = random_lower_string()
+    user_in = UserCreate(email=username, password=password)
+    user = crud.create_user(session=db, user_create=user_in)
+
+    username2 = random_email()
+    password2 = random_lower_string()
+    user_in2 = UserCreate(email=username2, password=password2)
+    user2 = crud.create_user(session=db, user_create=user_in2)
+
+    data = {"email": user2.email}
+    r = client.patch(
+        f"{settings.API_V1_STR}/users/{user.id}",
+        headers=superuser_token_headers,
+        json=data,
+    )
+    assert r.status_code == 409
+    assert r.json()["detail"] == "User with this email already exists"
+
+
+def test_delete_user_me(client: TestClient, db: Session) -> None:
+    username = random_email()
+    password = random_lower_string()
+    user_in = UserCreate(email=username, password=password)
+    user = crud.create_user(session=db, user_create=user_in)
+    user_id = user.id
+
+    login_data = {
+        "username": username,
+        "password": password,
+    }
+    r = client.post(f"{settings.API_V1_STR}/login/access-token", data=login_data)
+    tokens = r.json()
+    a_token = tokens["access_token"]
+    headers = {"Authorization": f"Bearer {a_token}"}
+
+    r = client.delete(
+        f"{settings.API_V1_STR}/users/me",
+        headers=headers,
+    )
+    assert r.status_code == 200
+    deleted_user = r.json()
+    assert deleted_user["message"] == "User deleted successfully"
+    result = db.exec(select(User).where(User.id == user_id)).first()
+    assert result is None
+
+    user_query = select(User).where(User.id == user_id)
+    user_db = db.execute(user_query).first()
+    assert user_db is None
+
+
+def test_delete_user_me_as_superuser(
+    client: TestClient, superuser_token_headers: dict[str, str]
+) -> None:
+    r = client.delete(
+        f"{settings.API_V1_STR}/users/me",
+        headers=superuser_token_headers,
+    )
+    assert r.status_code == 403
+    response = r.json()
+    assert response["detail"] == "Super users are not allowed to delete themselves"
+
+
+def test_delete_user_super_user(
+    client: TestClient, superuser_token_headers: dict[str, str], db: Session
+) -> None:
+    username = random_email()
+    password = random_lower_string()
+    user_in = UserCreate(email=username, password=password)
+    user = crud.create_user(session=db, user_create=user_in)
+    user_id = user.id
+    r = client.delete(
+        f"{settings.API_V1_STR}/users/{user_id}",
+        headers=superuser_token_headers,
+    )
+    assert r.status_code == 200
+    deleted_user = r.json()
+    assert deleted_user["message"] == "User deleted successfully"
+    result = db.exec(select(User).where(User.id == user_id)).first()
+    assert result is None
+
+
+def test_delete_user_not_found(
+    client: TestClient, superuser_token_headers: dict[str, str]
+) -> None:
+    r = client.delete(
+        f"{settings.API_V1_STR}/users/{uuid.uuid4()}",
+        headers=superuser_token_headers,
+    )
+    assert r.status_code == 404
+    assert r.json()["detail"] == "User not found"
+
+
+def test_delete_user_current_super_user_error(
+    client: TestClient, superuser_token_headers: dict[str, str], db: Session
+) -> None:
+    super_user = crud.get_user_by_email(session=db, email=settings.FIRST_SUPERUSER)
+    assert super_user
+    user_id = super_user.id
+
+    r = client.delete(
+        f"{settings.API_V1_STR}/users/{user_id}",
+        headers=superuser_token_headers,
+    )
+    assert r.status_code == 403
+    assert r.json()["detail"] == "Super users are not allowed to delete themselves"
+
+
+def test_delete_user_without_privileges(
+    client: TestClient, normal_user_token_headers: dict[str, str], db: Session
+) -> None:
+    username = random_email()
+    password = random_lower_string()
+    user_in = UserCreate(email=username, password=password)
+    user = crud.create_user(session=db, user_create=user_in)
+
+    r = client.delete(
+        f"{settings.API_V1_STR}/users/{user.id}",
+        headers=normal_user_token_headers,
+    )
+    assert r.status_code == 403
+    assert r.json()["detail"] == "The user doesn't have enough privileges"