working tests

Signed-off-by: rafsaf <rafal.safin12@gmail.com>

Author: rafsaf

alembic.ini

@@ -48,7 +48,7 @@ truncate_slug_length = 40
 # version_path_separator = ;
 # version_path_separator = space
 version_path_separator = os  # Use os.pathsep. Default configuration used for new projects.
-
+path_separator=os
 # set to 'true' to search source files recursively
 # in each "version_locations" directory
 # new in Alembic version 1.10

app/auth/tests/test_access_token.py

@@ -28,8 +28,7 @@ async def test_login_access_token_has_response_status_code(
         },
         headers={"Content-Type": "application/x-www-form-urlencoded"},
     )
-
-    assert response.status_code == status.HTTP_200_OK
+    assert response.status_code == status.HTTP_200_OK, response.text
 
 
 @pytest.mark.asyncio(loop_scope="session")
@@ -45,7 +44,7 @@ async def test_login_access_token_jwt_has_valid_token_type(
         },
         headers={"Content-Type": "application/x-www-form-urlencoded"},
     )
-
+    assert response.status_code == status.HTTP_200_OK, response.text
     token = response.json()
     assert token["token_type"] == "Bearer"
 
@@ -64,7 +63,7 @@ async def test_login_access_token_jwt_has_valid_expire_time(
         },
         headers={"Content-Type": "application/x-www-form-urlencoded"},
     )
-
+    assert response.status_code == status.HTTP_200_OK, response.text
     token = response.json()
     current_timestamp = int(time.time())
     assert (
@@ -87,6 +86,7 @@ async def test_login_access_token_returns_valid_jwt_access_token(
         },
         headers={"Content-Type": "application/x-www-form-urlencoded"},
     )
+    assert response.status_code == status.HTTP_200_OK, response.text
 
     now = int(time.time())
     token = response.json()
@@ -110,6 +110,7 @@ async def test_login_access_token_refresh_token_has_valid_expire_time(
         },
         headers={"Content-Type": "application/x-www-form-urlencoded"},
     )
+    assert response.status_code == status.HTTP_200_OK, response.text
 
     token = response.json()
     current_time = int(time.time())
@@ -133,6 +134,7 @@ async def test_login_access_token_refresh_token_exists_in_db(
         },
         headers={"Content-Type": "application/x-www-form-urlencoded"},
     )
+    assert response.status_code == status.HTTP_200_OK, response.text
 
     token = response.json()
 
@@ -156,6 +158,7 @@ async def test_login_access_token_refresh_token_in_db_has_valid_fields(
         },
         headers={"Content-Type": "application/x-www-form-urlencoded"},
     )
+    assert response.status_code == status.HTTP_200_OK, response.text
 
     token = response.json()
     result = await session.scalars(
@@ -181,7 +184,7 @@ async def test_auth_access_token_fail_for_not_existing_user_with_message(
         headers={"Content-Type": "application/x-www-form-urlencoded"},
     )
 
-    assert response.status_code == status.HTTP_400_BAD_REQUEST
+    assert response.status_code == status.HTTP_400_BAD_REQUEST, response.text
     assert response.json() == {"detail": api_messages.PASSWORD_INVALID}
 
 
@@ -199,5 +202,5 @@ async def test_auth_access_token_fail_for_invalid_password_with_message(
         headers={"Content-Type": "application/x-www-form-urlencoded"},
     )
 
-    assert response.status_code == status.HTTP_400_BAD_REQUEST
+    assert response.status_code == status.HTTP_400_BAD_REQUEST, response.text
     assert response.json() == {"detail": api_messages.PASSWORD_INVALID}

app/auth/tests/test_api_router_jwt_errors.py

@@ -1,5 +1,5 @@
 import pytest
-from fastapi import routing, status
+from fastapi import status
 from freezegun import freeze_time
 from httpx import AsyncClient
 from sqlalchemy import delete
@@ -8,62 +8,52 @@
 from app.auth import api_messages
 from app.auth.jwt import create_jwt_token
 from app.auth.models import User
-from app.auth.views import router
+from app.main import app
 
 
 @pytest.mark.asyncio(loop_scope="session")
-@pytest.mark.parametrize("api_route", router.routes)
-async def test_api_routes_raise_401_on_jwt_decode_errors(
+async def test_api_raise_401_on_jwt_decode_errors(
     client: AsyncClient,
-    api_route: routing.APIRoute,
 ) -> None:
-    for method in api_route.methods:
-        response = await client.request(
-            method=method,
-            url=api_route.path,
-            headers={"Authorization": "Bearer garbage-invalid-jwt"},
-        )
-        assert response.status_code == status.HTTP_401_UNAUTHORIZED
-        assert response.json() == {"detail": "Token invalid: Not enough segments"}
+    response = await client.get(
+        app.url_path_for("read_current_user"),
+        headers={"Authorization": "Bearer garbage-invalid-jwt"},
+    )
+
+    assert response.status_code == status.HTTP_401_UNAUTHORIZED, response.text
+    assert response.json() == {"detail": "Token invalid: Not enough segments"}
 
 
 @pytest.mark.asyncio(loop_scope="session")
-@pytest.mark.parametrize("api_route", router.routes)
-async def test_api_routes_raise_401_on_jwt_expired_token(
+async def test_api_raise_401_on_jwt_expired_token(
     client: AsyncClient,
     default_user: User,
-    api_route: routing.APIRoute,
 ) -> None:
     with freeze_time("2023-01-01"):
         jwt = create_jwt_token(default_user.user_id)
     with freeze_time("2023-02-01"):
-        for method in api_route.methods:
-            response = await client.request(
-                method=method,
-                url=api_route.path,
-                headers={"Authorization": f"Bearer {jwt.access_token}"},
-            )
-            assert response.status_code == status.HTTP_401_UNAUTHORIZED
-            assert response.json() == {"detail": "Token invalid: Signature has expired"}
+        response = await client.get(
+            app.url_path_for("read_current_user"),
+            headers={"Authorization": f"Bearer {jwt.access_token}"},
+        )
+
+        assert response.status_code == status.HTTP_401_UNAUTHORIZED, response.text
+        assert response.json() == {"detail": "Token invalid: Signature has expired"}
 
 
 @pytest.mark.asyncio(loop_scope="session")
-@pytest.mark.parametrize("api_route", router.routes)
-async def test_api_routes_raise_401_on_jwt_user_deleted(
+async def test_api_raise_401_on_jwt_user_deleted(
     client: AsyncClient,
     default_user_headers: dict[str, str],
     default_user: User,
-    api_route: routing.APIRoute,
     session: AsyncSession,
 ) -> None:
     await session.execute(delete(User).where(User.user_id == default_user.user_id))
     await session.commit()
 
-    for method in api_route.methods:
-        response = await client.request(
-            method=method,
-            url=api_route.path,
-            headers=default_user_headers,
-        )
-        assert response.status_code == status.HTTP_401_UNAUTHORIZED
-        assert response.json() == {"detail": api_messages.JWT_ERROR_USER_REMOVED}
+    response = await client.get(
+        app.url_path_for("read_current_user"),
+        headers=default_user_headers,
+    )
+    assert response.status_code == status.HTTP_401_UNAUTHORIZED, response.text
+    assert response.json() == {"detail": api_messages.JWT_ERROR_USER_REMOVED}

app/auth/tests/test_jwt.py

@@ -76,7 +76,7 @@ def test_jwt_error_with_invalid_secret_key() -> None:
     user_id = "test_user_id"
     token = jwt.create_jwt_token(user_id)
 
-    get_settings().security.jwt_secret_key = SecretStr("the secret has changed now!")
+    get_settings().security.jwt_secret_key = SecretStr("x" * 32)
 
     with pytest.raises(HTTPException) as e:
         jwt.verify_jwt_token(token=token.access_token)

app/auth/views.py

@@ -22,8 +22,8 @@
     UserResponse,
     UserUpdatePasswordRequest,
 )
-from app.core import database_session
 from app.core.config import get_settings
+from app.core.database_session import new_async_session
 
 router = APIRouter(responses=api_messages.UNAUTHORIZED_RESPONSES)
 
@@ -42,7 +42,7 @@ async def read_current_user(
 )
 async def delete_current_user(
     current_user: User = Depends(dependencies.get_current_user),
-    session: AsyncSession = Depends(database_session.new_async_session),
+    session: AsyncSession = Depends(new_async_session),
 ) -> None:
     await session.execute(delete(User).where(User.user_id == current_user.user_id))
     await session.commit()
@@ -55,7 +55,7 @@ async def delete_current_user(
 )
 async def reset_current_user_password(
     user_update_password: UserUpdatePasswordRequest,
-    session: AsyncSession = Depends(database_session.new_async_session),
+    session: AsyncSession = Depends(new_async_session),
     current_user: User = Depends(dependencies.get_current_user),
 ) -> None:
     current_user.hashed_password = get_password_hash(user_update_password.password)
@@ -70,7 +70,7 @@ async def reset_current_user_password(
     description="OAuth2 compatible token, get an access token for future requests using username and password",
 )
 async def login_access_token(
-    session: AsyncSession = Depends(database_session.new_async_session),
+    session: AsyncSession = Depends(new_async_session),
     form_data: OAuth2PasswordRequestForm = Depends(),
 ) -> AccessTokenResponse:
     user = await session.scalar(select(User).where(User.email == form_data.username))
@@ -116,7 +116,7 @@ async def login_access_token(
 )
 async def refresh_token(
     data: RefreshTokenRequest,
-    session: AsyncSession = Depends(database_session.new_async_session),
+    session: AsyncSession = Depends(new_async_session),
 ) -> AccessTokenResponse:
     token = await session.scalar(
         select(RefreshToken)
@@ -169,7 +169,7 @@ async def refresh_token(
 )
 async def register_new_user(
     new_user: UserCreateRequest,
-    session: AsyncSession = Depends(database_session.new_async_session),
+    session: AsyncSession = Depends(new_async_session),
 ) -> User:
     user = await session.scalar(select(User).where(User.email == new_user.email))
     if user is not None:

app/conftest.py

@@ -17,6 +17,7 @@
 from app.auth.models import User
 from app.core import database_session
 from app.core.config import PROJECT_DIR, get_settings
+from app.core.database_session import new_async_session
 from app.main import app as fastapi_app
 from app.tests.factories import (
     SQLAASyncPersistence,
@@ -26,7 +27,7 @@
 
 
 @pytest_asyncio.fixture(scope="session", autouse=True)
-async def fixture_setup_new_test_database() -> None:
+async def fixture_setup_new_test_database() -> AsyncGenerator[None]:
     worker_name = os.getenv("PYTEST_XDIST_WORKER", "gw0")
     test_db_name = f"test_db_{worker_name}"
 
@@ -37,6 +38,9 @@ async def fixture_setup_new_test_database() -> None:
     await conn.execute(sqlalchemy.text(f"CREATE DATABASE {test_db_name}"))
     await conn.close()
 
+    # dispose the original engine before switching to test database
+    await database_session._ASYNC_ENGINE.dispose()
+
     session_mpatch = pytest.MonkeyPatch()
     session_mpatch.setenv("DATABASE__DB", test_db_name)
     session_mpatch.setenv("SECURITY__PASSWORD_BCRYPT_ROUNDS", "4")
@@ -66,6 +70,11 @@ def alembic_upgrade() -> None:
     loop = asyncio.get_running_loop()
     await loop.run_in_executor(None, alembic_upgrade)
 
+    yield
+
+    # cleanup: dispose the test engine
+    await engine.dispose()
+
 
 @pytest_asyncio.fixture(scope="function", autouse=True)
 async def fixture_clean_get_settings_between_tests() -> AsyncGenerator[None]:
@@ -98,9 +107,7 @@ async def fixture_session_with_rollback(
         lambda: session,
     )
 
-    fastapi_app.dependency_overrides[database_session.new_async_session] = (
-        lambda: session
-    )
+    fastapi_app.dependency_overrides[new_async_session] = lambda: session
 
     persistence_handler = SQLAASyncPersistence(session=session)
     setattr(SQLAlchemySessionMixin, "__async_persistence__", persistence_handler)
@@ -109,7 +116,7 @@ async def fixture_session_with_rollback(
 
     setattr(SQLAlchemySessionMixin, "__async_persistence__", None)
 
-    fastapi_app.dependency_overrides.pop(database_session.new_async_session, None)
+    fastapi_app.dependency_overrides.pop(new_async_session, None)
 
     await session.close()
     await transaction.rollback()
@@ -131,5 +138,5 @@ async def fixture_default_user(session: AsyncSession) -> AsyncGenerator[User]:
 
 @pytest_asyncio.fixture(name="default_user_headers", scope="function")
 async def fixture_default_user_headers(default_user: User) -> dict[str, str]:
-    access_token = create_jwt_token(user_id=default_user.user_id)
+    access_token = create_jwt_token(user_id=default_user.user_id).access_token
     return {"Authorization": f"Bearer {access_token}"}

app/core/config.py

@@ -27,7 +27,9 @@
 
 class Security(BaseModel):
     jwt_issuer: str = "my-app"
-    jwt_secret_key: SecretStr = SecretStr("sk-change-me")
+    jwt_secret_key: SecretStr = SecretStr(
+        "change-me-to-a-strong-secret-key-at-least-32-chars-long"
+    )
     jwt_access_token_expire_secs: int = Field(default=15 * 60, gt=10)  # 15min
     jwt_refresh_token_expire_secs: int = Field(default=28 * 24 * 3600, gt=60)  # 28d
     jwt_algorithm: str = "HS256"

app/core/database_session.py

@@ -35,7 +35,7 @@ def new_async_engine(uri: URL) -> AsyncEngine:
 _ASYNC_SESSIONMAKER = async_sessionmaker(_ASYNC_ENGINE, expire_on_commit=False)
 
 
-async def new_async_session() -> AsyncGenerator[AsyncSession]:
+async def new_async_session() -> AsyncGenerator[AsyncSession]:  # pragma: no cover
     session = _ASYNC_SESSIONMAKER()
     try:
         yield session
@@ -44,7 +44,9 @@ async def new_async_session() -> AsyncGenerator[AsyncSession]:
 
 
 @asynccontextmanager
-async def new_script_async_session() -> AsyncGenerator[AsyncSession]:
+async def new_script_async_session() -> AsyncGenerator[
+    AsyncSession
+]:  # pragma: no cover
     _engine = create_async_engine(
         get_settings().sqlalchemy_database_uri, pool_pre_ping=True
     )

app/tests/auth.py

@@ -1,4 +1 @@
-from app.auth.password import get_password_hash
-
 TESTS_USER_PASSWORD = "geralt"
-TESTS_USER_PASSWORD_HASH = get_password_hash(TESTS_USER_PASSWORD)

app/tests/factories.py

@@ -9,7 +9,8 @@
 from polyfactory.fields import Use
 
 from app.auth.models import User
-from app.tests.auth import TESTS_USER_PASSWORD_HASH
+from app.auth.password import get_password_hash
+from app.tests.auth import TESTS_USER_PASSWORD
 
 logging.getLogger("factory").setLevel(logging.ERROR)
 
@@ -26,4 +27,4 @@ class SQLAlchemySessionMixin[T]:
 
 class UserFactory(SQLAlchemySessionMixin[User], SQLAlchemyFactory[User]):
     email = Use(Faker().email)
-    hashed_password = TESTS_USER_PASSWORD_HASH
+    hashed_password = Use(lambda: get_password_hash(TESTS_USER_PASSWORD))