diff --git a/app/two_factor_authentication/models/__init__.py b/app/two_factor_authentication/models/__init__.py index 83b8f2e..354e05d 100644 --- a/app/two_factor_authentication/models/__init__.py +++ b/app/two_factor_authentication/models/__init__.py @@ -1 +1 @@ -from .user_2fa import Users2FA \ No newline at end of file +from .user_2fa import Users2FA diff --git a/app/two_factor_authentication/models/user_2fa.py b/app/two_factor_authentication/models/user_2fa.py index 78b914c..0dcc59d 100644 --- a/app/two_factor_authentication/models/user_2fa.py +++ b/app/two_factor_authentication/models/user_2fa.py @@ -1,15 +1,25 @@ +from typing import TYPE_CHECKING from uuid import UUID -from app.common.models.base_class import Base from sqlalchemy import ForeignKey -from sqlalchemy.orm import ( - Mapped, - mapped_column, -) +from sqlalchemy.orm import Mapped, mapped_column, relationship + +from app.common.models.base_class import Base + + +if TYPE_CHECKING: + from app.users.models import User class Users2FA(Base): __tablename__ = "users_2fa" + + # fields secret_key: Mapped[str] user_id: Mapped[UUID] = mapped_column(ForeignKey("users.id")) active: Mapped[bool] + + # relationships + user: Mapped["User"] = relationship( + back_populates="two_factor_authentications", + ) diff --git a/app/two_factor_authentication/use_cases/__init__.py b/app/two_factor_authentication/use_cases/__init__.py index e69de29..28eac66 100644 --- a/app/two_factor_authentication/use_cases/__init__.py +++ b/app/two_factor_authentication/use_cases/__init__.py @@ -0,0 +1,2 @@ +from .create_new_user_2fa_use_case import CreateNewUser2FAUseCase +from .verify_user_2fa_use_case import VerifyUser2FAUseCase diff --git a/app/users/models/__init__.py b/app/users/models/__init__.py index a5057a0..ee4c00b 100644 --- a/app/users/models/__init__.py +++ b/app/users/models/__init__.py @@ -1 +1 @@ -from .user import User # noqa +from .user import User diff --git a/app/users/models/user.py b/app/users/models/user.py index 503186e..f71b329 100644 --- a/app/users/models/user.py +++ b/app/users/models/user.py @@ -1,14 +1,27 @@ -from sqlalchemy.orm import Mapped, mapped_column +from typing import TYPE_CHECKING + +from sqlalchemy.orm import Mapped, mapped_column, relationship from sqlalchemy.types import String from app.common.models.base_class import Base - from app.users.constants.user_constants import USER_EMAIL_MAX_LENGTH +if TYPE_CHECKING: + from app.two_factor_authentication.models.user_2fa import Users2FA + + class User(Base): __tablename__ = "users" + + # fields email: Mapped[str] = mapped_column( - String(USER_EMAIL_MAX_LENGTH), unique=True + String(USER_EMAIL_MAX_LENGTH), + unique=True, ) hashed_password: Mapped[str] + + # relationships + two_factor_authentications: Mapped[list["Users2FA"]] = relationship( + back_populates="user", + ) diff --git a/app/users/repositories/__init__.py b/app/users/repositories/__init__.py index e69de29..44767f0 100644 --- a/app/users/repositories/__init__.py +++ b/app/users/repositories/__init__.py @@ -0,0 +1 @@ +from .users_repository import UsersRepository, users_repository diff --git a/tests/auth/api/endpoints/test_auth.py b/tests/auth/api/endpoints/test_auth.py index 7428560..f5ccf05 100644 --- a/tests/auth/api/endpoints/test_auth.py +++ b/tests/auth/api/endpoints/test_auth.py @@ -1,35 +1,38 @@ -from sqlalchemy.orm import Session - from fastapi.testclient import TestClient -from tests.utils.create_user import create_user +from tests.factories import UserFactory + login_path = "api/v1/auth/login" class TestLogin: - def test_login(self, client: TestClient, session: Session): - created_user = create_user(session) + def test_login( + self, client: TestClient, user_factory: UserFactory + ) -> None: + user_password = "password" + + user = user_factory.create(password=user_password) response = client.post( login_path, json={ - "email": created_user.email, - "password": "password", + "email": user.email, + "password": user_password, }, ) assert response.status_code == 204 def test_login_incorrect_password( - self, client: TestClient, session: Session - ): - created_user = create_user(session) + self, client: TestClient, user_factory: UserFactory + ) -> None: + user = user_factory.create(password="password") response = client.post( login_path, json={ - "email": created_user.email, + "email": user.email, "password": "incorrect_password", }, ) @@ -37,9 +40,9 @@ def test_login_incorrect_password( assert response.status_code == 401 def test_login_non_existent_email( - self, client: TestClient, session: Session - ): - create_user(session) + self, client: TestClient, user_factory: UserFactory + ) -> None: + user_factory.create(email="test@user.com") response = client.post( login_path, diff --git a/tests/auth/api/endpoints/test_reset_password.py b/tests/auth/api/endpoints/test_reset_password.py index c13b4f8..a74482b 100644 --- a/tests/auth/api/endpoints/test_reset_password.py +++ b/tests/auth/api/endpoints/test_reset_password.py @@ -3,17 +3,17 @@ from app.auth.schemas.token_schema import EmailTokenPayload from app.auth.utils.security import create_access_token, verify_password -from app.users.models.user import User -from tests.utils.create_user import create_user + +from tests.factories import UserFactory class TestResetPasswordEndpoint: def test_reset_password_success( - self, client: TestClient, session: Session + self, client: TestClient, user_factory: UserFactory, session: Session ) -> None: - created_user = create_user(session) + user = user_factory.create() reset_token = create_access_token( - EmailTokenPayload(user_email=created_user.email) + EmailTokenPayload(user_email=user.email) ) new_password = "MyNewSecurePassword123!" @@ -25,8 +25,8 @@ def test_reset_password_success( assert response.status_code == 204 - user = session.query(User).filter(User.id == created_user.id).first() - assert user and verify_password(new_password, user.hashed_password) + session.refresh(user) + assert verify_password(new_password, user.hashed_password) def test_reset_password_invalid_token( self, diff --git a/tests/conftest.py b/tests/conftest.py index b94c9d4..d27d048 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,15 +1,17 @@ from typing import Generator + +from fastapi.testclient import TestClient import pytest from sqlalchemy import RootTransaction, event -from fastapi.testclient import TestClient +from sqlalchemy.orm import Session -from app.common.api.dependencies.get_session import get_session -from app.core.config import get_settings +from app.core.config import settings from app.db.session import engine, SessionLocal from app.main import app -from sqlalchemy.orm import Session -settings = get_settings() +from tests.factories import ClientFactory, UserFactory + + TEST_DATABASE_URI = settings.SQLALCHEMY_DATABASE_URI @@ -39,12 +41,15 @@ def end_savepoint(session: Session, transaction: RootTransaction) -> None: @pytest.fixture() -def client(session: Session) -> Generator: - # Use the same session as the session fixture - def override_get_session() -> Generator: - yield session +def client_factory(session: Session) -> ClientFactory: + return ClientFactory(app, session) + - app.dependency_overrides[get_session] = override_get_session +@pytest.fixture() +def client(client_factory: ClientFactory) -> TestClient: + return client_factory.create() - with TestClient(app) as client: - yield client + +@pytest.fixture() +def user_factory(session: Session) -> UserFactory: + return UserFactory(session) diff --git a/tests/factories/__init__.py b/tests/factories/__init__.py new file mode 100644 index 0000000..a7d8ede --- /dev/null +++ b/tests/factories/__init__.py @@ -0,0 +1,2 @@ +from .client_factory import ClientFactory +from .user_factory import UserFactory diff --git a/tests/factories/client_factory.py b/tests/factories/client_factory.py new file mode 100644 index 0000000..0057f4f --- /dev/null +++ b/tests/factories/client_factory.py @@ -0,0 +1,34 @@ +from typing import Generator +from fastapi import FastAPI +from fastapi.testclient import TestClient +from sqlalchemy.orm import Session + +from app.auth.schemas.token_schema import TokenPayload +from app.auth.utils import security +from app.common.api.dependencies.get_session import get_session +from app.users.models.user import User + + +class ClientFactory: + def __init__(self, app: FastAPI, session: Session): + self._app = app + self._session = session + + def _get_session() -> Generator: + yield self._session + + self._app.dependency_overrides[get_session] = _get_session + + def create(self, user_to_log_in: User | None = None, /) -> TestClient: + client = TestClient(self._app) + + if not user_to_log_in: + return client + + # NOTE: set access token cookie to get around login endpoint limiter + access_token = security.create_access_token( + TokenPayload(user_id=str(user_to_log_in.id)) + ) + client.cookies = {"access_token": access_token} + + return client diff --git a/tests/factories/user_factory.py b/tests/factories/user_factory.py new file mode 100644 index 0000000..d08a414 --- /dev/null +++ b/tests/factories/user_factory.py @@ -0,0 +1,126 @@ +from datetime import datetime +from typing import Any, Iterable, Sequence +from uuid import UUID + +from pydantic import BaseModel +from sqlalchemy.orm import Session + +from app.auth.utils import security +from app.two_factor_authentication.use_cases import CreateNewUser2FAUseCase +from app.users.models import User + + +class UserFactory: + class _UserConfig(BaseModel): + id: UUID | None = None + created_at: datetime | None = None + email: str = "test@user.com" + password: str = "password" + with_2fa: bool = False + + @property + def config(self) -> type[_UserConfig]: + return self.__class__._UserConfig + + def __init__(self, session: Session): + self.session = session + self._create_2fa_use_case = CreateNewUser2FAUseCase(session) + + def create( + self, + config: _UserConfig | None = None, + /, + **kw: Any, + ) -> User: + if config is None: + config = self.config(**kw) + + user = self._create_model(config) + self.session.add(user) + self.session.flush() + + if config.with_2fa: + self._create_2fa(user) + + return user + + def create_many( + self, + amount: int, + /, + *, + email_base: str = "test@mail.com", + configurations: Sequence[_UserConfig] | None = None, + **kw: Any, + ) -> list[User]: + assert amount > 1, "Amount must be greater than 1" + assert email_base.count("@") == 1, ( + "`email_base` must contain exactly one `@`" + ) + + configurations = self._parse_configurations( + amount, email_base, configurations, kw + ) + + users = [self._create_model(config) for config in configurations] + + self.session.add_all(users) + self.session.flush() + + self._create_2fa( + user + for user, config in zip(users, configurations, strict=True) + if config.with_2fa + ) + + return users + + def _create_model(self, config: _UserConfig, /) -> User: + user = User( + email=config.email, + hashed_password=security.get_password_hash(config.password), + ) + + if config.id: + user.id = config.id + + if config.created_at: + user.created_at = config.created_at + + return user + + def _create_2fa(self, users: User | Iterable[User], /) -> None: + if isinstance(users, User): + users = [users] + + for user in users: + self._create_2fa_use_case.execute(user.id) + self.session.refresh( + user, attribute_names=("two_factor_authentications",) + ) + assert len(user.two_factor_authentications) == 1, ( + "UserFactory failed to create user 2fa" + ) + + def _parse_configurations( + self, + amount: int, + email_base: str, + configurations: Sequence[_UserConfig] | None, + kw: dict[str, Any], + ) -> list[_UserConfig]: + address, domain = email_base.split("@") + + parsed_configurations = [] + for i in range(amount): + if configurations is None or len(configurations) <= i: + config = self.config(**kw) + else: + config = configurations[i] + + if "email" not in config.model_fields_set: + config.email = f"{address}{i}@{domain}" + + parsed_configurations.append(config) + + return parsed_configurations diff --git a/tests/two_factor_authentication/endpoints/test_create_new_user_2fa_endpoint.py b/tests/two_factor_authentication/endpoints/test_create_new_user_2fa_endpoint.py index 677406d..ab72e9e 100644 --- a/tests/two_factor_authentication/endpoints/test_create_new_user_2fa_endpoint.py +++ b/tests/two_factor_authentication/endpoints/test_create_new_user_2fa_endpoint.py @@ -1,32 +1,29 @@ +from urllib.parse import quote + from fastapi.testclient import TestClient from sqlalchemy.orm import Session -from app.core.config import get_settings +from app.core.config import settings from app.two_factor_authentication.services.users_2fa_service import ( Users2FAService, ) -from tests.utils.create_user import create_user -from urllib.parse import quote -settings = get_settings() +from tests.factories import UserFactory class TestCreateNewUser2FAEndpoint: def test_create_new_user_2fa( - self, client: TestClient, session: Session + self, client: TestClient, session: Session, user_factory: UserFactory ) -> None: - created_user = create_user(session) - - response = client.post( - f"api/v1/users/{created_user.id}/totp", - ) - user_2fa = Users2FAService(session).get_by_user_id(created_user.id) + user = user_factory.create() + response = client.post(f"api/v1/users/{user.id}/totp") assert response.status_code == 200 - assert user_2fa + user_2fa = Users2FAService(session).get_by_user_id(user.id) + assert user_2fa, "User2FA not created" assert ( response.json()["provisioning_url"] - == f"otpauth://totp/{settings.PROJECT_NAME}:{quote(created_user.email)}?secret={user_2fa.secret_key}&issuer={settings.PROJECT_NAME}" - ) + == f"otpauth://totp/{settings.PROJECT_NAME}:{quote(user.email)}?secret={user_2fa.secret_key}&issuer={settings.PROJECT_NAME}" + ), "Wrong provisioning url" diff --git a/tests/two_factor_authentication/endpoints/test_verify_user_2fa_endpoint.py b/tests/two_factor_authentication/endpoints/test_verify_user_2fa_endpoint.py index f9b9728..7487b05 100644 --- a/tests/two_factor_authentication/endpoints/test_verify_user_2fa_endpoint.py +++ b/tests/two_factor_authentication/endpoints/test_verify_user_2fa_endpoint.py @@ -1,79 +1,71 @@ +from typing import Any, Final +from uuid import UUID + +from fastapi import Response from fastapi.testclient import TestClient import pyotp from sqlalchemy.orm import Session -from app.core.config import get_settings -from app.two_factor_authentication.services.users_2fa_service import ( - Users2FAService, -) -from app.two_factor_authentication.use_cases.create_new_user_2fa_use_case import ( - CreateNewUser2FAUseCase, -) -from tests.utils.create_user import create_user +from app.two_factor_authentication.models import Users2FA -settings = get_settings() +from tests.factories import UserFactory class TestVerifyUser2FAEndpoint: + URL: Final = "api/v1/users/{user_id}/totp/verify" + def test_verify_correct_user_2fa( - self, client: TestClient, session: Session + self, + client: TestClient, + user_factory: UserFactory, ) -> None: - created_user = create_user(session) - CreateNewUser2FAUseCase(session).execute(created_user.id) - - user_2fa = Users2FAService(session).get_by_user_id(created_user.id) - - assert user_2fa + user = user_factory.create(with_2fa=True) + user_2fa = user.two_factor_authentications[0] totp = pyotp.TOTP(user_2fa.secret_key) valid_code = totp.now() - response = client.post( - f"api/v1/users/{created_user.id}/totp/verify", - json={"user_code": valid_code}, - ) - + response = self._post(client, user.id, valid_code) assert response.status_code == 200 def test_verify_correct_user_2fa_and_mark_active( - self, client: TestClient, session: Session + self, + client: TestClient, + session: Session, + user_factory: UserFactory, ) -> None: - created_user = create_user(session) - CreateNewUser2FAUseCase(session).execute(created_user.id) - - user_2fa = Users2FAService(session).get_by_user_id(created_user.id) - - assert user_2fa + user = user_factory.create(with_2fa=True) + user_2fa = user.two_factor_authentications[0] totp = pyotp.TOTP(user_2fa.secret_key) valid_code = totp.now() - response = client.post( - f"api/v1/users/{created_user.id}/totp/verify", - json={"user_code": valid_code, "mark_active": True}, - ) - + response = self._post(client, user.id, valid_code, mark_active=True) assert response.status_code == 200 - user_2fa = Users2FAService(session).get_by_user_id(created_user.id) - - assert user_2fa and user_2fa.active + assert (user_2fa := session.get(Users2FA, user_2fa.id)) + assert user_2fa.active def test_verify_incorrect_user_2fa( - self, client: TestClient, session: Session + self, + client: TestClient, + user_factory: UserFactory, ) -> None: - created_user = create_user(session) - CreateNewUser2FAUseCase(session).execute(created_user.id) - - user_2fa = Users2FAService(session).get_by_user_id(created_user.id) - - assert user_2fa - - response = client.post( - f"api/v1/users/{created_user.id}/totp/verify", - json={ - "user_code": "invalid_code", - }, - ) + user = user_factory.create(with_2fa=True) + response = self._post(client, user.id, "invalid_code") assert response.status_code == 401 + + def _post( + self, + client: TestClient, + user_id: UUID, + code: str, + mark_active: bool | None = None, + ) -> Response: + url = self.URL.format(user_id=user_id) + json: dict[str, Any] = {"user_code": code} + if mark_active is not None: + json["mark_active"] = mark_active + + return client.post(url, json=json)