🚚 Refactor and simplify backend file structure (#609)

src/backend/app/tests/.gitignore

@@ -0,0 +1 @@
+.cache

src/backend/app/tests/api/api_v1/test_celery.py

@@ -0,0 +1,18 @@
+from typing import Dict
+
+from fastapi.testclient import TestClient
+
+from app.core.config import settings
+
+
+def test_celery_worker_test(
+    client: TestClient, superuser_token_headers: Dict[str, str]
+) -> None:
+    data = {"message": "test"}
+    r = client.post(
+        f"{settings.API_V1_STR}/utils/test-celery/",
+        json=data,
+        headers=superuser_token_headers,
+    )
+    response = r.json()
+    assert response["message"] == "Word received"

src/backend/app/tests/api/api_v1/test_items.py

@@ -0,0 +1,35 @@
+from fastapi.testclient import TestClient
+from sqlalchemy.orm 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, db: Session
+) -> 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, 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"] == item.id
+    assert content["owner_id"] == item.owner_id

src/backend/app/tests/api/api_v1/test_login.py

@@ -0,0 +1,28 @@
+from typing import Dict
+
+from fastapi.testclient import TestClient
+
+from app.core.config import settings
+
+
+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_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

src/backend/app/tests/api/api_v1/test_users.py

@@ -0,0 +1,116 @@
+from typing import Dict
+
+from fastapi.testclient import TestClient
+from sqlalchemy.orm import Session
+
+from app import crud
+from app.core.config import settings
+from app.schemas.user import 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, db: Session
+) -> None:
+    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.user.get_by_email(db, email=username)
+    assert user
+    assert user.email == created_user["email"]
+
+
+def test_get_existing_user(
+    client: TestClient, superuser_token_headers: dict, db: Session
+) -> None:
+    username = random_email()
+    password = random_lower_string()
+    user_in = UserCreate(email=username, password=password)
+    user = crud.user.create(db, obj_in=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.user.get_by_email(db, email=username)
+    assert existing_user
+    assert existing_user.email == api_user["email"]
+
+
+def test_create_user_existing_username(
+    client: TestClient, superuser_token_headers: dict, db: Session
+) -> None:
+    username = random_email()
+    # username = email
+    password = random_lower_string()
+    user_in = UserCreate(email=username, password=password)
+    crud.user.create(db, obj_in=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 == 400
+
+
+def test_retrieve_users(
+    client: TestClient, superuser_token_headers: dict, db: Session
+) -> None:
+    username = random_email()
+    password = random_lower_string()
+    user_in = UserCreate(email=username, password=password)
+    crud.user.create(db, obj_in=user_in)
+
+    username2 = random_email()
+    password2 = random_lower_string()
+    user_in2 = UserCreate(email=username2, password=password2)
+    crud.user.create(db, obj_in=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

src/backend/app/tests/conftest.py

@@ -0,0 +1,35 @@
+from typing import Dict, Generator
+
+import pytest
+from fastapi.testclient import TestClient
+from sqlalchemy.orm import Session
+
+from app.core.config import settings
+from app.db.engine import engine
+from app.main import app
+from app.tests.utils.user import authentication_token_from_email
+from app.tests.utils.utils import get_superuser_token_headers
+
+
+@pytest.fixture(scope="session")
+def db() -> Generator:
+    with Session(engine) as session:
+        yield session
+
+
+@pytest.fixture(scope="module")
+def client() -> Generator:
+    with TestClient(app) as c:
+        yield c
+
+
+@pytest.fixture(scope="module")
+def superuser_token_headers(client: TestClient) -> Dict[str, str]:
+    return get_superuser_token_headers(client)
+
+
+@pytest.fixture(scope="module")
+def normal_user_token_headers(client: TestClient, db: Session) -> Dict[str, str]:
+    return authentication_token_from_email(
+        client=client, email=settings.EMAIL_TEST_USER, db=db
+    )

src/backend/app/tests/crud/test_item.py

@@ -0,0 +1,61 @@
+from sqlalchemy.orm import Session
+
+from app import crud
+from app.schemas.item import ItemCreate, ItemUpdate
+from app.tests.utils.user import create_random_user
+from app.tests.utils.utils import random_lower_string
+
+
+def test_create_item(db: Session) -> None:
+    title = random_lower_string()
+    description = random_lower_string()
+    item_in = ItemCreate(title=title, description=description)
+    user = create_random_user(db)
+    item = crud.item.create_with_owner(db=db, obj_in=item_in, owner_id=user.id)
+    assert item.title == title
+    assert item.description == description
+    assert item.owner_id == user.id
+
+
+def test_get_item(db: Session) -> None:
+    title = random_lower_string()
+    description = random_lower_string()
+    item_in = ItemCreate(title=title, description=description)
+    user = create_random_user(db)
+    item = crud.item.create_with_owner(db=db, obj_in=item_in, owner_id=user.id)
+    stored_item = crud.item.get(db=db, id=item.id)
+    assert stored_item
+    assert item.id == stored_item.id
+    assert item.title == stored_item.title
+    assert item.description == stored_item.description
+    assert item.owner_id == stored_item.owner_id
+
+
+def test_update_item(db: Session) -> None:
+    title = random_lower_string()
+    description = random_lower_string()
+    item_in = ItemCreate(title=title, description=description)
+    user = create_random_user(db)
+    item = crud.item.create_with_owner(db=db, obj_in=item_in, owner_id=user.id)
+    description2 = random_lower_string()
+    item_update = ItemUpdate(description=description2)
+    item2 = crud.item.update(db=db, db_obj=item, obj_in=item_update)
+    assert item.id == item2.id
+    assert item.title == item2.title
+    assert item2.description == description2
+    assert item.owner_id == item2.owner_id
+
+
+def test_delete_item(db: Session) -> None:
+    title = random_lower_string()
+    description = random_lower_string()
+    item_in = ItemCreate(title=title, description=description)
+    user = create_random_user(db)
+    item = crud.item.create_with_owner(db=db, obj_in=item_in, owner_id=user.id)
+    item2 = crud.item.remove(db=db, id=item.id)
+    item3 = crud.item.get(db=db, id=item.id)
+    assert item3 is None
+    assert item2.id == item.id
+    assert item2.title == title
+    assert item2.description == description
+    assert item2.owner_id == user.id

src/backend/app/tests/crud/test_user.py

@@ -0,0 +1,94 @@
+from fastapi.encoders import jsonable_encoder
+from sqlalchemy.orm import Session
+
+from app import crud
+from app.core.security import verify_password
+from app.schemas.user import UserCreate, UserUpdate
+from app.tests.utils.utils import random_email, random_lower_string
+
+
+def test_create_user(db: Session) -> None:
+    email = random_email()
+    password = random_lower_string()
+    user_in = UserCreate(email=email, password=password)
+    user = crud.user.create(db, obj_in=user_in)
+    assert user.email == email
+    assert hasattr(user, "hashed_password")
+
+
+def test_authenticate_user(db: Session) -> None:
+    email = random_email()
+    password = random_lower_string()
+    user_in = UserCreate(email=email, password=password)
+    user = crud.user.create(db, obj_in=user_in)
+    authenticated_user = crud.user.authenticate(db, email=email, password=password)
+    assert authenticated_user
+    assert user.email == authenticated_user.email
+
+
+def test_not_authenticate_user(db: Session) -> None:
+    email = random_email()
+    password = random_lower_string()
+    user = crud.user.authenticate(db, email=email, password=password)
+    assert user is None
+
+
+def test_check_if_user_is_active(db: Session) -> None:
+    email = random_email()
+    password = random_lower_string()
+    user_in = UserCreate(email=email, password=password)
+    user = crud.user.create(db, obj_in=user_in)
+    is_active = crud.user.is_active(user)
+    assert is_active is True
+
+
+def test_check_if_user_is_active_inactive(db: Session) -> None:
+    email = random_email()
+    password = random_lower_string()
+    user_in = UserCreate(email=email, password=password, disabled=True)
+    user = crud.user.create(db, obj_in=user_in)
+    is_active = crud.user.is_active(user)
+    assert is_active
+
+
+def test_check_if_user_is_superuser(db: Session) -> None:
+    email = random_email()
+    password = random_lower_string()
+    user_in = UserCreate(email=email, password=password, is_superuser=True)
+    user = crud.user.create(db, obj_in=user_in)
+    is_superuser = crud.user.is_superuser(user)
+    assert is_superuser is True
+
+
+def test_check_if_user_is_superuser_normal_user(db: Session) -> None:
+    username = random_email()
+    password = random_lower_string()
+    user_in = UserCreate(email=username, password=password)
+    user = crud.user.create(db, obj_in=user_in)
+    is_superuser = crud.user.is_superuser(user)
+    assert is_superuser is False
+
+
+def test_get_user(db: Session) -> None:
+    password = random_lower_string()
+    username = random_email()
+    user_in = UserCreate(email=username, password=password, is_superuser=True)
+    user = crud.user.create(db, obj_in=user_in)
+    user_2 = crud.user.get(db, id=user.id)
+    assert user_2
+    assert user.email == user_2.email
+    assert jsonable_encoder(user) == jsonable_encoder(user_2)
+
+
+def test_update_user(db: Session) -> None:
+    password = random_lower_string()
+    email = random_email()
+    user_in = UserCreate(email=email, password=password, is_superuser=True)
+    user = crud.user.create(db, obj_in=user_in)
+    new_password = random_lower_string()
+    user_in_update = UserUpdate(password=new_password, is_superuser=True)
+    crud.user.update(db, db_obj=user, obj_in=user_in_update)
+    user_2 = crud.user.get(db, id=user.id)
+    assert user_2
+    assert user.email == user_2.email
+    assert verify_password(new_password, user_2.hashed_password)

src/backend/app/tests/utils/item.py

@@ -0,0 +1,18 @@
+from typing import Optional
+
+from sqlalchemy.orm import Session
+
+from app import crud, models
+from app.schemas.item import ItemCreate
+from app.tests.utils.user import create_random_user
+from app.tests.utils.utils import random_lower_string
+
+
+def create_random_item(db: Session, *, owner_id: Optional[int] = None) -> models.Item:
+    if owner_id is None:
+        user = create_random_user(db)
+        owner_id = user.id
+    title = random_lower_string()
+    description = random_lower_string()
+    item_in = ItemCreate(title=title, description=description, id=id)
+    return crud.item.create_with_owner(db=db, obj_in=item_in, owner_id=owner_id)

src/backend/app/tests/utils/user.py

@@ -0,0 +1,50 @@
+from typing import Dict
+
+from fastapi.testclient import TestClient
+from sqlalchemy.orm import Session
+
+from app import crud
+from app.core.config import settings
+from app.models import User
+from app.schemas.user import UserCreate, UserUpdate
+from app.tests.utils.utils import random_email, random_lower_string
+
+
+def user_authentication_headers(
+    *, client: TestClient, email: str, password: str
+) -> Dict[str, str]:
+    data = {"username": email, "password": password}
+
+    r = client.post(f"{settings.API_V1_STR}/login/access-token", data=data)
+    response = r.json()
+    auth_token = response["access_token"]
+    headers = {"Authorization": f"Bearer {auth_token}"}
+    return headers
+
+
+def create_random_user(db: Session) -> User:
+    email = random_email()
+    password = random_lower_string()
+    user_in = UserCreate(username=email, email=email, password=password)
+    user = crud.user.create(db=db, obj_in=user_in)
+    return user
+
+
+def authentication_token_from_email(
+    *, client: TestClient, email: str, db: Session
+) -> Dict[str, str]:
+    """
+    Return a valid token for the user with given email.
+
+    If the user doesn't exist it is created first.
+    """
+    password = random_lower_string()
+    user = crud.user.get_by_email(db, email=email)
+    if not user:
+        user_in_create = UserCreate(username=email, email=email, password=password)
+        user = crud.user.create(db, obj_in=user_in_create)
+    else:
+        user_in_update = UserUpdate(password=password)
+        user = crud.user.update(db, db_obj=user, obj_in=user_in_update)
+
+    return user_authentication_headers(client=client, email=email, password=password)

src/backend/app/tests/utils/utils.py

@@ -0,0 +1,27 @@
+import random
+import string
+from typing import Dict
+
+from fastapi.testclient import TestClient
+
+from app.core.config import settings
+
+
+def random_lower_string() -> str:
+    return "".join(random.choices(string.ascii_lowercase, k=32))
+
+
+def random_email() -> str:
+    return f"{random_lower_string()}@{random_lower_string()}.com"
+
+
+def get_superuser_token_headers(client: TestClient) -> Dict[str, str]:
+    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()
+    a_token = tokens["access_token"]
+    headers = {"Authorization": f"Bearer {a_token}"}
+    return headers